Uso de servidores MCP administrados de Databricks

Los servidores MCP administrados de Databricks son servidores listos para usar que conectan los agentes de IA a los datos almacenados en el Catálogo de Unity, los índices de Búsqueda vectorial de Databricks, los espacios de Genie y las funciones personalizadas.

Los permisos de Unity Catalog siempre se aplican, por lo que los agentes y los usuarios solo pueden acceder a las herramientas y datos para los que tienen permiso.

Servidores administrados disponibles

Databricks proporciona los siguientes servidores MCP que funcionan de forma predeterminada.

Para ver los servidores MCP y sus direcciones URL de punto de conexión, vaya al área de trabajo >agentes>servidores MCP:

Escenario de ejemplo

Considere un agente personalizado que ayude con el soporte técnico al cliente. Puede conectarlo a varios servidores MCP administrados:

• Búsqueda vectorial: https://<workspace-hostname>/api/2.0/mcp/vector-search/prod/customer_support
  • Buscar tickets de soporte y documentación
• Espacio de Genie: https://<workspace-hostname>/api/2.0/mcp/genie/{billing_space_id}
  • Consulta de datos de facturación e información del cliente
• Funciones UC: https://<workspace-hostname>/api/2.0/mcp/functions/prod/billing
  • Ejecución de funciones personalizadas para búsquedas y actualizaciones de cuentas

Esto proporciona al agente acceso a datos no estructurados (vales de soporte técnico), datos estructurados (tablas de facturación) y lógica de negocios personalizada.

Cuadernos de ejemplo: Compilación de un agente con servidores MCP de Databricks

En los cuadernos siguientes se muestra cómo crear agentes LangGraph y OpenAI que llaman a herramientas MCP mediante servidores MCP gestionados por Databricks.

Agente de llamada a herramientas MCP de LangGraph

Obtener el cuaderno

Agente de llamada a herramientas de MCP de OpenAI

Obtener el cuaderno

Agente de llamada de herramientas MCP del SDK de agentes

Obtener el cuaderno

Construir un agente desde un entorno local

La conexión a un servidor MCP en Databricks es similar a la conexión a cualquier otro servidor MCP remoto. Puede conectarse al servidor mediante SDK estándar, como el SDK de Python de MCP. La principal diferencia es que los servidores MCP de Databricks son seguros de forma predeterminada y requieren que los clientes especifiquen la autenticación.

La biblioteca de Python databricks-mcp ayuda a simplificar la autenticación en código de agente personalizado.

La manera más sencilla de desarrollar código de agente es ejecutarlo localmente y autenticarse en el área de trabajo. Siga estos pasos para crear un agente de IA que se conecte a un servidor MCP de Databricks.

Configuración del entorno

1. Use OAuth para autenticarse en el área de trabajo. Ejecute el comando siguiente en un terminal local:

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

2. Cuando se le solicite, escriba un nombre de perfil y anótelo para su uso posterior. El nombre de perfil predeterminado es DEFAULT.

3. Asegúrese de que tiene un entorno local con Python 3.12 o superior y, a continuación, instale las dependencias:

   pip install -U "mcp>=1.9" "databricks-sdk[openai]" "mlflow>=3.1.0" "databricks-agents>=1.0.0" "databricks-mcp"
   

Prueba de la conexión del entorno local

El proceso sin servidor debe estar habilitado en el área de trabajo para ejecutar este fragmento de código.

Valide la conexión con el servidor MCP enumerando las herramientas del catálogo de Unity y ejecutando la herramienta de intérprete de código de Python integrada.

1. Ejecute el código siguiente para validar la conexión con el servidor MCP:

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

Crea tu agente

1. Desarrollar a partir del código anterior para definir un agente básico de turno único que utiliza herramientas. Guarde el código del agente localmente como un archivo denominado mcp_agent.py:

    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 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 = 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)
   
            # 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)
   
   

Desplegar el agente

Cuando esté listo para implementar un agente que se conecte a servidores MCP administrados, consulte Implementación de un agente para aplicaciones de IA generativas (Servicio de modelos).

Especifique todos los recursos que necesita el agente en el momento del registro. Consulte Autenticación para recursos de Databricks.

Por ejemplo, si su agente utiliza las direcciones URL del servidor MCP que se enumeran a continuación, debe especificar todos los índices de búsqueda vectorial en los esquemas prod.customer_support y prod.billing. También debe especificar todas las funciones del catálogo de Unity en prod.billing:

• 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

Si el agente se conecta a servidores MCP en Databricks para detectar y ejecutar herramientas, registre los recursos necesarios para estos servidores MCP con el agente. Databricks recomienda instalar el databricks-mcp paquete pyPI para simplificar este proceso.

En concreto, si usa servidores MCP administrados, puede usar databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) para recuperar los recursos necesarios para el servidor MCP administrado. Si el agente consulta un servidor MCP personalizado hospedado en una aplicación de Databricks, puede configurar la autorización incluyendo explícitamente el servidor como un recurso al registrar el modelo.

Por ejemplo, para implementar el agente definido anteriormente, ejecute el código siguiente, suponiendo que guardó la definición de código del agente en 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

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

Pasos siguientes

• Conecte clientes a servidores MCP como Cursor y Claude Desktop.
• Metadatos para servidores MCP administrados de Databricks