使用 Databricks 受控 MCP 伺服器

Important

這項功能位於測試版 (Beta) 中。工作區管理員可以從 「預覽 」頁面控制對此功能的存取。請參閱管理 Azure Databricks 預覽。

Databricks 受控 MCP 伺服器是現成的伺服器，可將您的 AI 代理程式連線到儲存在 Unity 目錄、Databricks 向量搜尋索引、Genie 空間和自訂函式中的資料。

可用的受管理伺服器

若要檢視 MCP 伺服器及其端點 URL，請移至您的工作區>代理程式>MCP 伺服器：

代理程式 MCP 伺服器索引標籤

Databricks 提供下列現成可用的 MCP 伺服器。

MCP 伺服器	Description	URL 模式
向量搜尋	查詢向量搜尋索引以尋找相關文件。僅支援具有 Databricks 受控內嵌的索引。	`https://<workspace-hostname>/api/2.0/mcp/vector-search/{catalog}/{schema}`
Genie 空間	查詢 Genie 空間以使用自然語言分析結構化資料。將外部 AI 助手（例如 Claude 或 ChatGPT）連接到您的數據進行只讀操作時，請使用 Genie MCP 服務器。便條： Genie 的受管理 MCP 伺服器會叫用 Genie 做為 MCP 工具，這表示叫用 Genie API 時不會傳遞歷程記錄。或者，您可以在多代理程序系統中使用 Genie。	`https://<workspace-hostname>/api/2.0/mcp/genie/{genie_space_id}`
Unity 目錄函式	使用 Unity 目錄函式來執行預先定義的 SQL 查詢。	`https://<workspace-hostname>/api/2.0/mcp/functions/{catalog}/{schema}`
DBSQL	運行 AI 生成的 SQL。使用 DBSQL MCP 伺服器使用 AI 編碼工具（Claude Code、Cursor、Codex 等）編寫資料管道。對於唯讀資料檢索和聊天機器人整合，請改用 Genie。	`https://<workspace-hostname>/api/2.0/mcp/sql`

範例案例

考慮一個幫助客戶支援的客製化代理。您可以將它連線到多個受管理的 MCP 伺服器：

向量搜尋： https://<workspace-hostname>/api/2.0/mcp/vector-search/prod/customer_support
- 搜尋支援票證和文件
Genie 空間： https://<workspace-hostname>/api/2.0/mcp/genie/{billing_space_id}
- 查詢帳單資料和客戶資訊
UC 函式： https://<workspace-hostname>/api/2.0/mcp/functions/prod/billing
- 執行自訂函數以進行帳戶查詢和更新

這使您的客服專員可以存取非結構化資料（支援票證）、結構化資料（帳單表）和自訂業務邏輯。

範例筆記本：使用 Databricks MCP 伺服器建置代理程式

下列筆記本示範如何使用 Databricks 受控 MCP 伺服器撰寫呼叫 MCP 工具的 LangGraph 和 OpenAI 代理程式。

LangGraph MCP 工具呼叫代理程式

拿筆記本

OpenAI MCP 工具呼叫代理程式

拿筆記本

從本機環境建置代理程式

連線到 Databricks 上的 MCP 伺服器類似於連線到任何其他遠端 MCP 伺服器。您可以使用標準 SDK 連線到伺服器，例如 MCP Python SDK。主要差異在於 Databricks MCP 伺服器預設是安全的，而且需要用戶端指定驗證。

databricks-mcp Python 連結庫可協助簡化自定義代理程式碼中的驗證。

開發代理程式代碼最簡單的方式，就是在本機執行代理程序代碼，並向工作區進行驗證。請遵循下列步驟來建置連線到 Databricks MCP 伺服器的 AI 代理程式。

設定您的環境

使用 OAuth 向工作區進行驗證。在本機終端機中執行下列命令：
```
databricks auth login --host https://<your-workspace-hostname>
```
出現提示時，重新設定設定檔名稱並記下以供日後使用。預設設定檔名稱為 DEFAULT。

請確定您有具有 Python 3.12 或更新版本之本機環境，然後安裝相依性：

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

測試本機環境連線

您的工作區中必須啟用無伺服器計算，才能執行此代碼段。

列出您的 Unity 目錄工具並執行內建的 Python 程式碼解譯器工具，以驗證您與 MCP 伺服器的連線。

執行下列程式碼來驗證您與 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()

建立代理程式

以上述程式碼為基礎，定義使用工具的基本單回合代理程式。將代理程式代碼儲存在本機作為名為 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)

部署代理程式

當您準備好部署連線到受管理 MCP 伺服器的代理程式時，請參閱部署生成式 AI 應用程式的代理程式。

指定代理程式在記錄時所需的所有資源。請參閱 Databricks 資源的驗證

例如，如果您的代理程式使用下列 MCP 伺服器 URL，您必須在 prod.customer_support 和 prod.billing 結構描述中指定所有向量搜尋索引。您也必須在 prod.billing中指定所有 Unity 目錄函式：

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

如果您的代理程式連線到 Databricks 上的 MCP 伺服器以探索和執行工具，請使用代理程式記錄這些 MCP 伺服器所需的資源。 Databricks 建議安裝 databricks-mcp PyPI 套件，以簡化此程式。

特別是，如果您使用受管理 MCP 伺服器，則可用來 databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) 擷取受管理 MCP 伺服器所需的資源。如果您的代理程式查詢裝載在 Databricks 應用程式上的自訂 MCP 伺服器，您可以在記錄模型時明確包含伺服器作為資源，以設定授權。

例如，若要部署上述定義的代理程式，請執行下列程式碼，假設您已將代理程式程式碼定義儲存在 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,
)

後續步驟

將客戶端連接到 Cursor 和 Claude Desktop 等 MCP 服務器。
Databricks 向量搜尋 MCP 伺服器的中繼參數

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-12-09