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 MCP (Model Context Protocol) atuam como pontes que permitem que os agentes AI acessem dados e ferramentas externos. Em vez de criar essas conexões do zero, é possível utilizar o gerenciamento de servidores MCP ( Databricks ) para conectar instantaneamente seus agentes aos dados armazenados em Unity Catalog, índices de pesquisa vetorial 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

Consulte índices de pesquisa vetorial para encontrar documentos e dados relevantes

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 história não é transmitido ao invocar o Genie APIs. Como alternativa, é possível utilizar o Genie em um sistema multiagente.

Exemplo: agente de suporte ao cliente​

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 (tíquetes de suporte) e dados estruturados (tabelas de cobrança), além de lógica comercial personalizada.

Notebook: Configure 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​

A conexão com um servidor MCP no Databricks é semelhante à conexão com qualquer outro servidor MCP remoto. É possível conectar-se ao servidor utilizando SDKs padrão, como o MCP Python SDK. A principal diferença é que os servidores MCP Databricks são protegidos por default e exigem que os clientes especifiquem a 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 o código do agente é executá-lo localmente e autenticar-se em workspace. Siga as etapas a seguir para criar um agente AI que se conecte a um servidor MCP Databricks.

Configure seu ambiente​

  1. Utilize OAuth para autenticar-se em workspace. Execute o seguinte em um terminal local:

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

  2. Crie um nome de perfil quando solicitado e lembre-se desse nome.

  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 interpretadora de código Python integrada.

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

  1. Execute o seguinte trecho de 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 snippet below 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. Com base no trecho acima, defina um agente básico de turno único que utilize 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, e.g
     # 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 gerenciar MCP, utilize o processo padrão de implantação de agentes.

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

É necessário especificar todos os índices de pesquisa vetorial necessários ao seu agente nos esquemas prod.customer_support e prod.billing como recurso, bem como todas as funções Unity Catalog em prod.billing.

Se o seu agente se conectar a servidores MCP em Databricks para descobrir e executar ferramentas, certifique-se de log o recurso necessário para esses servidores MCP com o seu agente. A Databricks recomenda a instalação do pacote databricks-mcp do PyPI para simplificar esse processo.

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

Por exemplo, para implantar o agente definido acima, é possível executar o seguinte trecho de código, presumindo 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