Utiliser les outils MCP dans un agent Python

Connectez-vous à un serveur MCP, découvrez ses outils et exécutez un agent Python qui les utilise. Utilisez l'URL d'un MCP fourni par Databricks, de votre MCP enregistré ou de votre serveur sur Databricks Apps.

Pour utiliser des MCP à partir de Claude Code, de Codex ou d'un autre agent de codage, choisissez votre client dans Agents de codage pris en charge. Pour les autres assistants et clients MCP, consultez Autres clients MCP.

Pour un projet Agent Bricks CLI, ajoutez des outils MCP avec la CLI. Les exemples ci-dessous montrent comment vous connecter à partir de votre propre code Python.

Prerequisites

• Python 3.12 sur votre ordinateur.
• L’URL MCP de votre serveur. Si quelqu’un a partagé un MCP avec vous, cette personne doit vous accorder l’accès. Effectuez la connexion au fournisseur s’il utilise l’OAuth par utilisateur. Pour un serveur sur Databricks Apps, vous devez disposer des droits 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 connaître les options d’identité et les exigences de connectivité, consultez Authentification et accès au réseau.

É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 la rubrique Authentification et mise en réseau 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 hérité, remplacez server_url par l'URL de son endpoint hérité.

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.
• Gouvernez le MCP pour restreindre les outils, appliquer des règles, définir des limites de débit et surveiller les appels.

Pour les serveurs de workspace hérités ou les serveurs hébergés sur Databricks Apps, accordez l'accès aux ressources ou à l'application sous-jacentes. Voir Agent authentication.

Ressources supplémentaires

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