Twórz narzędzia dla agenta za pomocą funkcji Unity Catalog

Używaj funkcji Unity Catalog, gdy agent musi uruchomić zdefiniowaną logikę, na przykład zapytanie SQL, którego parametry agent podaje. W przypadku innych narzędzi zacznij od MCPs.

Ten przewodnik wykorzystuje Unity Catalog AI do stworzenia małej funkcji Python, przetestowania jej i dodania do agenta za pomocą wybranego przez siebie frameworka. Jeśli już masz funkcję, zainstaluj biblioteki w kroku 1 i przejdź do kroku 3.

Requirements

• Notatnik usługi Azure Databricks z Pythonem 3.10 lub nowszym. W klasycznym środowisku obliczeniowym używaj Databricks Runtime 15.0 lub nowszego.
• Bezserwerowe przetwarzanie włączone w Twoim obszarze roboczym. Klient Function wykorzystuje ogólne obliczenia bezserwerowe do uruchamiania narzędzi; sam magazyn SQL nie wystarcza.
• USE CATALOG oraz USE SCHEMA w katalogu nadrzędnym funkcji i jej schemacie. Utworzenie funkcji wymaga CREATE FUNCTION także na schemacie; wywołanie istniejącej funkcji wymaga EXECUTE na niej. Zobacz Przywileje funkcji.
• Model wspierający wywoływanie narzędzi. Przykład LangGraph wykorzystuje model endpoint Azure Databricks. Przykłady OpenAI, Anthropic i LlamaIndex używają poświadczeń API dostawcy, jak opisano w ich zakładkach.

Krok 1: Zainstaluj biblioteki

Uruchom polecenie instalacji dla swojego frameworka w komórce notatnika:

Kod źródłowy pakietu można znaleźć w repozytorium Unity Catalog AI.

Następnie zrestartuj Python w osobnej komórce:

dbutils.library.restartPython()

Krok 2: Stwórz i przetestuj funkcję

Użyj DatabricksFunctionClient do tworzenia i testowania funkcji. Zastąp my_catalog i my_schema istniejącym katalogiem i schematem, w których możesz tworzyć funkcje. Ten przykład tworzy add_numbers; wybierz inną nazwę, jeśli już istnieje.

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

Krok 3: Użyj funkcji w swoim agencie

Ustaw function_name na pełną nazwę funkcji, a następnie uruchom przykład dla frameworka. Każdy z nich używa UCFunctionToolkit, aby wywołać funkcję bezpośrednio.

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

Ten przykład wykorzystuje punktu końcowego databricks-claude-sonnet-4-5. Zastąp go modelem wywoływania narzędzi dostępnym w Twoim obszarze roboczym.

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)

Zobacz dokumentację referencyjną integracji LangChain.

OpenAI

Ustaw zmienne środowiskowe OPENAI_API_KEY i OPENAI_MODEL dla procesu notebooka. Użyj modelu obsługującego wywoływanie narzędzi i przechowuj klucz API w sekrecie Azure Databricks.

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.")

Pomocnik uruchamia żądane funkcje i formatuje ich wyniki na potrzeby następnego wywołania modelu. Zobacz dokumentację integracji OpenAI.

Anthropic

Ustaw zmienne środowiskowe ANTHROPIC_API_KEY i ANTHROPIC_MODEL dla procesu notebooka. Użyj modelu obsługującego wywoływanie narzędzi i przechowuj klucz API w sekrecie Azure Databricks.

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.")

Pomocnik wykonuje żądane funkcje. Dodaj jego wiadomości do rozmowy przed następnym wywołaniem modelu. Zobacz odniesienie do integracji Anthropic.

LlamaIndex

Ten przykład wykorzystuje model OpenAI. Ustaw zmienne środowiskowe OPENAI_API_KEY i OPENAI_MODEL dla procesu notebooka. Użyj modelu obsługującego wywoływanie narzędzi i przechowuj klucz API w sekrecie Azure Databricks.

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)

Ostatnie dwie linie uruchom w komórce notatnika, która obsługuje await. Zobacz dokumentację integracji LlamaIndex.

Agent wywołuje add_numbers i zwraca odpowiedź zawierającą 5. Jeśli używasz istniejącej funkcji, zmień question na zadanie obsługiwane przez tę funkcję.

Aby dowiedzieć się o wdrożeniu i uwierzytelnianiu, zobacz Tworzenie niestandardowego agenta oraz uwierzytelnianie agentów.

Programowanie lokalne

Aby testować zaufane funkcje Python, zainicjalizuj klienta za pomocą execution_mode="local" i przekaż go do zestawu narzędzi w Twoim frameworku:

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

Tryb lokalny pobiera funkcję z Unity Catalog i uruchamia ją w lokalnym podprocesie. Nie obsługuje funkcji SQL. Używaj domyślnego trybu serwerless do produkcji i uruchamiaj tylko zaufany kod. Zobacz Tryby wykonania.

Aby poznać limity czasowe, limity pamięci i ustawienia bezserwerowe, zobacz lokalne limity wykonania oraz zmienne środowiskowe Function Client.

Użyj punktu końcowego MCP w przestrzeni roboczej

Dla istniejących integracji MCP ten przykład LangGraph łączy się z tą samą funkcją przez punkt końcowy workspace MCP. Włącz podgląd Zarządzanych Serwerów MCP i użyj tożsamości z dostępem do funkcji.

Uruchom %pip install --upgrade databricks-langchain langgraph "mcp>=1.24,<2" w komórce notatnika, potem zrestartuj Python. Zamień my_catalog/my_schema/add_numbers na ścieżkę swojej funkcji.

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)

Informacje o wdrażaniu znajdziesz w sekcji uwierzytelnianie agentów. W przypadku funkcji wywołującej zewnętrzne API za pomocą http_request, zobacz Legacy REST API tools.