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

Configure seu ambiente

1. Use OAuth para autenticar-se no seu workspace:

   Bash

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

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

3. 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

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)

Service principal

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.

On-behalf-of-user

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

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

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:

Agente de chamada de ferramentas LangGraph MCP

Abrir notebook em uma nova aba

Agente de chamada de ferramentas MCP da OpenAI

Abrir notebook em uma nova aba

Agente de invocação de ferramentas do SDK MCP

Abrir notebook em uma nova aba

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.

Qualquer que seja a sua seleção, conceda ao agente acesso a cada recurso do qual seus servidores MCP dependem—por exemplo, CAN_RUN em um Genie Space ou SELECT em um índice de pesquisa de AI.

Databricks Apps (recommended)

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.

Model Serving

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.