Appeler des outils MCP à partir d'un agent Python

Utilisez les outils MCP dans un agent Python personnalisé construit avec LangGraph, le SDK OpenAI Agents ou le client OpenAI. Les exemples montrent comment se connecter, découvrir des outils et exécuter un agent localement avec vos propres informations d’identification Databricks.

Pour une application qui appelle des outils au nom d’utilisateurs connectés, consultez Connectez votre application aux outils MCP.

Pour d'autres façons de vous connecter, consultez les agents de codage pris en charge, les autres clients MCP ou ajoutez des outils MCP avec la CLI Agent Bricks.

Prerequisites

• Python 3.12 sur votre ordinateur.
• L'URL d'un MCP fourni par Databricks, d'un MCP enregistré ou d'un serveur sur Databricks Apps. Si quelqu'un a partagé un MCP avec vous, cette personne doit vous accorder l'accès. Terminez la connexion au fournisseur si celle-ci utilise l'OAuth par utilisateur. Pour un serveur sur Databricks Apps, vous avez besoin de l'autorisation CAN USE sur l'application.
• Accès à un endpoint de modèle Databricks qui prend en charge l'appel d'outils. L'exemple utilise databricks-claude-sonnet-4-5. Remplacez-le par un endpoint disponible dans votre workspace.

Pour utiliser system.ai.dbsql, system.ai.sandbox ou system.ai.web_search, un administrateur de compte doit activer la version bêta de Unity Gateway depuis la page Previews de la console du compte. Consultez Gérer les aperçus du compte.

Pour les choix d’identité, voir Authentification sortante prise en charge. Pour les exigences de connectivité, voir Réseautage MCP.

Étape 1 : Installer et se connecter

1. Si vous n’avez pas installé la CLI Databricks, exécutez cette commande sur macOS ou Linux :

   Bash

   curl -fsSL https://raw.githubusercontent.com/databricks/setup-cli/main/install.sh | sh

   Pour Windows ou d’autres méthodes d’installation, consultez Installer la CLI Databricks.

2. Connectez-vous à votre workspace :

   Bash

   databricks auth login --host https://<workspace-hostname> --profile DEFAULT

3. Installez les bibliothèques Python :

   Bash

   pip install --upgrade databricks-mcp databricks-sdk "mcp>=1.24,<2"

Les exemples utilisent MCP Python 1.x, qui est compatible avec les frameworks d’agents ci-dessous.

Étape 2 : Connectez-vous à votre serveur

Enregistrez le code suivant sous mcp_agent.py. Remplacez <mcp-server-url> par l'URL de votre serveur :

• MCP fourni par Databricks ou MCP enregistré : https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<service>.
• Server on Databricks Apps : copiez l’URL de l’application depuis sa page de présentation et ajoutez /mcp à la fin.

Python

from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient

workspace_client = WorkspaceClient()
server_url = "<mcp-server-url>"
mcp_client = DatabricksMCPClient(
    server_url=server_url,
    workspace_client=workspace_client,
)

tools = mcp_client.list_tools()
for tool in tools:
    print(tool.name, tool.description, tool.inputSchema, sep="\n")

Exécutez le script :

Bash

python mcp_agent.py

Vous devriez voir les outils de votre serveur, avec leurs descriptions et leurs schémas d'entrée. Choisissez une tâche en lecture seule prise en charge par l'un de ces outils pour l'étape suivante. Si la liste est vide ou si la connexion échoue, consultez l'authentification MCP.

Utilisez un autre serveur

• For a intégré MCP, use its full name in the MCP URL, such as system.ai.github.
• Pour une intégration existante avec un serveur MCP de workspace, remplacez server_url par l'URL de l'endpoint de son workspace.

Pour trouver les MCP accessibles dans un catalogue et un schéma, exécutez :

Bash

databricks ai-gateway list-mcp-services --parent schemas/system.ai

Remplacez system.ai par votre <catalog>.<schema> pour répertorier les MCP enregistrés. La CLI gère la pagination.

Étape 3 : Exécutez un agent avec ces outils

Choisissez votre framework, installez son package et ajoutez son exemple Python à mcp_agent.py. Chaque exemple utilise les mêmes paramètres server_url et la même connexion au workspace de l'étape 2.

LangGraph

L'exemple convertit les blocs de contenu MCP au format de message de chat du modèle avec convert_to_openai_messages.

Bash

pip install --upgrade databricks-langchain langgraph

Python

import asyncio
from databricks_langchain import (
    ChatDatabricks,
    DatabricksMCPServer,
    DatabricksMultiServerMCPClient,
)
from langchain_core.messages import convert_to_openai_messages
from langgraph.prebuilt import create_react_agent

async def main():
    client = DatabricksMultiServerMCPClient([
        DatabricksMCPServer(
            name="my-mcp-server",
            url=server_url,
            workspace_client=workspace_client,
        ),
    ])
    agent = create_react_agent(
        ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
        tools=await client.get_tools(),
        prompt=lambda state: convert_to_openai_messages(state["messages"]),
    )
    task = input("Ask the agent to use a tool: ")
    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": task}],
    })
    for message in result["messages"]:
        print(message)

asyncio.run(main())

Notebook de déploiement (facultatif)

Pour le déploiement de Model Serving, adaptez ce notebook afin d’utiliser l’URL de votre serveur :

LangGraph MCP tool-calling agent

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

OpenAI Agents SDK

Bash

pip install --upgrade databricks-openai openai-agents

Python

import asyncio
from agents import Agent, Runner, set_default_openai_api, set_default_openai_client
from agents.mcp import MCPServerStreamableHttpParams
from agents.tracing import set_trace_processors
from databricks_openai import AsyncDatabricksOpenAI
from databricks_openai.agents.mcp_server import McpServer

set_default_openai_client(AsyncDatabricksOpenAI())
set_default_openai_api("chat_completions")
set_trace_processors([])

async def main():
    async with McpServer(
        name="my-mcp-server",
        params=MCPServerStreamableHttpParams(url=server_url),
        workspace_client=workspace_client,
    ) as server:
        agent = Agent(
            name="Tool-using agent",
            instructions="Use the available tools to answer the user's question.",
            model="databricks-claude-sonnet-4-5",
            mcp_servers=[server],
        )
        task = input("Ask the agent to use a tool: ")
        result = await Runner.run(agent, task)
        for item in result.new_items:
            print(item.to_input_item())

asyncio.run(main())

Notebook de déploiement (facultatif)

Pour le déploiement de Model Serving, adaptez ce notebook afin d’utiliser l’URL de votre serveur :

Agents SDK MCP tool-calling agent

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

OpenAI client

Bash

pip install --upgrade databricks-openai

Cet exemple exécute explicitement la boucle d'appel d'outils à l'aide du client compatible OpenAI.

Python

import json
from databricks_openai import DatabricksOpenAI, McpServerToolkit

toolkit = McpServerToolkit(url=server_url, workspace_client=workspace_client)
tools_by_name = {tool.name: tool for tool in toolkit.get_tools()}
model_client = DatabricksOpenAI()
messages = [{"role": "user", "content": input("Ask the agent to use a tool: ")}]

for _ in range(10):
    response = model_client.chat.completions.create(
        model="databricks-claude-sonnet-4-5",
        messages=messages,
        tools=[tool.spec for tool in tools_by_name.values()],
    )
    message = response.choices[0].message
    messages.append(message.model_dump(exclude_none=True))
    if not message.tool_calls:
        print(message.content)
        break
    for call in message.tool_calls:
        try:
            tool = tools_by_name.get(call.function.name)
            if tool is None:
                raise ValueError(f"Unknown tool: {call.function.name}")
            arguments = json.loads(call.function.arguments or "{}")
            if not isinstance(arguments, dict):
                raise ValueError("Tool arguments must be a JSON object.")
            output = tool.execute(**arguments)
        except Exception as error:
            output = json.dumps({"error": str(error)})
        print(call.function.name, output)
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": str(output),
        })
else:
    raise RuntimeError("The agent reached the tool-calling limit.")

Notebook de déploiement (facultatif)

Pour le déploiement de Model Serving, adaptez ce notebook afin d’utiliser l’URL de votre serveur :

OpenAI MCP tool-calling agent

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Exécutez de nouveau python mcp_agent.py. Lorsque vous y êtes invité, demandez la tâche en lecture seule que vous avez choisie, en incluant les entrées requises. Par exemple, si votre serveur dispose d'un outil de recherche de tickets, demandez-lui de trouver les tickets ouverts dans un projet spécifique.

Vérifiez la conversation imprimée pour y trouver l'appel d'outil, son résultat et la réponse de l'agent. Une réponse sans appel d'outil ne confirme pas que le serveur MCP a été utilisé.

Call a tool directly to troubleshoot

Utilisez le nom de l'outil et le schéma d'entrée affichés à l'étape 2. Cet exemple demande le nom et les arguments afin de fonctionner avec les outils de votre serveur. Exécutez-le après le code de connexion de l'étape 2 :

Python

import json

tool_name = input("Read-only tool name: ")
arguments = json.loads(input("Tool arguments as a JSON object: "))
result = mcp_client.call_tool(tool_name, arguments)
print(result)

Vérifiez que le résultat ne comporte pas d'erreur d'outil et qu'il contient les données attendues.

Découvrez les noms d’outils et les schémas d’entrée avec list_tools() avant d’appeler un outil. Les formats de résultat varient selon l’outil :

• Si structuredContent est présent, utilisez directement ce résultat structuré. Un outil peut décrire sa forme avec outputSchema.
• Sinon, inspectez les blocs content. Analysez un bloc de texte au format JSON uniquement si l'outil renvoie du JSON. MCP prend également en charge le texte brut et d'autres types de contenu.
• Vérifiez isError et inspectez un exemple de réponse avant de vous fier à des champs de sortie particuliers.

Déployez et partagez lorsque vous êtes prêt

L'exemple local s'exécute en tant que vous-même. Lorsque vous déployez l'agent sur Databricks Apps, choisissez l'identité qu'il utilise: le service principal de l'application pour un accès partagé, ou l'utilisateur appelant pour un accès par utilisateur.

Pour les MCP fournis ou enregistrés par Databricks :

• Accordez à l'appelant l'accès au MCP ainsi qu'à son catalogue et son schéma parent.
• Configurez l'accès par utilisateur si votre agent agit pour le compte d'un utilisateur. Si votre application connecte les utilisateurs avec votre propre fournisseur d'identité, voir Connecter votre application aux outils MCP.
• Gouvernez le MCP pour restreindre les outils, appliquer des règles, définir des limites de débit et surveiller les appels.

Pour les endpoints MCP de Workspace ou les serveurs hébergés sur Databricks Apps, accordez l'accès aux ressources ou à l'application sous-jacentes. Consultez Authentification de l'agent.

Ressources supplémentaires

• Déployer un agent sur Model Serving.
• Configurer l’authentification pour les agents Model Serving.