Utiliser des serveurs MCP dans des agents personnalisés

info

Aperçu

Cette fonctionnalité est en Aperçu public.

Connectez le code de votre 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. Ils exposent tous la même interface MCP, le code de l'agent est donc identique. Ce qui diffère, c'est 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 MCP Databricks, 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ébergez 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ébergez votre propre serveur MCP

Configurez votre environnement

1. Utilisez OAuth pour vous authentifier sur votre workspace :

   Bash

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

2. Lorsque vous y êtes invité, saisissez un nom de profil et notez-le pour plus tard. Le nom de profil default est DEFAULT.

3. 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 lister les 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 system.ai gérés.

S’authentifier

Sélectionnez la méthode d’authentification qui correspond à l’emplacement 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

Authentifiez-vous sur 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)

Service principal

Utilisez les identifiants 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 enregistrez l'agent, utilisez DatabricksApps (personnalisé) ou la ressource pertinente en tant que ressource. Voir Transmission automatique des identifiants.

On-behalf-of-user

Utilisez ModelServingUserCredentials afin que l’agent agisse avec les autorisations de l’utilisateur appelant. Voir l’authentification utilisateur « Au nom de »:

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)

Enregistrez le modèle d'agent en utilisant la portée apps et, pour les serveurs gérés, incluez la portée OAuth correspondante 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 agent. Pointez le framework vers l'URL du serveur et transmettez votre WorkspaceClient authentifié.

OpenAI Agents 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())

LangGraph

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)

MCP Python SDK

Créez un agent indépendant du framework qui découvre et appelle des outils sur un ou plusieurs serveurs MCP. Enregistrez ce qui suit sous mcp_agent.py. Il accepte une liste d'URL de serveurs gérés, de services MCP et de serveurs 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 :

Agent d'appel d'outils LangGraph MCP

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Agent d'appel d'outils OpenAI MCP

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Agent d'appel d'outils MCP du SDK Agents

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Déployez votre agent

Databricks recommande de déployer les agents sur Databricks Apps, ce qui vous permet de gérer entièrement le code de l'agent, la configuration du serveur et le versioning basé sur Git. Alternativement, déployez sur Model Serving.

Quel que soit votre choix, 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)

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 et le déployer sur Databricks Apps. Pour tous les types de ressources et les valeurs d’autorisation, consultez Authentification des agents.

Model Serving

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

• Pour les serveurs MCP gérés, utilisez databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) afin de 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 lors de l’enregistrement du 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.
• AI Search et Databricks SQL pour configurer le comportement de l'outil serveur géré avec _meta paramètres.
• Connectez 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 à des MCP et outils externes pour obtenir une vue d’ensemble de toutes les approches permettant de connecter des agents à des services externes.