Erstellen von Agenttools mithilfe von Unity-Katalogfunktionen

Verwenden Sie Unity-Catalog-Funktionen, wenn Ihr Agent vordefinierte Logik ausführen muss, wie zum Beispiel eine SQL-Abfrage, deren Parameter der Agent liefert. Für andere Werkzeuge fangen Sie mit MCPs an.

Dieser Leitfaden verwendet Unity Catalog AI, um eine kleine Python-Funktion zu erstellen, sie zu testen und mit deinem gewählten Framework zu einem Agenten hinzuzufügen. Wenn du bereits eine Funktion hast, installiere die Bibliotheken in Schritt 1 und gehe zu Schritt 3.

Requirements

• Ein Azure Databricks Notebook mit Python 3.10 oder höher. Bei klassischem Compute verwenden Sie Databricks Runtime 15.0 oder höher.
• Serverloses Computing ist in Ihrem Arbeitsbereich aktiviert. Der Function Client verwendet Serverless General Compute, um Tools auszuführen; ein SQL-Warehouse allein reicht nicht aus.
• USE CATALOG und USE SCHEMA im übergeordneten Katalog und Schema der Funktion. Das Erstellen einer Funktion erfordert außerdem CREATE FUNCTION für das Schema; das Aufrufen einer vorhandenen Funktion erfordert EXECUTE dafür. Siehe Funktionsprivilegien.
• Ein Modell, das Werkzeugaufrufe unterstützt. Das LangGraph-Beispiel verwendet einen Azure Databricks-Modellendpunkt. Die Beispiele OpenAI, Anthropic und LlamaIndex verwenden Provider-API-Zugangsdaten, wie in ihren Tabs beschrieben.

Schritt 1: Installieren Sie die Bibliotheken

Führe den Installationsbefehl für dein Framework in einer Notizbuchzelle aus:

Den Quellcode der Pakete finden Sie im Unity Catalog AI-Repository.

Dann starten Sie Python in einer separaten Zelle neu:

dbutils.library.restartPython()

Schritt 2: Erstellen und testen Sie eine Funktion

Verwenden DatabricksFunctionClient Sie es, um die Funktion zu erstellen und zu testen. Ersetzen Sie my_catalog und my_schema durch einen bestehenden Katalog und ein Schema, in dem Sie Funktionen erstellen können. Dieses Beispiel erzeugt add_numbers; wähle einen anderen Namen, falls er bereits existiert.

from unitycatalog.ai.core.databricks import DatabricksFunctionClient

client = DatabricksFunctionClient()
CATALOG = "my_catalog"
SCHEMA = "my_schema"


def add_numbers(number_1: float, number_2: float) -> float:
    """Add two numbers.

    Args:
        number_1: The first number to add.
        number_2: The second number to add.

    Returns:
        The sum of the two numbers.
    """
    return number_1 + number_2


client.create_python_function(func=add_numbers, catalog=CATALOG, schema=SCHEMA)

result = client.execute_function(
    function_name=f"{CATALOG}.{SCHEMA}.add_numbers",
    parameters={"number_1": 2.0, "number_2": 3.0},
)
if result.error:
    raise RuntimeError(result.error)
print(result.value)  # 5.0

Schritt 3: Nutzen Sie die Funktion in Ihrem Agenten

Setze function_name auf den vollständigen Namen deiner Funktion und führe dann das Beispiel für dein Framework aus. Alle verwenden UCFunctionToolkit, um die Funktion direkt aufzurufen.

from unitycatalog.ai.core.databricks import DatabricksFunctionClient

function_name = "my_catalog.my_schema.add_numbers"
client = DatabricksFunctionClient()
question = "Use the tool to add 2.0 and 3.0."

LangGraph

Dieses Beispiel verwendet den Endpunkt databricks-claude-sonnet-4-5 . Ersetzen Sie es durch ein Tool-Calling-Modell, das in Ihrem Arbeitsbereich verfügbar ist.

from databricks_langchain import ChatDatabricks, UCFunctionToolkit
from langgraph.prebuilt import create_react_agent

toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
agent = create_react_agent(
    ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
    tools=toolkit.tools,
)
result = agent.invoke({"messages": [{"role": "user", "content": question}]})
print(result["messages"][-1].content)

Siehe die LangChain-Integrationsreferenz.

OpenAI

Setze die OPENAI_API_KEY und OPENAI_MODEL Umgebungsvariablen für deinen Notebook-Prozess. Verwenden Sie ein Modell, das Tool-Aufrufe unterstützt, und speichern Sie den API-Schlüssel in einem Azure Databricks Secret.

import os
from openai import OpenAI
from unitycatalog.ai.openai.toolkit import UCFunctionToolkit
from unitycatalog.ai.openai.utils import generate_tool_call_messages

toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
model_client = OpenAI()
messages = [{"role": "user", "content": question}]

for _ in range(10):
    response = model_client.chat.completions.create(
        model=os.environ["OPENAI_MODEL"],
        messages=messages,
        tools=toolkit.tools,
    )
    if not response.choices[0].message.tool_calls:
        print(response.choices[0].message.content)
        break
    messages.extend(generate_tool_call_messages(response=response, client=client))
else:
    raise RuntimeError("The agent reached the tool-calling limit.")

Der Helfer führt die gewünschten Funktionen aus und formatiert deren Ergebnisse für den nächsten Modellaufruf. Siehe die OpenAI-Integrationsreferenz.

Anthropic

Setze die ANTHROPIC_API_KEY und ANTHROPIC_MODEL Umgebungsvariablen für deinen Notebook-Prozess. Verwenden Sie ein Modell, das Tool-Aufrufe unterstützt, und speichern Sie den API-Schlüssel in einem Azure Databricks Secret.

import os
from anthropic import Anthropic
from unitycatalog.ai.anthropic.toolkit import UCFunctionToolkit
from unitycatalog.ai.anthropic.utils import generate_tool_call_messages

toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
model_client = Anthropic()
messages = [{"role": "user", "content": question}]

for _ in range(10):
    response = model_client.messages.create(
        model=os.environ["ANTHROPIC_MODEL"],
        max_tokens=1024,
        tools=toolkit.tools,
        messages=messages,
    )
    if response.stop_reason != "tool_use":
        print("\n".join(block.text for block in response.content if block.type == "text"))
        break
    messages.extend(
        generate_tool_call_messages(
            response=response, client=client, conversation_history=[],
        )
    )
else:
    raise RuntimeError("The agent reached the tool-calling limit.")

Der Helfer führt die angeforderten Funktionen aus. Fügen Sie die Nachrichten vor dem nächsten Modellaufruf dem Gespräch an. Siehe die Referenz zur Anthropic-Integration.

LlamaIndex

Dieses Beispiel verwendet ein OpenAI-Modell. Setze die OPENAI_API_KEY und OPENAI_MODEL Umgebungsvariablen für deinen Notebook-Prozess. Verwenden Sie ein Modell, das Tool-Aufrufe unterstützt, und speichern Sie den API-Schlüssel in einem Azure Databricks Secret.

import os
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI
from unitycatalog.ai.llama_index.toolkit import UCFunctionToolkit

toolkit = UCFunctionToolkit(function_names=[function_name], client=client)
agent = FunctionAgent(
    tools=toolkit.tools,
    llm=OpenAI(model=os.environ["OPENAI_MODEL"]),
)
result = await agent.run(user_msg=question)
print(result)

Führe die letzten beiden Zeilen in einer Notebook-Zelle aus, die await unterstützt. Siehe die LlamaIndex-Integrationsreferenz.

Der Agent ruft add_numbers auf und gibt eine Antwort zurück, die 5 enthält. Wenn Sie eine vorhandene Funktion verwenden, ändern Sie question in eine Aufgabe, die von dieser Funktion unterstützt wird.

Für Bereitstellung und Authentifizierung siehe Einen benutzerdefinierten Agenten erstellen und Agentenauthentifizierung.

Lokale Entwicklung

Um vertrauenswürdige Python-Funktionen zu testen, initialisieren Sie den Client mit execution_mode="local" und übergeben Sie ihn an das Toolkit Ihres Frameworks:

client = DatabricksFunctionClient(execution_mode="local")
toolkit = UCFunctionToolkit(function_names=[function_name], client=client)

Der lokale Modus ruft die Funktion aus dem Unity-Katalog ab und führt sie in einem lokalen Teilprozess aus. Es unterstützt keine SQL-Funktionen. Verwenden Sie für die Produktion den standardmäßigen serverlosen Modus und führen Sie nur vertrauenswürdigen Code aus. Siehe Ausführungsmodi.

Für Timeouts, Speicherlimits und serverlose Einstellungen siehe die lokalen Ausführungsgrenzen und die Function Client-Umgebungsvariablen.

Verwenden Sie den Workspace MCP-Endpunkt (Legacy)

Für bestehende MCP-Integrationen verbindet sich dieses LangGraph-Beispiel mit derselben Funktion über den Legacy-Workspace-Endpunkt. Aktivieren Sie die Managed MCP Server-Vorschau und verwenden Sie eine Identität mit Zugriff auf die Funktion.

Führe %pip install --upgrade databricks-langchain langgraph "mcp>=1.24,<2" in einer Notebook-Zelle aus und starte dann Python neu. Ersetzen Sie my_catalog/my_schema/add_numbers durch den Pfad Ihrer Funktion.

from databricks.sdk import WorkspaceClient
from databricks_langchain import (
    ChatDatabricks,
    DatabricksMCPServer,
    DatabricksMultiServerMCPClient,
)
from langchain_core.messages import convert_to_openai_messages
from langgraph.prebuilt import create_react_agent

workspace_client = WorkspaceClient()
function_path = "my_catalog/my_schema/add_numbers"
mcp_client = DatabricksMultiServerMCPClient([
    DatabricksMCPServer(
        name="uc-functions",
        url=f"{workspace_client.config.host}/api/2.0/mcp/functions/{function_path}",
        workspace_client=workspace_client,
    ),
])
agent = create_react_agent(
    ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
    tools=await mcp_client.get_tools(),
    prompt=lambda state: convert_to_openai_messages(state["messages"]),
)
result = await agent.ainvoke({
    "messages": [{"role": "user", "content": "Use the tool to add 2.0 and 3.0."}],
})
print(result["messages"][-1].content)

Für die Bereitstellung siehe Agenten-Authentifizierung. Für eine Funktion, die eine externe API mit http_requestaufruft, siehe Legacy REST API Tools.