Creación de herramientas de agente mediante funciones de Catálogo de Unity

Utiliza funciones del Catálogo de Unity cuando tu agente necesite ejecutar lógica predefinida, como una consulta SQL cuyos parámetros suministra el agente. Para otras herramientas, empieza con MCP.

Esta guía utiliza Unity Catalog AI para crear una pequeña función en Python, probarla y añadirla a un agente usando el framework que elijas. Si ya tienes una función, instala las librerías en el paso 1 y luego pasa al paso 3.

Requirements

• Un cuaderno de Azure Databricks con Python 3.10 o superior. En cómputo clásico, utiliza Databricks Runtime 15.0 o superior.
• Computación serverless habilitada en tu espacio de trabajo. El Function Client utiliza computación general sin servidor para ejecutar herramientas; un almacén SQL por sí solo no es suficiente.
• USE CATALOG y USE SCHEMA en el catálogo y esquema padre de la función. Crear una función también requiere CREATE FUNCTION en el esquema; llamar a una función existente requiere EXECUTE en ella. Consulte privilegios de función.
• Un modelo que admite la invocación de herramientas. El ejemplo de LangGraph utiliza un punto final de modelo de Azure Databricks. Los ejemplos de OpenAI, Anthropic y LlamaIndex utilizan credenciales de API de proveedores, tal como se describe en sus pestañas.

Paso 1: Instala las bibliotecas

Ejecuta el comando de instalación de tu framework en una celda de notebook:

Para el código fuente del paquete, consulta el repositorio Unity Catalog AI.

Luego reinicia Python en una celda separada:

dbutils.library.restartPython()

Paso 2: Crear y probar una función

Usa DatabricksFunctionClient para crear y probar la función. Sustituye my_catalog y my_schema por un catálogo y esquema existentes donde puedas crear funciones. Este ejemplo crea add_numbers; elige un nombre diferente si ya existe.

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

Paso 3: Utiliza la función en tu agente

Establezca function_name con el nombre completo de su función y, a continuación, ejecute el ejemplo para su marco. Cada uno usa UCFunctionToolkit para llamar directamente a la función.

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 ejemplo utiliza el databricks-claude-sonnet-4-5 endpoint. Sustituyelo por un modelo de llamada de herramientas disponible en tu espacio de trabajo.

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)

Consulta la referencia de integración de LangChain.

OpenAI

Establezca las variables de entorno OPENAI_API_KEY y OPENAI_MODEL para el proceso del cuaderno. Utiliza un modelo que soporte la llamada a herramientas y almacena la clave API en un secreto de 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.")

El ayudante ejecuta las funciones solicitadas y formatea sus resultados para la siguiente llamada al modelo. Consulta la referencia de integración de OpenAI.

Anthropic

Establece las variables de entorno ANTHROPIC_API_KEY y ANTHROPIC_MODEL para el proceso de tu notebook. Utiliza un modelo que soporte la llamada a herramientas y almacena la clave API en un secreto de 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.")

El ayudante ejecuta las funciones solicitadas. Añade sus mensajes a la conversación antes de la siguiente llamada de modelo. Véase la referencia de integración Anthropic.

LlamaIndex

Este ejemplo utiliza un modelo OpenAI. Establezca las variables de entorno OPENAI_API_KEY y OPENAI_MODEL para el proceso del cuaderno. Utiliza un modelo que soporte la llamada a herramientas y almacena la clave API en un secreto de 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)

Ejecuta las dos últimas líneas en una celda de notebook que soporte await. Consulta la referencia de integración de LlamaIndex.

El agente llama add_numbers y devuelve una respuesta que contiene 5. Si usas una función existente, cambia question a una tarea que la función soporte.

Para la implementación y la autenticación, consulte Crear un agente personalizado y Autenticación del agente.

Desarrollo local

Para probar funciones Python confiables, inicializa el cliente con execution_mode="local" y pásalo al toolkit de tu framework:

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

El modo local recupera la función del Catálogo de Unity y la ejecuta en un subproceso local. No soporta funciones SQL. Usa el modo serverless predeterminado para producción y ejecuta solo código confiable. Véase Modos de ejecución.

Para tiempos de espera, límites de memoria y configuraciones serverless, consulta los límites de ejecución local y las variables de entorno del Cliente de Función.

Usa el endpoint MCP del espacio de trabajo (heredado)

Para integraciones MCP existentes, este ejemplo de LangGraph se conecta a la misma función a través del endpoint del espacio de trabajo heredado. Activa la vista previa de los Servidores MCP gestionados y utiliza una identidad con acceso a la función.

Ejecuta %pip install --upgrade databricks-langchain langgraph "mcp>=1.24,<2" en una celda de notebook y luego reinicia Python. Sustituye my_catalog/my_schema/add_numbers por la ruta de tu función.

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 la implementación, consulta Autenticación de agentes. Para una función que llama a una API externa con http_request, véase Herramientas heredadas de la API REST.