Korzystanie z zarządzanych serwerów MCP usługi Databricks

Zarządzane przez Databricks serwery MCP to gotowe do użycia serwery, które łączą agentów sztucznej inteligencji z danymi przechowywanymi w katalogu Unity, indeksami wyszukiwania wektorowego Databricks, przestrzeniami Genie i funkcjami niestandardowymi.

Dostępne serwery zarządzane

Aby wyświetlić serwery MCP i ich adresy URL punktów końcowych, przejdź do obszaru roboczego >Agenci>serwery MCP:

Databricks udostępnia następujące serwery MCP, które działają od razu po uruchomieniu.

Przykładowy scenariusz

Rozważ użycie agenta niestandardowego, który pomaga w obsłudze klienta. Można połączyć go z wieloma zarządzanymi serwerami MCP:

• Wyszukiwanie wektorowe: https://<workspace-hostname>/api/2.0/mcp/vector-search/prod/customer_support
  • Szukaj zgłoszeń pomocy technicznej i dokumentacji
• Przestrzeń Genie: https://<workspace-hostname>/api/2.0/mcp/genie/{billing_space_id}
  • Wykonywanie zapytań dotyczących danych rozliczeniowych i informacji o klientach
• Funkcje UC: https://<workspace-hostname>/api/2.0/mcp/functions/prod/billing
  • Uruchamianie funkcji niestandardowych na potrzeby wyszukiwania i aktualizacji kont

Zapewnia to agentowi dostęp do danych bez struktury (biletów pomocy technicznej), danych ustrukturyzowanych (tabel rozliczeniowych) i niestandardowej logiki biznesowej.

Przykładowe notesy: tworzenie agenta za pomocą serwerów MCP usługi Databricks

W poniższych notatnikach pokazano, jak tworzyć agentów LangGraph i OpenAI, którzy wywołują narzędzia MCP na zarządzanych przez Databricks serwerach MCP.

Agent wywołujący narzędzia LangGraph MCP

Pobierz laptopa

Agent wywołujący narzędzia OpenAI MCP

Pobierz laptopa

Tworzenie agenta ze środowiska lokalnego

Nawiązywanie połączenia z serwerem MCP w usłudze Databricks jest podobne do nawiązywania połączenia z innym zdalnym serwerem MCP. Możesz nawiązać połączenie z serwerem przy użyciu standardowych zestawów SDK, takich jak zestaw SDK języka Python MCP. Główną różnicą jest to, że serwery MCP usługi Databricks są domyślnie bezpieczne i wymagają od klientów określenia uwierzytelniania.

Biblioteka języka Python databricks-mcp ułatwia uwierzytelnianie w niestandardowym kodzie agenta.

Najprostszym sposobem opracowywania kodu agenta jest uruchamianie go lokalnie i uwierzytelnianie w obszarze roboczym. Wykonaj następujące kroki, aby utworzyć agenta sztucznej inteligencji, który nawiązuje połączenie z serwerem MCP usługi Databricks.

Konfigurowanie środowiska

1. Użyj protokołu OAuth, aby uwierzytelnić się w obszarze roboczym. Uruchom następujące polecenie w lokalnym terminalu:

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

2. Po wyświetleniu monitu ponownie wprowadź nazwę profilu i zanotuj ją do późniejszego użycia. Domyślna nazwa profilu to DEFAULT.

3. Upewnij się, że masz środowisko lokalne z językiem Python 3.12 lub nowszym, a następnie zainstaluj zależności:

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

Testowanie połączenia środowiska lokalnego

Aby można było uruchomić ten fragment kodu, w obszarze roboczym należy włączyć bezserwerowe obliczenia.

Zweryfikuj połączenie z serwerem MCP, wyświetlając listę narzędzi Unity Catalog i uruchamiając wbudowane narzędzie interpretera kodu Python.

1. Uruchom następujący kod, aby zweryfikować połączenie z serwerem 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()

Tworzenie agenta

1. Rozszerz powyższy kod, aby zdefiniować podstawowego agenta jednokrotnego użycia, korzystającego z narzędzi. Zapisz kod agenta lokalnie jako plik o nazwie 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
   
    # 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 = 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)
   
   

Wdrażanie agenta

Gdy wszystko będzie gotowe do wdrożenia agenta łączącego się z zarządzanymi serwerami MCP, zobacz Wdrażanie agenta na potrzeby generowania aplikacji sztucznej inteligencji.

Określ wszystkie zasoby wymagane przez agenta w czasie logowania. Zobacz Uwierzytelnianie zasobów usługi Databricks

Jeśli na przykład agent używa poniżej wymienionych adresów URL serwera MCP, należy określić wszystkie indeksy wyszukiwania wektorów w schematach prod.customer_support i prod.billing. Należy również określić wszystkie funkcje Unity Catalog w 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

Jeśli agent łączy się z serwerami MCP w usłudze Databricks w celu odnajdywania i uruchamiania narzędzi, zarejestruj zasoby wymagane przez te serwery MCP za pomocą agenta. Usługa Databricks zaleca zainstalowanie databricks-mcp pakietu PyPI w celu uproszczenia tego procesu.

W szczególności, jeśli używasz zarządzanych serwerów MCP, możesz użyć databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) do pobierania zasobów wymaganych przez zarządzany serwer MCP. Jeśli agent wysyła zapytanie do niestandardowego serwera MCP hostowanego w aplikacji usługi Databricks, możesz skonfigurować autoryzację, jawnie dołączając serwer jako zasób podczas rejestrowania modelu.

Aby na przykład wdrożyć agenta zdefiniowanego powyżej, uruchom następujący kod, przy założeniu, że zapisano definicję kodu agenta w pliku 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,
)

Dalsze kroki

• Łączenie klientów z serwerami MCP , takimi jak Cursor i Claude Desktop.
• Meta parametry dla serwerów wyszukiwania wektorów Databricks MCP