Pular para o conteúdo principal

Utilize o gerenciamento de servidores MCP ( Databricks ) para gerenciar servidores MCP.

info

Beta

Este recurso está em fase beta.

Os servidores do Model Context Protocol (MCP) atuam como pontes que permitem que agentes AI acessem dados e ferramentas externas. Em vez de criar essas conexões do zero, use os servidores MCP do Databricks para conectar instantaneamente seus agentes aos dados armazenados no Unity Catalog, índices de pesquisa de vetores e funções personalizadas.

Servidores gerenciáveis disponíveis​

Databricks oferece três tipos de servidores gerenciar MCP que funcionam imediatamente:

Servidor MCP

Descrição

Padrão de URL

Vector Search

Índices de pesquisa do Query Vector para encontrar documentos e dados relevantes. Somente índices com incorporações gerenciais Databricks são suportados.

https://<workspace-hostname>/api/2.0/mcp/vector-search/{catalog}/{schema}

Unity Catalog funções

A execução de funções como Unity Catalog funciona de maneira semelhante às ferramentas personalizadas Python ou SQL.

https://<workspace-hostname>/api/2.0/mcp/functions/{catalog}/{schema}

Espaço Genie

Consulta Genie spaces para obter percepções a partir de tabelas de dados estruturados

https://<workspace-hostname>/api/2.0/mcp/genie/{genie_space_id}

nota

O servidor gerenciar MCP para Genie invoca Genie como uma ferramenta MCP, o que significa que o histórico não é passado ao invocar as APIs Genie . Como alternativa, você pode usar o Genie em um sistema multiagente.

Cenário de exemplo​

Imagine que você queira criar um agente que ajude no suporte ao cliente. É possível conectá-lo a vários servidores gerenciar MCP:

  • Pesquisa vetorial : https://<workspace-hostname>/api/2.0/mcp/vector-search/prod/customer_support

    • Pesquisa tickets de suporte e documentação

  • Espaço Genie : https://<workspace-hostname>/api/2.0/mcp/genie/{billing_space_id}

    • Consultar dados de faturamento e informações do cliente

  • Funções de UC : https://<workspace-hostname>/api/2.0/mcp/functions/prod/billing

    • Execução de funções personalizadas para pesquisas e atualizações de account

Isso dá ao seu agente acesso a dados não estruturados (tickets de suporte), dados estruturados (tabelas de cobrança) e lógica de negócios personalizada.

Exemplo de Notebook: Construir um agente com gerenciar servidores MCP​

O Notebook a seguir demonstra como criar agentes LangGraph e OpenAI que chamam ferramentas MCP.

Agente de chamada de ferramentas LangGraph MCP

Open notebook in new tab

Agente de chamada de ferramentas OpenAI MCP

Open notebook in new tab

IDE local: Crie um agente com gerenciar servidores MCP​

Conectar-se a um servidor MCP no Databricks é semelhante a conectar-se a qualquer outro servidor MCP remoto. Você pode se conectar ao servidor usando SDKs padrão, como o MCP Python SDK. A principal diferença é que os servidores MCP Databricks são seguros por default e exigem que os clientes especifiquem autenticação.

A biblioteca Python databricks-mcp auxilia na simplificação da autenticação em código de agente personalizado.

A maneira mais simples de desenvolver código de agente é executá-lo localmente e autenticar em seu workspace. Siga estes passos para criar um agente AI que se conecta a um servidor MCP Databricks .

Configure seu ambiente​

  1. Use OAuth para autenticar seu workspace. execute o seguinte comando em um terminal local:

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

  2. Crie um nome de perfil quando solicitado e anote esse nome para uso posterior.

  3. Certifique-se de que possui um ambiente local com o Python 3.12 ou superior 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"

Teste sua conexão com o ambiente local​

Valide sua conexão com o servidor MCP listando suas ferramentas do Unity Catalog e executando a ferramenta de interpretação de código Python integrada.

computesem servidor deve estar habilitado em seu workspace para executar este snippet.

  1. execute o seguinte código para validar sua conexão com o servidor MCP.
Python
from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient

# TODO: Update to the Databricks CLI profile name you specified when
# configuring authentication to the workspace.
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)
workspace_hostname = workspace_client.config.host
mcp_server_url = f"{workspace_hostname}/api/2.0/mcp/functions/system/ai"

# This code uses the Unity Catalog functions MCP server to expose built-in
# AI tools under `system.ai`, like the `system.ai.python_exec` code interpreter tool
def test_connect_to_server():
    mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)
    tools = mcp_client.list_tools()

    print(
        f"Discovered tools {[t.name for t in tools]} "
        f"from MCP server {mcp_server_url}"
    )

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


if __name__ == "__main__":
    test_connect_to_server()

Crie seu agente​

  1. Desenvolva o código acima para definir um agente básico de turno único que usa ferramentas. Salve o código do agente localmente como um arquivo chamado mcp_agent.py:

    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

     # 1) CONFIGURE YOUR ENDPOINTS/PROFILE
     LLM_ENDPOINT_NAME = "databricks-claude-3-7-sonnet"
     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 MCP server URLs here if desired, for example:
     # f"{host}/api/2.0/mcp/vector-search/prod/billing"
     # to include vector search indexes under the prod.billing schema, or
     # f"{host}/api/2.0/mcp/genie/<genie_space_id>"
     # to include a Genie space
     MANAGED_MCP_SERVER_URLS = [
         f"{host}/api/2.0/mcp/functions/system/ai",
     ]
     # Add Custom MCP Servers hosted on Databricks Apps
     CUSTOM_MCP_SERVER_URLS = []



     # 2) HELPER: convert between ResponsesAgent “message dict” and ChatCompletions format
     def _to_chat_messages(msg: dict[str, Any]) -> List[dict]:
         """
         Take a single ResponsesAgent‐style dict and turn it into one or more
         ChatCompletions‐compatible dict entries.
         """
         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:
             # fallback for plain {"role": ..., "content": "..."} or similar
             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):
             """
             Send current history → LLM, returning the raw response dict.
             """
             client = ws.serving_endpoints.get_open_ai_client()
             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)

             # 1) build initial history: system + user
             history: List[dict] = [{"role": "system", "content": SYSTEM_PROMPT}]
             for inp in request.input:
                 history.append(inp.model_dump())

             # 2) call LLM once
             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:
                 # (we only support a single tool in this “single‐turn” example)
                 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}"

                 # 4) append the “tool” output
                 history.append(
                     {
                         "type": "function_call_output",
                         "role": "tool",
                         "id": uuid.uuid4().hex,
                         "tool_call_id": fc["id"],
                         "output": result,
                     }
                 )

                 # 5) call LLM a second time and treat that reply as final
                 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,
                 )

             # 6) if no tool_calls at all, return the assistant’s original reply
             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)

implantado seu agente​

Quando estiver pronto para implantar um agente que se conecta aos servidores MCP do gerenciador, siga o processo padrão de implantação do agente.

Certifique-se de especificar todos os recursos aos quais o seu agente precisa ter acesso no momento do registro. Por exemplo, se seu agente usa os seguintes URLs do servidor MCP:

  • https://<your-workspace-hostname>/api/2.0/mcp/vector-search/prod/customer_support
  • https://<your-workspace-hostname>/api/2.0/mcp/vector-search/prod/billing
  • https://<your-workspace-hostname>/api/2.0/mcp/functions/prod/billing

Você deve especificar todos os índices de pesquisa de vetores que seu agente precisa nos esquemas prod.customer_support e prod.billing como recurso, bem como todas as funções do Unity Catalog em prod.billing.

Se o seu agente se conectar aos servidores MCP no Databricks para descobrir e executar ferramentas, log o recurso necessário para esses servidores MCP com seu agente. A Databricks recomenda instalar o pacote PyPI databricks-mcp para simplificar esse processo.

Em particular, se você usar servidores gerenciar MCP, você pode usar databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) para recuperar o recurso necessário para o servidor gerenciar MCP. Se o seu agente consultar um servidor MCP personalizado hospedado no aplicativo Databricks, você poderá configurar a autorização incluindo explicitamente o servidor como um recurso ao registrar seu modelo.

Por exemplo, para implementar o agente definido acima, execute o seguinte código, supondo que você salvou a definição do código do agente 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

# TODO: Update this to your Databricks CLI profile name
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)


# Configure MLflow and the Databricks SDK to use your Databricks CLI profile
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",
]
# Log the agent defined in mcp_agent.py
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 and edit the following lines to include custom mcp servers hosted on Databricks Apps ---
    # 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,
    )

# TODO Specify your UC model name here
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​

Última atualização em