Use servidores MCP em agentes

info

Visualização

Este recurso está em Pré-lançamento público.

Conecte o código do agente a qualquer servidor MCP no Databricks — servidores gerenciados pelo Databricks, servidores MCP externos registrados como serviços MCP e servidores personalizados hospedados como aplicativos Databricks. Todos eles expõem a mesma interface MCP, portanto, o código do agente é o mesmo. O que difere é a **URL do servidor** e como autenticar.

A biblioteca Python databricks-mcp lida com a autenticação para servidores MCP do Databricks, então o mesmo código de cliente funciona em todos os três tipos de servidor.

Obter o seu URL do servidor

Configure primeiro o servidor MCP e, em seguida, utilize o URL dele nos exemplos a seguir:

Tipo de servidor	Padrão de URL	Configuração
Gerenciadas	`https://<workspace-hostname>/api/2.0/mcp/<service>/<path>`	Servidores gerenciados disponíveis
Externo (Serviço MCP)	`https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<mcp-service>`	Conecte agentes a ferramentas de terceiros com os Serviços MCP
Personalizada	`https://<app-url>/mcp`	Hospede seu próprio servidor MCP

Tipo de servidor	Padrão de URL	Configuração
Gerenciadas	`https://<workspace-hostname>/api/2.0/mcp/<service>/<path>`	Servidores gerenciados disponíveis
Externo (Serviço MCP)	`https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<mcp-service>`	Conecte agentes a ferramentas de terceiros com os Serviços MCP
Personalizada	`https://<app-url>/mcp`	Hospede seu próprio servidor MCP

Configure seu ambiente

Use OAuth para autenticar-se no seu workspace:

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

Quando solicitado, insira um nome de perfil e anote-o para mais tarde. O nome de perfil default é DEFAULT.

Verifique se você tem um ambiente local com Python 3.12 ou acima, em seguida, instale as dependências:

Bash
pip install -U "mcp>=1.9" "databricks-sdk[openai]" "mlflow>=3.1.0" "databricks-agents>=1.0.0" "databricks-mcp"

Conectar e listar ferramentas

Crie um DatabricksMCPClient com a URL do servidor e liste suas ferramentas. O mesmo cliente funciona para URLs de servidor gerenciadas, externas (serviço MCP) e personalizadas:

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]}")

Para chamar uma ferramenta diretamente:

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

nota

O compute serverless deve ser habilitado em seu workspace para executar system.ai ferramentas gerenciadas.

Autenticar

Selecione o método de autenticação que corresponde ao local onde seu agente está em execução. Para um Serviço MCP externo, o chamador também deve ter EXECUTE no serviço. O Gateway de AI aplica esta permissão em cada chamada.

Local environment
Service principal
On-behalf-of-user

Autentique-se em seu workspace com OAuth (consulte Configurar seu ambiente) e passe o perfil para o cliente:

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

Use as credenciais OAuth de uma entidade de serviço do Databricks. Passe os valores diretamente, ou recupere-os de segredos do Databricks (por exemplo, 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)

Ao log o agente, use DatabricksApps (personalizado) ou o recurso relevante como um recurso. Consulte Passagem de autenticação automática.

Use ModelServingUserCredentials para que o agente atue com as permissões do usuário que faz a chamada. Consulte autenticação em nome do usuário:

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)

Log o modelo do agente usando o escopo apps e, para servidores gerenciados, inclua o escopo OAuth correspondente para cada servidor. Veja Servidores gerenciados disponíveis.

Construir um agente

Use uma estrutura de agente para transformar as ferramentas do servidor MCP em um agente. Aponte a estrutura para a URL do servidor e passe seus WorkspaceClient autenticados.

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)

Crie um agente independente de framework que descubra e chame ferramentas em um ou mais servidores MCP. Salve o seguinte como mcp_agent.py. Ele aceita uma lista de URLs de servidores gerenciados, de Serviço MCP e personalizados:

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)

Notebooks de exemplo

Os Notebooks a seguir mostram como criar agentes LangGraph e OpenAI que chamam ferramentas MCP em servidores MCP gerenciados, externos e personalizados:

Implante seu agente

A Databricks recomenda implantar agentes no Databricks Apps, o que permite o código do agente, a configuração do servidor e o controle de versão baseado em Git totalmente gerenciados. Alternativamente, implantado no Model Serving.

Seja qual for a sua seleção, conceda ao agente acesso a cada recurso do qual seus servidores MCP dependem—por exemplo, CAN_RUN em um Genie Agent ou SELECT em um índice de pesquisa de IA.

Databricks Apps (recommended)
Model Serving

Declare cada recurso que seu agente usa—incluindo o recurso por trás de cada servidor MCP—em resources.apps.<app>.resources em databricks.yml, então o pacote foi implantado para conceder acesso à entidade de serviço Databricks do aplicativo. Por exemplo, para um agente que usa os servidores Genie e AI Search gerenciados:

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

Para o fluxo de trabalho completo de autoria e implantação, consulte Crie um agente de AI e implante-o em Databricks Apps. Para todos os tipos de recurso e valores de permissão, consulte Autenticação para agentes de IA.

Registre o agente com todos os recursos de que ele precisa no momento do log e depois implante. Consulte Implante um agente para aplicações de AI generativa (Model Serving) e Autenticação para recurso do Databricks. A Databricks recomenda o pacote databricks-mcp para derivar os recursos do servidor MCP:

Para servidores MCP gerenciados, use databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) para recuperar os recursos que o servidor precisa.
Para um servidor MCP personalizado hospedado em um aplicativo Databricks, inclua o aplicativo como um recurso ao fazer log do modelo.

Por exemplo, para implantar o agente definido em 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,
)

Próximos os passos

Conecte agentes a ferramentas de terceiros com Serviços MCP para governar servidores MCP externos no Unity Catalog.
Metaparâmetros para servidores MCP gerenciados da Databricks para configurar o comportamento da ferramenta de servidor gerenciado com _meta parâmetros.
Conecte MCPs a assistentes de AI e agentes de codificação para usar servidores MCP do Claude, Cursor e outras ferramentas.
Conectar agentes a ferramentas para uma visão geral de todas as abordagens para conectar agentes a serviços externos.

Obter o seu URL do servidor​

Configure seu ambiente​

Conectar e listar ferramentas​

Autenticar​

Construir um agente​

Notebooks de exemplo​

Agente de chamada de ferramentas LangGraph MCP

Agente de chamada de ferramentas MCP da OpenAI

Agente de invocação de ferramentas do SDK MCP

Implante seu agente​

Próximos os passos​