Używaj serwerów MCP w Custom Agents

Important

Ta funkcja jest dostępna w publicznej wersji testowej.

Połącz swój kod agenta z dowolnym serwerem MCP w Azure Databricks: serwery zarządzane przez Databricks, zewnętrzne serwery MCP zarejestrowane jako usługi MCP oraz serwery niestandardowe hostowane jako aplikacje Databricks. Wszystkie z nich uwidacznia ten sam interfejs MCP, więc kod agenta jest taki sam. To, co się różni, to adres URL serwera i sposób uwierzytelniania.

Biblioteka języka Python databricks-mcp obsługuje uwierzytelnianie do serwerów Azure Databricks MCP, więc ten sam kod klienta działa ze wszystkimi trzema typami serwerów.

Uzyskiwanie adresu URL serwera

Najpierw skonfiguruj serwer MCP, a następnie użyj jego adresu URL w następujących przykładach:

Typ serwera Wzorzec adresu URL Konfigurowanie
Zarządzany https://<workspace-hostname>/api/2.0/mcp/<service>/<path> Dostępne serwery zarządzane
Zewnętrzna (usługa MCP) https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<mcp-service> Łączenie agentów z narzędziami innych firm za pomocą usług MCP
Custom https://<app-url>/mcp Hostowanie własnego serwera MCP

Konfigurowanie środowiska

  1. Użyj OAuth, aby uwierzytelnić się w swoim obszarze roboczym:

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

  2. Po wyświetleniu monitu wprowadź nazwę profilu i zanotuj ją później. Domyślna nazwa profilu to DEFAULT.

  3. Sprawdź, czy masz środowisko lokalne z 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"

Łączenie i wyświetlanie listy narzędzi

Utwórz element DatabricksMCPClient z adresem URL serwera i wyświetl jego narzędzia. Ten sam klient działa w przypadku zarządzanych, zewnętrznych (usługi MCP) i niestandardowych adresów URL serwera:

from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient

workspace_client = WorkspaceClient(profile="DEFAULT")
host = workspace_client.config.host

# Use a managed, MCP Service, or custom server URL:
mcp_server_url = f"{host}/api/2.0/mcp/functions/system/ai"

mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)
tools = mcp_client.list_tools()
print(f"Available tools: {[t.name for t in tools]}")

Aby wywołać narzędzie bezpośrednio:

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

Uwaga / Notatka

Aby można było uruchamiać narzędzia zarządzane, w obszarze roboczym należy włączyć system.ai.

Authenticate

Wybierz metodę uwierzytelniania zgodną z lokalizacją, w której działa agent. W przypadku zewnętrznej usługi MCP obiekt wywołujący musi również mieć EXECUTE w tej usłudze. Brama AI egzekwuje to uprawnienie przy każdym wywołaniu.

Środowisko lokalne

Uwierzytelnianie w obszarze roboczym przy użyciu protokołu OAuth (zobacz Konfigurowanie środowiska) i przekazanie profilu do klienta:

workspace_client = WorkspaceClient(profile="DEFAULT")
mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)

Główna usługa

Użyj poświadczeń OAuth jednostki usługi. Przekaż wartości bezpośrednio lub pobierz je z wpisów tajnych usługi Azure Databricks (na przykład client_id=dbutils.secrets.get(scope="my-scope", key="client-id")):

workspace_client = WorkspaceClient(
    host="https://<workspace-hostname>",
    client_id="<client-id>",
    client_secret="<client-secret>",
)
mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)

Gdy rejestrujesz agenta, użyj DatabricksApps (niestandardowego) lub odpowiedniego zasobu. Zobacz Przekazywanie automatycznego uwierzytelniania.

W imieniu użytkownika

Użyj ModelServingUserCredentials polecenia , aby agent działał z uprawnieniami użytkownika wywołującego. Zobacz Uwierzytelnianie w imieniu użytkownika:

from databricks.sdk.credentials_provider import ModelServingUserCredentials

workspace_client = WorkspaceClient(credentials_strategy=ModelServingUserCredentials())
mcp_client = DatabricksMCPClient(server_url=mcp_server_url, workspace_client=workspace_client)

Rejestruj model agenta przy użyciu zakresu apps, a w przypadku serwerów zarządzanych uwzględnij odpowiedni zakres OAuth dla każdego serwera. Zobacz Dostępne serwery zarządzane.

Utwórz agenta

Użyj struktury agenta, aby przekształcić narzędzia serwera MCP w agenta. Skieruj framework na adres URL serwera i przekaż uwierzytelnione WorkspaceClient.

Zestaw SDK agentów OpenAI

import asyncio
from agents import Agent, Runner
from databricks.sdk import WorkspaceClient
from databricks_openai.agents import McpServer


async def main():
    workspace_client = WorkspaceClient()
    host = workspace_client.config.host

    async with McpServer(
        url=f"{host}/ai-gateway/mcp-services/main.default.github_mcp",
        name="github-mcp",
        workspace_client=workspace_client,
    ) as mcp_server:
        agent = Agent(
            name="Local agent",
            instructions="You are a helpful assistant with access to external services.",
            model="databricks-claude-sonnet-4-5",
            mcp_servers=[mcp_server],
        )
        result = await Runner.run(agent, "List my open GitHub pull requests.")
        print(result.final_output)


asyncio.run(main())

LangGraph

from databricks.sdk import WorkspaceClient
from databricks_langchain import ChatDatabricks, DatabricksMCPServer, DatabricksMultiServerMCPClient
from langgraph.prebuilt import create_react_agent

workspace_client = WorkspaceClient()
host = workspace_client.config.host

mcp_client = DatabricksMultiServerMCPClient([
    DatabricksMCPServer(
        name="external-service",
        url=f"{host}/ai-gateway/mcp-services/main.default.github_mcp",
        workspace_client=workspace_client,
    ),
])

async with mcp_client:
    tools = await mcp_client.get_tools()
    agent = create_react_agent(
        ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
        tools=tools,
    )
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "List my open GitHub pull requests."}]}
    )
    print(result["messages"][-1].content)

Zestaw SDK Python MCP

Utwórz niezależnego od platformy agenta, który odnajduje i wywołuje narzędzia na co najmniej jednym serwerze MCP. Zapisz następujące polecenie jako mcp_agent.py. Akceptuje listę adresów URL zarządzanych serwerów, usługi MCP i serwerów niestandardowych:

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 server URLs here — managed, MCP Service, or custom:
MANAGED_MCP_SERVER_URLS = [
    f"{host}/api/2.0/mcp/functions/system/ai",
]
# Custom MCP servers hosted on Databricks apps, or MCP Service endpoints:
CUSTOM_MCP_SERVER_URLS = []


# 2) HELPER: convert between ResponsesAgent "message dict" and ChatCompletions format
def _to_chat_messages(msg: dict[str, Any]) -> List[dict]:
    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:
        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):
        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)

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

        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:
            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}"

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

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

        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)

Przykładowe notatniki

Poniższe notatniki pokazują, jak zbudować agentów LangGraph i OpenAI, którzy wywołują narzędzia MCP na zarządzanych, zewnętrznych i niestandardowych serwerach MCP:

Agent wywołujący narzędzia LangGraph MCP

Pobierz laptopa

Agent wywołujący narzędzia OpenAI MCP

Pobierz laptopa

Agent sdk MCP tool-calling agent

Pobierz laptopa

Wdrażanie agenta

Azure Databricks zaleca wdrażanie agentów w usłudze Databricks Apps, co pozwala w pełni zarządzać kodem agenta, konfiguracją serwera i przechowywaniem wersji opartych na usłudze Git. Alternatywnie należy wdrożyć w usłudze Model Serving.

Niezależnie od tego, którą wybierzesz, przyznaj agentowi dostęp do wszystkich zasobów, od których zależą jego serwery MCP. Na przykład CAN_RUN na Genie Agent lub SELECT na indeksie AI Search.

Zadeklaruj każdy zasób używany przez twojego agenta, włącznie z zasobami stojącymi za każdym serwerem MCP, w podsekcji resources.apps.<app>.resources , databricks.ymla następnie wdroż pakiet, aby przyznać dostęp do głównej usługi aplikacji. Na przykład, w przypadku agenta korzystającego z zarządzanych serwerów Genie i AI Search:

resources:
  apps:
    my_agent_app:
      name: 'my-agent-app'
      source_code_path: ./
      resources:
        - name: 'llm'
          serving_endpoint:
            name: 'databricks-claude-sonnet-4-5'
            permission: 'CAN_QUERY'
        - name: 'genie_space'
          genie_space:
            space_id: '<genie-space-id>'
            permission: 'CAN_RUN'
        - name: 'vector_index'
          uc_securable:
            securable_full_name: '<catalog>.<schema>.<index-name>'
            securable_type: 'TABLE'
            permission: 'SELECT'

databricks bundle deploy
databricks bundle run my_agent_app

Aby uzyskać pełny przepływ pracy tworzenia i wdrażania, zobacz Tworzenie agenta i wdrażanie go w usłudze Databricks Apps. Aby uzyskać informacje o wszystkich typach zasobów i wartościach uprawnień, zobacz Uwierzytelnianie agentów.

Modelowa obsługa

Zarejestruj agenta ze wszystkimi zasobami, których potrzebuje, podczas rejestrowania, a następnie wdroż. Zobacz Wdrażanie agenta dla aplikacji sztucznej inteligencji (obsługa modelu) i uwierzytelnianie zasobów usługi Databricks. Azure Databricks zaleca pakiet databricks-mcp do uzyskiwania zasobów serwera MCP:

  • W przypadku zarządzanych serwerów MCP użyj polecenia databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>) , aby pobrać zasoby wymagane przez serwer.
  • W przypadku niestandardowego serwera MCP hostowanego w aplikacji usługi Databricks dołącz aplikację jako zasób podczas rejestrowania modelu.

Na przykład w celu wdrożenia agenta zdefiniowanego 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

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

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",
]

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 to include a custom MCP server hosted on a Databricks app:
    # 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,
    )

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

Następne kroki

  • Last updated on