Utiliser les serveurs MCP dans les agents

info

Aperçu

Cette fonctionnalité est en aperçu public.

Connectez votre code d’agent à n’importe quel serveur MCP sur Databricks — serveurs gérés par Databricks, serveurs MCP externes enregistrés en tant que services MCP et serveurs personnalisés hébergés en tant qu’applications Databricks. Toutes exposent la même interface MCP, l'agent code est donc le même. Ce qui diffère, ce sont l'**URL du serveur** et la manière dont vous vous **authentifiez**.

La bibliothèque Python databricks-mcp gère l'authentification auprès des serveurs Databricks MCP, de sorte que le même code client fonctionne sur les trois types de serveurs.

Obtenez l'URL de votre serveur

Configurez d'abord le serveur MCP, puis utilisez son URL dans les exemples suivants :

Type de serveur	Modèle d'URL	Installer
Géré	`https://<workspace-hostname>/api/2.0/mcp/<service>/<path>`	Serveurs gérés disponibles
Externe (Service MCP)	`https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<mcp-service>`	Connectez les agents à des outils tiers avec les services MCP.
Personnalisé	`https://<app-url>/mcp`	Héberger votre propre serveur MCP

Type de serveur	Modèle d'URL	Installer
Géré	`https://<workspace-hostname>/api/2.0/mcp/<service>/<path>`	Serveurs gérés disponibles
Externe (Service MCP)	`https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<mcp-service>`	Connectez les agents à des outils tiers avec les services MCP.
Personnalisé	`https://<app-url>/mcp`	Héberger votre propre serveur MCP

Configurez votre environnement

Utilisez OAuth pour vous authentifier auprès de votre Workspace :
Bash
```
databricks auth login --host https://<workspace-hostname>
```
Lorsque vous y êtes invité, saisissez un nom de profil et notez-le pour plus tard. Le nom de profil default est DEFAULT.
Vérifiez que vous disposez d’un environnement local avec Python 3.12 ou version ultérieure, puis installez les dépendances :
Bash
```
pip install -U "mcp>=1.9" "databricks-sdk[openai]" "mlflow>=3.1.0" "databricks-agents>=1.0.0" "databricks-mcp"
```

Connecter et répertorier des outils

Créez un DatabricksMCPClient avec l'URL du serveur et listez ses outils. Le même client fonctionne pour les URL de serveurs gérés, externes (service MCP) et personnalisés :

Python
from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient

workspace_client = WorkspaceClient(profile="DEFAULT")
host = workspace_client.config.host

# Use a managed, MCP Service, or custom server URL:
mcp_server_url = f"{host}/api/2.0/mcp/functions/system/ai"

mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)
tools = mcp_client.list_tools()
print(f"Available tools: {[t.name for t in tools]}")

Pour appeler un outil directement :

Python
result = mcp_client.call_tool("system__ai__python_exec", {"code": "print('Hello, world!')"})
print(result.content)

remarque

Le compute Serverless doit être activé dans votre Workspace pour exécuter des outils gérés system.ai.

S'authentifier

Sélectionnez la méthode d'authentification qui correspond à l'endroit où votre agent s'exécute. Pour un Service MCP externe, l'appelant doit également disposer de EXECUTE sur le service. AI Gateway applique cette autorisation à chaque appel.

Local environment
Service principal
On-behalf-of-user

Authentifiez-vous à votre workspace avec OAuth (voir Configurer votre environnement) et transmettez le profil au client :

Python
workspace_client = WorkspaceClient(profile="DEFAULT")
mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)

Utilisez les informations d'identification OAuth d'un Service Principal Databricks. Transmettez les valeurs directement, ou récupérez-les à partir des secrets Databricks (par exemple, client_id=dbutils.secrets.get(scope="my-scope", key="client-id")) :

Python
workspace_client = WorkspaceClient(
    host="https://<workspace-hostname>",
    client_id="<client-id>",
    client_secret="<client-secret>",
)
mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)

Lorsque vous journalisez l'agent, utilisez DatabricksApps (personnalisé) ou la ressource pertinente comme ressource. Consultez la transmission automatique de l'authentification.

Utilisez ModelServingUserCredentials afin que l'agent agisse avec les autorisations de l'utilisateur appelant. Voir l'authentification au nom de l'utilisateur:

Python
from databricks.sdk.credentials_provider import ModelServingUserCredentials

workspace_client = WorkspaceClient(credentials_strategy=ModelServingUserCredentials())
mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)

Journalisez le modèle d'agent en utilisant le champ d'application apps, et pour les serveurs gérés, incluez le champ d'application OAuth correspondant pour chaque serveur. Voir Serveurs gérés disponibles.

Créer un agent

Utilisez un framework d'agent pour transformer les outils du serveur MCP en un agent. Orientez le framework vers l'URL du serveur et passez votre WorkspaceClient authentifié.

OpenAI Agents SDK
LangGraph
MCP Python SDK

Python
import asyncio
from agents import Agent, Runner
from databricks.sdk import WorkspaceClient
from databricks_openai.agents import McpServer


async def main():
    workspace_client = WorkspaceClient()
    host = workspace_client.config.host

    async with McpServer(
        url=f"{host}/ai-gateway/mcp-services/main.default.github_mcp",
        name="github-mcp",
        workspace_client=workspace_client,
    ) as mcp_server:
        agent = Agent(
            name="Local agent",
            instructions="You are a helpful assistant with access to external services.",
            model="databricks-claude-sonnet-4-5",
            mcp_servers=[mcp_server],
        )
        result = await Runner.run(agent, "List my open GitHub pull requests.")
        print(result.final_output)


asyncio.run(main())

Python
from databricks.sdk import WorkspaceClient
from databricks_langchain import ChatDatabricks, DatabricksMCPServer, DatabricksMultiServerMCPClient
from langgraph.prebuilt import create_react_agent

workspace_client = WorkspaceClient()
host = workspace_client.config.host

mcp_client = DatabricksMultiServerMCPClient([
    DatabricksMCPServer(
        name="external-service",
        url=f"{host}/ai-gateway/mcp-services/main.default.github_mcp",
        workspace_client=workspace_client,
    ),
])

async with mcp_client:
    tools = await mcp_client.get_tools()
    agent = create_react_agent(
        ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
        tools=tools,
    )
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "List my open GitHub pull requests."}]}
    )
    print(result["messages"][-1].content)

Créez un agent indépendant du framework qui découvre et appelle des outils sur un ou plusieurs serveurs MCP. Enregistrez les éléments suivants sous mcp_agent.py. Il accepte une liste d'URL de serveurs gérés, de services MCP et personnalisés :

Python
import json
import uuid
import asyncio
from typing import Any, Callable, List
from pydantic import BaseModel

import mlflow
from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import ResponsesAgentRequest, ResponsesAgentResponse

from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient
from databricks_openai import DatabricksOpenAI

# 1) CONFIGURE YOUR ENDPOINTS/PROFILE
LLM_ENDPOINT_NAME = "databricks-claude-sonnet-4-5"
SYSTEM_PROMPT = "You are a helpful assistant."
DATABRICKS_CLI_PROFILE = "YOUR_DATABRICKS_CLI_PROFILE"
assert (
    DATABRICKS_CLI_PROFILE != "YOUR_DATABRICKS_CLI_PROFILE"
), "Set DATABRICKS_CLI_PROFILE to the Databricks CLI profile name you specified when configuring authentication to the workspace"
workspace_client = WorkspaceClient(profile=DATABRICKS_CLI_PROFILE)
host = workspace_client.config.host
# Add more server URLs here — managed, MCP Service, or custom:
MANAGED_MCP_SERVER_URLS = [
    f"{host}/api/2.0/mcp/functions/system/ai",
]
# Custom MCP servers hosted on Databricks apps, or MCP Service endpoints:
CUSTOM_MCP_SERVER_URLS = []


# 2) HELPER: convert between ResponsesAgent "message dict" and ChatCompletions format
def _to_chat_messages(msg: dict[str, Any]) -> List[dict]:
    msg_type = msg.get("type")
    if msg_type == "function_call":
        return [
            {
                "role": "assistant",
                "content": None,
                "tool_calls": [
                    {
                        "id": msg["call_id"],
                        "type": "function",
                        "function": {
                            "name": msg["name"],
                            "arguments": msg["arguments"],
                        },
                    }
                ],
            }
        ]
    elif msg_type == "message" and isinstance(msg["content"], list):
        return [
            {
                "role": "assistant" if msg["role"] == "assistant" else msg["role"],
                "content": content["text"],
            }
            for content in msg["content"]
        ]
    elif msg_type == "function_call_output":
        return [
            {
                "role": "tool",
                "content": msg["output"],
                "tool_call_id": msg["tool_call_id"],
            }
        ]
    else:
        return [
            {
                k: v
                for k, v in msg.items()
                if k in ("role", "content", "name", "tool_calls", "tool_call_id")
            }
        ]


# 3) MCP SESSION + TOOL-INVOCATION LOGIC
def _make_exec_fn(server_url: str, tool_name: str, ws: WorkspaceClient) -> Callable[..., str]:
    def exec_fn(**kwargs):
        mcp_client = DatabricksMCPClient(server_url=server_url, workspace_client=ws)
        response = mcp_client.call_tool(tool_name, kwargs)
        return "".join([c.text for c in response.content])

    return exec_fn


class ToolInfo(BaseModel):
    name: str
    spec: dict
    exec_fn: Callable


def _fetch_tool_infos(ws: WorkspaceClient, server_url: str) -> List[ToolInfo]:
    print(f"Listing tools from MCP server {server_url}")
    infos: List[ToolInfo] = []
    mcp_client = DatabricksMCPClient(server_url=server_url, workspace_client=ws)
    mcp_tools = mcp_client.list_tools()
    for t in mcp_tools:
        schema = t.inputSchema.copy()
        if "properties" not in schema:
            schema["properties"] = {}
        spec = {
            "type": "function",
            "function": {
                "name": t.name,
                "description": t.description,
                "parameters": schema,
            },
        }
        infos.append(
            ToolInfo(name=t.name, spec=spec, exec_fn=_make_exec_fn(server_url, t.name, ws))
        )
    return infos


# 4) SINGLE-TURN AGENT CLASS
class SingleTurnMCPAgent(ResponsesAgent):
    def _call_llm(self, history: List[dict], ws: WorkspaceClient, tool_infos):
        client = DatabricksOpenAI()
        flat_msgs = []
        for msg in history:
            flat_msgs.extend(_to_chat_messages(msg))
        return client.chat.completions.create(
            model=LLM_ENDPOINT_NAME,
            messages=flat_msgs,
            tools=[ti.spec for ti in tool_infos],
        )

    def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
        ws = WorkspaceClient(profile=DATABRICKS_CLI_PROFILE)

        history: List[dict] = [{"role": "system", "content": SYSTEM_PROMPT}]
        for inp in request.input:
            history.append(inp.model_dump())

        tool_infos = [
            tool_info
            for mcp_server_url in (MANAGED_MCP_SERVER_URLS + CUSTOM_MCP_SERVER_URLS)
            for tool_info in _fetch_tool_infos(ws, mcp_server_url)
        ]
        tools_dict = {tool_info.name: tool_info for tool_info in tool_infos}
        llm_resp = self._call_llm(history, ws, tool_infos)
        raw_choice = llm_resp.choices[0].message.to_dict()
        raw_choice["id"] = uuid.uuid4().hex
        history.append(raw_choice)

        tool_calls = raw_choice.get("tool_calls") or []
        if tool_calls:
            fc = tool_calls[0]
            name = fc["function"]["name"]
            args = json.loads(fc["function"]["arguments"])
            try:
                tool_info = tools_dict[name]
                result = tool_info.exec_fn(**args)
            except Exception as e:
                result = f"Error invoking {name}: {e}"

            history.append(
                {
                    "type": "function_call_output",
                    "role": "tool",
                    "id": uuid.uuid4().hex,
                    "tool_call_id": fc["id"],
                    "output": result,
                }
            )

            followup = self._call_llm(history, ws, tool_infos=[]).choices[0].message.to_dict()
            followup["id"] = uuid.uuid4().hex
            assistant_text = followup.get("content", "")
            return ResponsesAgentResponse(
                output=[
                    {
                        "id": uuid.uuid4().hex,
                        "type": "message",
                        "role": "assistant",
                        "content": [{"type": "output_text", "text": assistant_text}],
                    }
                ],
                custom_outputs=request.custom_inputs,
            )

        assistant_text = raw_choice.get("content", "")
        return ResponsesAgentResponse(
            output=[
                {
                    "id": uuid.uuid4().hex,
                    "type": "message",
                    "role": "assistant",
                    "content": [{"type": "output_text", "text": assistant_text}],
                }
            ],
            custom_outputs=request.custom_inputs,
        )


mlflow.models.set_model(SingleTurnMCPAgent())

if __name__ == "__main__":
    req = ResponsesAgentRequest(
        input=[{"role": "user", "content": "What's the 100th Fibonacci number?"}]
    )
    resp = SingleTurnMCPAgent().predict(req)
    for item in resp.output:
        print(item)

Exemples de Notebooks

Les Notebooks suivants montrent comment créer des agents LangGraph et OpenAI qui appellent des outils MCP sur des serveurs MCP gérés, externes et personnalisés :

Déployez votre agent

Databricks recommande de déployer des agents sur Databricks Apps, ce qui vous permet de gérer entièrement le code de l'agent, la configuration du serveur et le versionnement basé sur Git. Vous pouvez aussi déployer sur Model Serving.

Quelle que soit votre sélection, accordez à l'agent l'accès à toutes les ressources dont dépendent ses serveurs MCP — par exemple, CAN_RUN sur un Genie Agent ou SELECT sur un index de recherche IA.

Databricks Apps (recommended)
Model Serving

Déclarez chaque ressource utilisée par votre agent — y compris les ressources derrière chaque serveur MCP — sous resources.apps.<app>.resources dans databricks.yml, puis déployez le bundle pour accorder l'accès au service principal Databricks de l'application. Par exemple, pour un agent qui utilise les serveurs gérés Genie et AI Search :

YAML
resources:
  apps:
    my_agent_app:
      name: 'my-agent-app'
      source_code_path: ./
      resources:
        - name: 'llm'
          serving_endpoint:
            name: 'databricks-claude-sonnet-4-5'
            permission: 'CAN_QUERY'
        - name: 'genie_space'
          genie_space:
            space_id: '<genie-space-id>'
            permission: 'CAN_RUN'
        - name: 'vector_index'
          uc_securable:
            securable_full_name: '<catalog>.<schema>.<index-name>'
            securable_type: 'TABLE'
            permission: 'SELECT'

Bash
databricks bundle deploy
databricks bundle run my_agent_app

Pour le workflow complet de création et de déploiement, consultez Créer un agent IA et le déployer sur Databricks Apps. Pour tous les types de Ressources et les valeurs d’autorisation, consultez Authentification pour les agents IA.

Enregistrez l'agent avec toutes les ressources dont il a besoin au moment de l'enregistrement, puis déployez. Consultez Déployer un agent pour les applications d'IA générative (Model Serving) et Authentification pour les ressources Databricks. Databricks recommande le package databricks-mcp pour dériver les ressources du serveur MCP :

Pour les serveurs MCP gérés, utilisez databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) pour récupérer les ressources dont le serveur a besoin.
Pour un serveur MCP personnalisé hébergé sur une application Databricks, incluez l’application en tant que ressource lorsque vous enregistrez le modèle.

Par exemple, pour déployer l'agent défini dans mcp_agent.py:

Python
import os
from databricks.sdk import WorkspaceClient
from databricks import agents
import mlflow
from mlflow.models.resources import DatabricksFunction, DatabricksServingEndpoint, DatabricksVectorSearchIndex
from mcp_agent import LLM_ENDPOINT_NAME
from databricks_mcp import DatabricksMCPClient

databricks_cli_profile = "YOUR_DATABRICKS_CLI_PROFILE"
assert (
    databricks_cli_profile != "YOUR_DATABRICKS_CLI_PROFILE"
), "Set databricks_cli_profile to the Databricks CLI profile name you specified when configuring authentication to the workspace"
workspace_client = WorkspaceClient(profile=databricks_cli_profile)
host = workspace_client.config.host

current_user = workspace_client.current_user.me().user_name
mlflow.set_tracking_uri(f"databricks://{databricks_cli_profile}")
mlflow.set_registry_uri(f"databricks-uc://{databricks_cli_profile}")
mlflow.set_experiment(f"/Users/{current_user}/databricks_docs_example_mcp_agent")
os.environ["DATABRICKS_CONFIG_PROFILE"] = databricks_cli_profile

MANAGED_MCP_SERVER_URLS = [
    f"{host}/api/2.0/mcp/functions/system/ai",
]

here = os.path.dirname(os.path.abspath(__file__))
agent_script = os.path.join(here, "mcp_agent.py")
resources = [
    DatabricksServingEndpoint(endpoint_name=LLM_ENDPOINT_NAME),
    DatabricksFunction("system.ai.python_exec"),
    # Uncomment to include a custom MCP server hosted on a Databricks app:
    # DatabricksApp(app_name="app-name")
]

for mcp_server_url in MANAGED_MCP_SERVER_URLS:
    mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)
    resources.extend(mcp_client.get_databricks_resources())

with mlflow.start_run():
    logged_model_info = mlflow.pyfunc.log_model(
        artifact_path="mcp_agent",
        python_model=agent_script,
        resources=resources,
    )

UC_MODEL_NAME = "main.default.databricks_docs_mcp_agent"
registered_model = mlflow.register_model(logged_model_info.model_uri, UC_MODEL_NAME)

agents.deploy(
    model_name=UC_MODEL_NAME,
    model_version=registered_model.version,
)

Étapes suivantes

Connecter les agents à des outils tiers avec les services MCP pour régir les serveurs MCP externes dans Unity Catalog.
Méta-paramètres pour les serveurs MCP gérés de Databricks pour configurer le comportement de l'outil de serveur géré avec _meta paramètres.
Connecter les MCP aux assistants IA et aux agents de codage pour utiliser les serveurs MCP de Claude, Cursor et d'autres outils.
Connectez les agents aux outils pour un aperçu de toutes les approches permettant de connecter les agents aux services externes.

Obtenez l'URL de votre serveur​

Configurez votre environnement​

Connecter et répertorier des outils​

S'authentifier​

Créer un agent​

Exemples de Notebooks​

Agent d’appel d’outils LangGraph MCP

Agent d'appel d'outils OpenAI MCP

Agent d'appel d'outils du SDK MCP

Déployez votre agent​

Étapes suivantes​