Usar servidores MCP em Agentes Personalizados

Ligue o seu código de agente a qualquer servidor MCP no Azure Databricks: servidores geridos por Databricks, servidores MCP externos registados como Serviços MCP e servidores personalizados alojados como aplicações Databricks. Todos eles expõem a mesma interface MCP, por isso o código do agente é o mesmo. O que difere é a URL do servidor e a forma como se autentica.

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

Obtenha o URL do seu servidor

Configure primeiro o servidor MCP e depois use o seu URL nos seguintes exemplos:

Descubra os servidores e ferramentas MCP disponíveis

Antes de escreveres código para agentes, descobre que servidores e ferramentas podes usar. Não codifique diretamente nomes de servidores, nomes de ferramentas ou formas de argumentos a partir da memória; Confirma-os a partir do espaço de trabalho.

• Navegue pelos servidores no espaço de trabalho. Vai a AI Gateway>MCPs para veres os servidores MCP disponíveis para ti. O Azure Databricks disponibiliza servidores integrados prontos a usar: servidores MCP geridos para os seus próprios dados e funções do Catálogo Unity (espaços Genie, índices de Pesquisa Vetorial e funções do Catálogo Unity), e Serviços MCP integrados system.ai para ferramentas SaaS de terceiros como Slack, GitHub, Google Drive, Google Calendar, Gmail e Microsoft 365.

• Liste os Serviços MCP programáticamente. Liste os Serviços MCP em qualquer catálogo e esquema com a API REST do Unity Catalog. Por exemplo, os serviços incorporados:

  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 os serviços que registou. page_size está limitado a 100, e a resposta inclui um next_page_token quando existem mais serviços. Para enumerar todos os serviços num esquema, repita o pedido com page_token=<next_page_token> até que a resposta não retorne nenhum token:

  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 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, conforme mostrado em Connect and list tools. Esta é a forma fiável de aprender as ferramentas e argumentos exatos de um servidor.

Configura o teu ambiente

1. Use o OAuth para autenticar no seu espaço de trabalho:

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

2. Quando solicitado, introduza um nome de perfil e anote-o para mais tarde. O nome padrão do perfil é DEFAULT.

3. Verifica se tens um ambiente local com Python 3.12 ou superior, depois instala as dependências:

   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 do seu próprio SDK. Adicionar openai-agents databricks-openai para o SDK OpenAI Agents, ou databricks-langchain langgraph para LangGraph.

Ligar e listar ferramentas

Cria um DatabricksMCPClient com o URL do servidor e lista as suas ferramentas. O mesmo cliente é compatível com URLs de servidor geridas, externas (MCP Service) e personalizadas:

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 ligar diretamente a uma ferramenta:

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

Authenticate

Selecione o método de autenticação que corresponda ao local onde o seu agente está em execução. Para um serviço MCP externo, o chamador também tem de ter EXECUTE no serviço. O AI Gateway aplica esta permissão em todas as chamadas.

Ambiente local

Autentique no seu espaço de trabalho com OAuth (veja Configurar o seu ambiente) e passe o perfil ao cliente:

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

Serviço principal

Utilize as credenciais OAuth do principal de serviço. Passe os valores diretamente, ou recupere-os dos segredos do Azure Databricks (por exemplo, client_id=dbutils.secrets.get(scope="my-scope", key="client-id")):

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)

Quando registares o agente, usa DatabricksApps (personalizado) ou o recurso relevante como recurso. Consulte Passagem de autenticação automática.

Em nome do utilizador

Use ModelServingUserCredentials para que o agente atue com as permissões do utilizador que chama. Consulte autenticação em nome do utilizador:

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)

Registe o modelo do agente utilizando o âmbito apps e, para servidores geridos, inclua o âmbito OAuth correspondente para cada servidor. Ver Servidores geridos disponíveis.

Para um Serviço MCP externo ou incorporado, em vez disso, adicione o âmbito da API de utilizador ai-gateway (user_api_scopes: [ai-gateway]) e conceda ao utilizador que efetua a chamada EXECUTE no serviço. Veja Autenticar aos Serviços MCP.

Criar um agente

Usa uma framework de agentes para transformar as ferramentas do servidor MCP num agente. Aponte o framework para o URL do servidor e forneça as suas credenciais de autenticação WorkspaceClient.

SDK de agentes OpenAI

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

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

Construa um agente independente do framework que descubra e chame ferramentas através de um ou mais servidores MCP. Guarde o seguinte como mcp_agent.py. Aceita uma lista de URLs de servidores geridos, do MCP Service e personalizados:

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)

Exemplos de cadernos

Os cadernos seguintes mostram como construir agentes LangGraph e OpenAI que chamam ferramentas MCP através de servidores MCP geridos, externos e personalizados:

Agente de invocação de ferramentas LangGraph MCP

Obter caderno

Agente de chamada de ferramentas MCP da OpenAI

Obter caderno

Agente de chamador de ferramenta MCP do SDK de Agentes

Obter caderno

** Desdobre o seu agente

O Azure Databricks recomenda implementar agentes nas Databricks Apps, o que permite gerir totalmente o código do agente, a configuração do servidor e o versionamento baseado em git. Em alternativa, implemente em Model Serving.

Seja qual for a escolha, conceda ao agente acesso a todos os recursos de que dependem os seus servidores MCP. Por exemplo, CAN_RUN num Agente Génio ou SELECT num índice de Pesquisa por IA.

Databricks Apps (recomendado)

Declare cada recurso que o seu agente utiliza, incluindo os recursos por trás de cada servidor MCP, em resources.apps.<app>.resourcesdatabricks.yml, e depois implemente o bundle para conceder acesso ao principal de serviço da aplicação. Por exemplo, para um agente que utiliza os servidores geridos Genie e AI Search:

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'

databricks bundle deploy
databricks bundle run my_agent_app

Para o fluxo completo de autoria e implementação, consulte Criar um agente e implementá-lo nas Aplicações Databricks. Para todos os tipos de recursos e valores de permissões, veja Autenticação para agentes.

Serviço de Modelos

Registe o agente com todos os recursos de que necessita no momento do registo e, em seguida, implemente-o. Veja Implementar um agente para aplicações de IA (Model Serving) e Autenticação para recursos do Databricks. O Azure Databricks recomenda o databricks-mcp pacote para derivar recursos do servidor MCP:

• Para servidores MCP geridos, usar databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) para recuperar os recursos de que o servidor precisa.
• Para um servidor MCP personalizado alojado numa aplicação Databricks, inclua a aplicação como recurso ao registar o modelo.

Por exemplo, para implantar o agente definido em mcp_agent.py:

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,
)

Passos seguintes

• Ligue os agentes a ferramentas de terceiros com os Serviços MCP para gerir servidores MCP externos no Unity Catalog.
• AI Search e Databricks SQL para configurar o comportamento de ferramentas de servidor geridas com os parâmetros _meta.
• Ligue MCPs a assistentes de IA e agentes de programação para usar servidores MCP do Claude, Cursor e outras ferramentas.
• Ligue agentes a MCPs e ferramentas externas para uma visão geral de todas as abordagens para ligar agentes a serviços externos.