Créer des outils d’agent à l’aide des fonctions de catalogue Unity

Utilisez les fonctions du catalogue Unity lorsque votre agent doit exécuter une logique prédéfinie, comme une requête SQL dont les paramètres sont fournis par l’agent. Pour d’autres outils, commencez par MCPs.

Ce guide utilise Unity Catalog AI pour créer une petite fonction Python, la tester, puis l’ajouter à un agent en utilisant le framework de votre choix. Si vous avez déjà une fonction, installez les bibliothèques à l’étape 1 et passez à l’étape 3.

Requirements

• Un notebook Azure Databricks avec Python 3.10 ou supérieur. En calcul classique, utilisez Databricks Runtime 15.0 ou version ultérieure.
• Calcul sans serveur activé dans votre espace de travail. Le client de fonction utilise un calcul général serverless pour exécuter des outils ; un seul entrepôt SQL ne suffit pas.
• USE CATALOG et USE SCHEMA sur le catalogue parent et le schéma de la fonction. Créer une fonction nécessite CREATE FUNCTION également sur le schéma ; appeler une fonction existante nécessite EXECUTE sur celle-ci. Voir Privilèges des fonctions.
• Un modèle qui prend en charge l’appel d’outils. L’exemple de LangGraph utilise un point de terminaison de modèle Azure Databricks. Les exemples OpenAI, Anthropic et LlamaIndex utilisent les informations d’identification d’API des fournisseurs, comme décrit dans leurs onglets.

Étape 1 : Installer les bibliothèques

Exécutez la commande d’installation de votre framework dans une cellule notebook :

Pour le code source du package, voir le dépôt Unity Catalog AI.

Ensuite, redémarrez Python dans une cellule séparée :

dbutils.library.restartPython()

Étape 2 : Créer et tester une fonction

Utilisez DatabricksFunctionClient pour créer et tester la fonction. Remplacez my_catalog et my_schema par un catalogue et un schéma existants où vous pouvez créer des fonctions. Cet exemple crée add_numbers; choisissez un autre nom si celui-ci existe déjà.

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

Étape 3 : Utilisez la fonction dans votre agent

Définissez function_name sur le nom complet de votre fonction, puis exécutez l’exemple pour votre framework. Chacun utilise UCFunctionToolkit pour appeler directement la fonction.

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

Cet exemple utilise le databricks-claude-sonnet-4-5 point de terminaison. Remplacez-le par un modèle d’appel d’outils disponible dans votre espace de travail.

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)

Voir la référence d’intégration LangChain.

OpenAI

Définissez les variables d’environnement OPENAI_API_KEY et OPENAI_MODEL pour le processus de votre notebook. Utilisez un modèle qui supporte l’appel d’outils, et stockez la clé API dans un secret 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’assistant exécute les fonctions demandées et formate leurs résultats pour l’appel suivant du modèle. Voir la référence d’intégration OpenAI.

Anthropic

Définissez les variables d’environnement ANTHROPIC_API_KEY et ANTHROPIC_MODEL pour le processus de votre notebook. Utilisez un modèle qui supporte l’appel d’outils, et stockez la clé API dans un secret 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’assistant exécute les fonctions demandées. Ajoutez ses messages à la conversation avant le prochain appel de modèle. Voir la référence sur l’intégration Anthropic.

LlamaIndex

Cet exemple utilise un modèle OpenAI. Définissez les variables d’environnement OPENAI_API_KEY et OPENAI_MODEL pour le processus de votre notebook. Utilisez un modèle qui supporte l’appel d’outils, et stockez la clé API dans un secret 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)

Exécutez les deux dernières lignes dans une cellule notebook qui supporte await. Voir la référence d’intégration LlamaIndex.

L’agent appelle add_numbers et retourne une réponse contenant 5. Si vous utilisez une fonction existante, remplacez question par une tâche que cette fonction prend en charge.

Pour le déploiement et l’authentification, voir Construire un agent personnalisé et authentification d’agent.

Développement local

Pour tester des fonctions Python de confiance, initialisez le client avec execution_mode="local" et passez-le à la boîte à outils de votre framework :

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

Le mode local récupère la fonction depuis Unity Catalog et l’exécute dans un sous-processus local. Il ne prend pas en charge les fonctions SQL. Utilisez le mode serverless par défaut pour la production, et exécutez uniquement du code fiable. Voir les modes d’exécution.

Pour les délais d’attente, les limites mémoire et les réglages serverless, voir les limites d’exécution locales et les variables d’environnement de Function Client.

Utilisez le point de terminaison MCP de l’espace de travail

Pour les intégrations MCP existantes, cet exemple LangGraph se connecte à la même fonction via le point de terminaison MCP de l’espace de travail. Activez la prévisualisation des serveurs MCP gérés et utilisez une identité avec accès à la fonction.

Exécutez %pip install --upgrade databricks-langchain langgraph "mcp>=1.24,<2" dans une cellule de notebook, puis redémarrez Python. Remplacez my_catalog/my_schema/add_numbers par le chemin de votre fonction.

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)

Pour le déploiement, voir Authentification des agents. Pour une fonction qui appelle une API externe avec http_request, voir Outils d’API REST hérités.