Use servidores MCP em Agentes Personalizados

Conecte seu código de agente a qualquer servidor MCP no Azure 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 de Python databricks-mcp manipula a autenticação para Azure Databricks servidores MCP, de modo que o mesmo código do cliente funciona em todos os três tipos de servidor.

Obter a URL do servidor

Configure o servidor MCP primeiro e, em seguida, use sua URL nos seguintes exemplos:

Configure seu ambiente

1. Use o OAuth para autenticar em seu workspace:

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

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

3. Verifique se você tem um ambiente local com Python 3.12 ou superior e instale dependências:

   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:

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:

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 corresponde ao local em que o agente é executado. Para um serviço MCP externo, o chamador também deve ter EXECUTE no serviço. O Gateway de IA impõe essa permissão em cada chamada.

Ambiente local

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

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

Entidade de serviço

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

Ao registrar o agente em log, use DatabricksApps (personalizado) ou o recurso relevante como um recurso. Confira a passagem de autenticação automática.

Em nome do usuário

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

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)

Registre o modelo de agente usando o apps escopo e, para servidores gerenciados, inclua o escopo OAuth correspondente para cada servidor. Consulte os 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 sua credencial autenticada WorkspaceClient.

SDK de agentes do 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)

SDK MCP para Python

Crie um agente independente de estrutura que descubra e chame ferramentas em um ou mais servidores MCP. Salve o seguinte como mcp_agent.py. Ele aceita uma lista de URLs gerenciadas, de URLs do Serviço MCP e de URLs de servidores 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 notebooks

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

Agente de invocação de ferramentas LANGGraph MCP

Obter laptop

Agente de chamamentos de ferramentas do OpenAI MCP

Obter laptop

Agente de chamada de ferramenta do Agentes SDK-MCP

Obter laptop

Implantar seu agente

Azure Databricks recomenda a implantação de agentes nos Aplicativos do Databricks, o que permite gerenciar totalmente o código do agente, a configuração do servidor e o controle de versão baseado em git. Como alternativa, faça a implantação no Model Serving.

Qualquer que seja a escolha, conceda ao agente acesso a todos os recursos dos quais seus servidores MCP dependem. Por exemplo, CAN_RUN em um Agente Genie ou SELECT em um índice de busca por IA.

Aplicativos do Databricks (recomendado)

Declare cada recurso que seu agente usa, incluindo os recursos por trás de cada servidor MCP, em resources.apps.<app>.resourcesdatabricks.yml, e então implante o pacote para conceder acesso ao principal de serviço do app. Por exemplo, para um agente que usa os servidores Genie e AI Search gerenciados:

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 obter o fluxo de trabalho completo de criação e implantação, consulte Criar um agente e implantá-lo nos Aplicativos do Databricks. Para todos os tipos de recursos e valores de permissão, consulte Autenticação para agentes.

Serviço de Modelo

Registre o agente com todos os recursos necessários no momento do registro e, em seguida, implante. Consulte Implantar um agente para aplicativos de IA (Model Serving) e Autenticação para recursos do Databricks. Azure Databricks recomenda o databricks-mcp pacote 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 em log.

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

Próximas Etapas 

• Conecte agentes a ferramentas de terceiros com os Serviços MCP para controlar servidores MCP externos no Catálogo do Unity.
• AI Search e Databricks SQL para configurar o comportamento de ferramentas de servidores gerenciados com _meta parâmetros.
• Conecte MCPs a assistentes de IA e agentes de codificação para usar servidores MCP de Claude, Cursor e outras ferramentas.
• Conecte agentes a MCPs e ferramentas externas para uma visão geral de todas as abordagens de conectar agentes a serviços externos.