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 aux outils 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 aux outils avec les services MCP
Personnalisé	`https://<app-url>/mcp`	Hébergez votre propre serveur MCP

Découvrez les serveurs et outils MCP disponibles

Avant d’écrire le code de l’agent, déterminez quels serveurs et outils vous pouvez utiliser. Ne codez pas en dur les noms de serveurs, les noms d’outils ou les formes d’arguments de mémoire ; confirmez-les à partir du Workspace.

Parcourir les serveurs dans le workspace. Accédez à AI Gateway > MCPs pour voir les serveurs MCP qui vous sont disponibles. Databricks fournit des serveurs intégrés prêts à l’emploi : serveurs MCP gérés pour vos propres données et fonctions Unity Catalog (Genie spaces, index de recherche vectorielle et fonctions Unity Catalog), ainsi que des system.ai services MCP intégrés pour des outils SaaS tiers tels que Slack, GitHub, Google Drive, Google Calendar, Gmail et Microsoft 365.
Lister les services MCP par programmation. Lister les services MCP dans n’importe quel catalogue et schéma avec l’API REST Unity Catalog. Par exemple, les services intégrés :
Bash
```
databricks api get "/api/2.1/unity-catalog/mcp-services?parent=schemas/system.ai&page_size=100"
```
Remplacez system.ai par votre propre <catalog>.<schema> pour trouver les services que vous avez enregistrés. page_size est limité à 100 et la réponse inclut un next_page_token lorsque d'autres services existent. Pour énumérer chaque service dans un schéma, répétez la requête avec page_token=<next_page_token> jusqu’à ce que la réponse ne renvoie aucun jeton :
Bash
```
token=""
while :; do
  page=$(databricks api get "/api/2.1/unity-catalog/mcp-services?parent=schemas/system.ai&page_size=100&page_token=$token")
  echo "$page"
  token=$(echo "$page" | jq -r '.next_page_token // empty')
  [ -z "$token" ] && break
done
```
Lister les outils d’un serveur à partir du code. Pointez un DatabricksMCPClient vers n’importe quelle URL de serveur et appelez list_tools() pour obtenir le nom, la description et le schéma d’entrée de chaque outil au moment de l’exécution, comme indiqué dans Connecter et lister les outils. C’est le moyen fiable de connaître les outils et arguments exacts d’un serveur.

Configurez votre environnement

Utilisez OAuth pour vous authentifier sur 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"
```
Les exemples de framework d’agents ci-dessous nécessitent leur propre SDK. Ajoutez openai-agents databricks-openai pour le SDK OpenAI Agents, ou databricks-langchain langgraph pour LangGraph.

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
Service principal
On-behalf-of-user

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)

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.

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.

Pour un service MCP externe ou intégré, ajoutez plutôt l'étendue d'API utilisateur ai-gateway (user_api_scopes: [ai-gateway]) et accordez à l'utilisateur appelant EXECUTE sur le service. Voir S'authentifier auprès des services MCP.

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
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 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 :

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

remarque

Start from an agent template: il fournit le point d’entrée MLflow AgentServer (exécuté avec uv run start-app), l’assistant get_user_workspace_client() pour l’accès au nom de l’utilisateur, et un databricks.yml. Pin l’interpréteur avec requires-python = ">=3.12,<3.13" et commit uv.lock afin que l’image de build Databricks Apps ne résolve pas une version plus récente de Python (par exemple, 3.14) qui ne dispose pas de wheels préconstruits pour certaines dépendances de l’agent. Pour un service MCP, accordez également à l’appelant un accès hors bande (voir Activer l’accès par utilisateur (accès au nom de l’utilisateur)) ; bundle validate passe sans cela.

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

Connectez les agents aux outils avec les services MCP pour régir les serveurs MCP externes dans Unity Catalog.
Serveur MCP AI Search et serveur MCP Databricks SQL pour configurer le comportement de l’outil de serveur géré avec _meta parameters.
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.

Obtenez l’URL de votre serveur​

Découvrez les serveurs et outils MCP disponibles​

Configurez votre environnement​

Connecter et lister les 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 MCP du SDK Agents

Déployez votre agent​

Étapes suivantes​