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 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 com serviços MCP
Personalizada	`https://<app-url>/mcp`	Hospede seu próprio servidor MCP

Descubra servidores e ferramentas MCP disponíveis

Antes de escrever código de agente, descubra quais servidores e ferramentas você pode usar. Não codifique nomes de servidor, nomes de ferramenta ou formatos de argumento de memória; confirme-os no workspace.

Navegue pelos servidores no workspace. Vá para AI Gateway > MCPs para ver os servidores MCP disponíveis para você. O Databricks fornece servidores integrados prontos para uso: servidores MCP gerenciados para seus próprios dados e funções do Unity Catalog (Genie spaces, índices de Vector Search e funções do Unity Catalog), e system.ai serviços MCP integrados para ferramentas SaaS de terceiros, como Slack, GitHub, Google Drive, Google Calendar, Gmail e Microsoft 365.
Liste os serviços MCP programaticamente. Liste os serviços MCP em qualquer catálogo e esquema com a API REST do Unity Catalog. Por exemplo, os serviços integrados:
Bash
```
databricks api get "/api/2.1/unity-catalog/mcp-services?parent=schemas/system.ai&page_size=100"
```
Substitua system.ai pelo seu próprio <catalog>.<schema> para encontrar serviços que você registrou. page_size é limitado a 100, e a resposta inclui um next_page_token quando existem mais serviços. Para enumerar todos os serviços em um esquema, repita a solicitação com page_token=<next_page_token> até que a resposta não retorne nenhum token:
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
```
Liste as ferramentas de um servidor a partir do código. Aponte um DatabricksMCPClient para qualquer URL de servidor e chame list_tools() para obter o nome, a descrição e o esquema de entrada de cada ferramenta em tempo de execução (runtime), conforme mostrado em Conectar e listar ferramentas. Esta é a maneira confiável de aprender as ferramentas e os argumentos exatos de um servidor.

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"
```
Os exemplos de agent-framework abaixo precisam de seu próprio SDK. Adicione openai-agents databricks-openai para o OpenAI Agents SDK, ou databricks-langchain langgraph para o LangGraph.

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.

Para um serviço MCP externo ou integrado, adicione o escopo de API de usuário ai-gateway (user_api_scopes: [ai-gateway]) e conceda ao usuário chamador EXECUTE no serviço. Consulte Autenticar em serviços MCP.

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.

nota

Comece a partir de um padrão de agente: ele fornece o ponto de entrada AgentServer do MLflow (execução com uv run start-app), o auxiliar get_user_workspace_client() em nome de, e um databricks.yml. Pin o interpretador com requires-python = ">=3.12,<3.13" e faça commit de uv.lock para que a imagem de build do Databricks Apps não resolva um Python mais recente (por exemplo, 3.14) que não possua wheels pré-construídas para algumas dependências do agente. Para um serviço MCP, conceda também ao chamador acesso fora de banda (consulte Habilitar acesso por usuário (acesso em nome do usuário)); bundle validate passa sem isso.

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

Conectar agentes a ferramentas com serviços MCP para governar servidores MCP externos no Unity Catalog.
Servidor MCP de Pesquisa AI e servidor MCP do Databricks SQL para configurar o comportamento da ferramenta de servidor gerenciado com parâmetros _meta.
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​

Descubra servidores e ferramentas MCP disponíveis​

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​