Crea strumenti per agenti con le funzioni di Unity Catalog

Usa le funzioni del Catalogo Unity quando il tuo agente deve eseguire una logica predefinita, come una query SQL i cui parametri sono forniti dall'agente. Per altri strumenti, inizia con gli MCP.

Questa guida utilizza Unity Catalog AI per creare una piccola funzione Python, testarla e aggiungerla a un agente usando il framework che scegli. Se hai già una funzione, installa le librerie nel passo 1 e poi passa al passo 3.

Requirements

• Un notebook Azure Databricks con Python 3.10 o superiore. Sul compute classico, usa Databricks Runtime 15.0 o superiore.
• Calcolo serverless abilitato nel tuo spazio di lavoro. Il Function Client utilizza calcolo generale serverless per eseguire gli strumenti; un SQL warehouse da solo non è sufficiente.
• USE CATALOG e USE SCHEMA sul catalogo e lo schema genitore della funzione. Creare una funzione richiede anche CREATE FUNCTION sullo schema; chiamare una funzione esistente richiede EXECUTE su di essa. Vedi Privilegi di funzione.
• Un modello che supporta l’invocazione di strumenti. L'esempio di LangGraph utilizza un endpoint modello Azure Databricks. Gli esempi di OpenAI, Anthropic e LlamaIndex utilizzano credenziali API dei provider, come descritto nelle loro schede.

Passo 1: Installa le librerie

Esegui il comando di installazione del tuo framework in una cella notebook:

Per il codice sorgente del pacchetto, consulta il repository Unity Catalog AI.

Poi riavvia Python in una cella separata:

dbutils.library.restartPython()

Passo 2: Crea e testa una funzione

Usa DatabricksFunctionClient per creare e testare la funzione. Sostituisci my_catalog e my_schema con un catalogo e uno schema esistenti dove puoi creare funzioni. Questo esempio crea add_numbers; scegli un nome diverso se esiste già.

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

Passo 3: Usa la funzione nel tuo agente

Imposta function_name sul nome completo della tua funzione, poi esegui l'esempio per il tuo framework. Ciascuno usa UCFunctionToolkit per chiamare direttamente la funzione.

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

Questo esempio utilizza l'endpoint databricks-claude-sonnet-4-5 . Sostituiscilo con un modello con chiamata di strumenti disponibile nel tuo spazio di lavoro.

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)

Vedi il riferimento all'integrazione di LangChain.

OpenAI

Imposta le variabili di ambiente OPENAI_API_KEY e OPENAI_MODEL per il processo del tuo notebook. Usa un modello che supporti la chiamata di uno strumento e memorizza la chiave API in un segreto di 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.")

L'helper esegue le funzioni richieste e formatta i loro risultati per la chiamata successiva del modello. Vedi il riferimento all'integrazione OpenAI.

Anthropic

Imposta le variabili di ambiente ANTHROPIC_API_KEY e ANTHROPIC_MODEL per il processo del tuo notebook. Usa un modello che supporti la chiamata di uno strumento e memorizza la chiave API in un segreto di 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.")

L'assistente esegue le funzioni richieste. Aggiungi i suoi messaggi alla conversazione prima della prossima chiamata del modello. Vedi il riferimento all'integrazione Anthropic.

LlamaIndex

Questo esempio utilizza un modello OpenAI. Imposta le variabili di ambiente OPENAI_API_KEY e OPENAI_MODEL per il processo del tuo notebook. Usa un modello che supporti la chiamata di uno strumento e memorizza la chiave API in un segreto di 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)

Esegui le ultime due righe in una cella notebook che supporta await. Vedi il riferimento all'integrazione di LlamaIndex.

L'agente chiama add_numbers e restituisce una risposta contenente 5. Se usi una funzione esistente, cambia question in un'attività supportata da quella funzione.

Per la distribuzione e l'autenticazione, vedi Costruire un agente personalizzato e l'autenticazione dell'agente.

Sviluppo locale

Per testare funzioni Python affidabili, inizializza il client con execution_mode="local" e passalo al toolkit del tuo framework:

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

La modalità locale recupera la funzione dal Catalogo Unity ed esegue la funzione in un sottoprocesso locale. Non supporta le funzioni SQL. Usa la modalità serverless predefinita per la produzione e esegui solo codice affidabile. Vedi modalità di esecuzione.

Per timeout, limiti di memoria e impostazioni serverless, consulta i limiti di esecuzione locali e le variabili dell'ambiente Function Client.

Usa l'endpoint MCP del workspace (legacy)

Per le integrazioni MCP esistenti, questo esempio di LangGraph si collega alla stessa funzione tramite l'endpoint legacy dello spazio di lavoro. Abilita l'anteprima dei Managed MCP Server e usa un'identità con accesso alla funzione.

Esegui %pip install --upgrade databricks-langchain langgraph "mcp>=1.24,<2" in una cella di un notebook, poi riavvia Python. Sostituisci my_catalog/my_schema/add_numbers con il percorso della tua funzione.

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)

Per la distribuzione, vedi Autenticazione degli agenti. Per una funzione che chiama un'API esterna con http_request, vedi Strumenti REST API legacy.