Criar ferramentas de agente usando funções do Catálogo do Unity

Use as funções do Catálogo do Unity quando seu agente precisar executar lógica predefinida, como uma consulta SQL cujos parâmetros o agente fornece. Para outras ferramentas, comece com MCPs.

Este guia usa o Unity Catalog AI para criar uma pequena função em Python, testá-la e adicioná-la a um agente usando o framework de sua escolha. Se você já tem uma função, instale as bibliotecas no passo 1 e vá para o passo 3.

Requirements

• Um notebook Azure Databricks com Python 3.10 ou superior. No compute clássico, use o Databricks Runtime 15.0 ou superior.
• Computação serverless ativada no seu espaço de trabalho. O Function Client utiliza computação geral sem servidor para rodar ferramentas; um warehouse SQL sozinho não é suficiente.
• USE CATALOG e USE SCHEMA no catálogo e esquema pai da função. A criação de uma função também requer CREATE FUNCTION no esquema; chamar uma função existente requer EXECUTE nela. Veja Privilégios de função.
• Um modelo que suporta chamadas de ferramentas. O exemplo do LangGraph usa um endpoint do modelo Azure Databricks. Os exemplos OpenAI, Anthropic e LlamaIndex usam credenciais da API dos provedores, conforme descrito em suas abas.

Passo 1: Instale as bibliotecas

Execute o comando de instalação do seu framework em uma célula de notebook:

Para o código-fonte do pacote, veja o repositório Unity Catalog AI.

Depois, reinicie o Python em uma célula separada:

dbutils.library.restartPython()

Passo 2: Criar e testar uma função

Use DatabricksFunctionClient para criar e testar a função. Substitua my_catalog e my_schema por um catálogo e esquema existentes onde você possa criar funções. Este exemplo cria add_numbers; escolha um nome diferente se ele já existir.

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: Use a função no seu agente

Defina function_name para o nome completo da sua função, depois execute o exemplo para o seu framework. Cada um usa UCFunctionToolkit para chamar a função diretamente.

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

Este exemplo usa o databricks-claude-sonnet-4-5 endpoint. Substitua-o por um modelo de chamada de ferramentas disponível no seu espaço de trabalho.

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)

Veja a referência de integração do LangChain.

OpenAI

Defina as variáveis de ambiente OPENAI_API_KEY e OPENAI_MODEL para o processo do notebook. Use um modelo que suporte chamada de ferramentas e armazene a chave API em um secret do 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.")

O helper executa as funções solicitadas e formata seus resultados para a próxima chamada de modelo. Veja a referência de integração OpenAI.

Anthropic

Defina as variáveis de ambiente ANTHROPIC_API_KEY e ANTHROPIC_MODEL para o processo do notebook. Use um modelo que suporte chamada de ferramentas e armazene a chave API em um secret do 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.")

O assistente executa as funções solicitadas. Anexe suas mensagens à conversa antes da próxima chamada de modelo. Veja a referência de integração Anthropic.

LlamaIndex

Este exemplo utiliza um modelo OpenAI. Defina as variáveis de ambiente OPENAI_API_KEY e OPENAI_MODEL para o processo do notebook. Use um modelo que suporte chamada de ferramentas e armazene a chave API em um 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)

Execute as duas últimas linhas em uma célula de notebook que suporte await. Veja a referência de integração do LlamaIndex.

O agente chama add_numbers e retorna uma resposta contendo 5. Se você usar uma função existente, mude question para uma tarefa que essa função suporte.

Para implantação e autenticação, veja Construir um agente personalizado e autenticação de agente.

Desenvolvimento local

Para testar funções Python confiáveis, inicialize o cliente com execution_mode="local" e passe-o para o kit de ferramentas do seu framework:

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

O modo local recupera a função do Catálogo Unity e a executa em um subprocesso local. Não suporta funções SQL. Use o modo sem servidor padrão para produção e execute apenas código confiável. Veja Modos de execução.

Para timeouts, limites de memória e configurações serverless, veja os limites locais de execução e as variáveis de ambiente do Cliente de Funções.

Use o ponto de extremidade MCP do workspace (legado)

Para integrações MCP existentes, este exemplo do LangGraph se conecta à mesma função através do endpoint legado do workspace. Ative a pré-visualização dos Servidores MCP Gerenciados e use uma identidade com acesso à função.

Execute %pip install --upgrade databricks-langchain langgraph "mcp>=1.24,<2" em uma célula de notebook e depois reinicie o Python. Substitua my_catalog/my_schema/add_numbers pelo caminho da sua função.

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)

Para implantação, veja Autenticação de agente. Para uma função que chama uma API externa com http_request, veja Ferramentas Legadas da API REST.