Usar servidores MCP em Agentes personalizados

info

Visualização

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

Conecte seu código de 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 você se autentica .

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

Obtenha a URL do seu servidor

Configure o servidor MCP primeiro e, em seguida, use seu URL 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 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 Serviços MCP
Personalizada	`https://<app-url>/mcp`	Hospede seu próprio servidor MCP

Configure seu ambiente

Use OAuth para autenticar no seu workspace:

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

Quando solicitado, insira um nome de perfil e anote-o para uso posterior. O nome do perfil default é DEFAULT.

Verifique se você possui um ambiente local com Python 3.12 ou acima e, 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

Serverless compute deve estar habilitado em seu workspace para a execução de ferramentas system.ai gerenciadas.

Autenticar

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

Local environment
Service principal
On-behalf-of-user

Autentique-se no 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 Service Principal 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 registrar 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 o chamou. 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)

Faça o log do modelo de agente usando o escopo apps e, para servidores gerenciados, inclua o escopo OAuth correspondente para cada servidor. Consulte Servidores gerenciados disponíveis.

Criar um agente

Use uma estrutura de agente para transformar as ferramentas do servidor MCP em um agente. Aponte o framework para a URL do servidor e passe seu WorkspaceClient autenticado.

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 descobre e chama 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 seguintes Notebooks mostram como criar agentes LangGraph e OpenAI que chamam ferramentas MCP em servidores MCP gerenciados, externos e personalizados:

Implante seu agente

O Databricks recomenda implantar agentes no Databricks Apps, o que permite gerenciar totalmente o código do agente, a configuração do servidor e o versionamento baseado em git. Alternativamente, implante no Model Serving.

Independentemente do que você selecionar, conceda ao agente acesso a todos os recursos dos quais seus servidores MCP dependem. Por exemplo, CAN_RUN em um Genie Agent ou SELECT em um Índice de Pesquisa de AI.

Databricks Apps (recommended)
Model Serving

Declare cada recurso que seu agente usa, incluindo os recursos por trás de cada servidor MCP, em resources.apps.<app>.resources no databricks.yml, e então faça o deploy do pacote para conceder acesso ao service principal do Databricks do aplicativo. Por exemplo, para um agente que usa os servidores gerenciados Genie e 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

Para o fluxo de trabalho completo de criação e implantação, consulte Criar um agente e implantá-lo nos Databricks Apps. Para todos os tipos de recurso e valores de permissão, consulte Autenticação para agentes.

Registre o agente com todos os recursos necessários no momento do registro e, em seguida, implante-o. Consulte Implantar um agente para aplicativos de AI (Model Serving) e Autenticação para recursos do Databricks. A Databricks recomenda o pacote databricks-mcp para derivar recursos do servidor MCP:

Para servidores MCP gerenciados, use databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) para recuperar os recursos de que o servidor precisa.
Para um servidor MCP personalizado hospedado em um aplicativo do Databricks, inclua o aplicativo como um recurso ao registrar o 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 passos

Conecte agentes a ferramentas de terceiros com serviços MCP para governar servidores MCP externos no Unity Catalog.
AI Search e Databricks SQL 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.
Connect agents to external MCPs and tools para obter uma visão geral de todas as abordagens para conectar agentes a serviços externos.

Obtenha a URL do seu servidor​

Configure seu ambiente​

Conectar e listar ferramentas​

Autenticar​

Criar um agente​

Notebooks de exemplo​

Agente de chamada de ferramentas LangGraph MCP

Agente de chamada de ferramenta OpenAI MCP

Agente de chamada de ferramenta MCP do SDK de agentes

Implante seu agente​

Próximos passos​