エージェントでMCPサーバーを使用する

備考

プレビュー

Databricks上の任意のMCPサーバーにエージェントコードを接続します：Databricksが管理するサーバー、MCPサービスとして登録された外部MCPサーバー、およびDatabricksアプリとしてホストされるカスタムサーバー。すべて同じMCPインターフェースを公開しているため、エージェントコードは同じです。異なるのは、**サーバーURL**と**認証方法**です。

「databricks-mcp」Pythonライブラリは、Databricks MCPサーバーへの認証を処理するため、同じクライアントコードが3種類のすべてのサーバータイプで機能します。

サーバーURLを取得します

まずMCPサーバーをセットアップし、そのURLを以下の例で使用します。

サーバータイプ	URLパターン	セットアップ
マネージド	`https://<workspace-hostname>/api/2.0/mcp/<service>/<path>`	利用可能なマネージドサーバー
外部（MCPサービス）	`https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<mcp-service>`	MCPサービスを使用してエージェントをサードパーティツールに接続します
カスタム	`https://<app-url>/mcp`	独自のMCPサーバーをホストする

環境をセットアップする

OAuth を使用してワークスペースへの認証を行います：
Bash
```
databricks auth login --host https://<workspace-hostname>
```
プロンプトが表示されたら、プロファイル名を入力し、後で使用するためにメモしておきます。デフォルトのプロファイル名はDEFAULTです。
Python 3.12以上を搭載したローカル環境があることを確認し、依存関係をインストールします:
Bash
```
pip install -U "mcp>=1.9" "databricks-sdk[openai]" "mlflow>=3.1.0" "databricks-agents>=1.0.0" "databricks-mcp"
```

ツールを接続および一覧表示

サーバーURLを使用してDatabricksMCPClientを作成し、そのツールをリスト表示します。同じクライアントで、マネージド、外部（MCPサービス）、およびカスタムサーバーURLが動作します：

Python
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]}")

ツールを直接呼び出すには:

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

注記

マネージドsystem.aiツールを実行するには、ワークスペースでサーバレスコンピュートを有効にする必要があります。

認証

エージェントが実行される場所に一致する認証方法を選択します。外部 MCP サービスの場合、呼び出し元はそのサービスに対しても EXECUTE を持っている必要があります。AI Gateway は、この権限をすべての呼び出しに適用します。

Local environment
Service principal
On-behalf-of-user

OAuthを使用してワークスペースに認証し（「環境をセットアップする」を参照）、プロファイルをクライアントに渡します。

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

Databricks サービスプリンシパルの OAuth 認証情報を使用します。値を直接渡すか、Databricks シークレットから取得します (例: client_id=dbutils.secrets.get(scope="my-scope", key="client-id"))：

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

エージェントをログに記録する際は、DatabricksApps (カスタム) または関連するリソースをリソースとして使用します。自動認証パススルーを参照してください。

エージェントが呼び出し元ユーザーの権限で動作するように、ModelServingUserCredentials を使用します。ユーザー代理認証をご覧ください。

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

エージェントモデルを apps スコープを使用してログに記録し、マネージドサーバーの場合は、各サーバーに対応する OAuth スコープを含めます。利用可能なマネージドサーバーを参照してください。

エージェントを構築する

エージェントフレームワークを使用して、MCPサーバーのツールをエージェントに変換します。フレームワークをサーバーURLに向け、認証されたWorkspaceClientを渡します。

OpenAI Agents SDK
LangGraph
MCP Python SDK

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

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

1つまたは複数のMCPサーバー間でツールを検出して呼び出すフレームワークに依存しないエージェントを構築します。以下をmcp_agent.pyとして保存します。マネージド、MCPサービス、カスタムサーバーのURLのリストを受け入れます。

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

ノートブックの例

以下のノートブックでは、マネージド、外部、およびカスタムの MCP サーバー全体で MCP ツールを呼び出す LangGraph および OpenAI エージェントの構築方法を示します。

エージェントをデプロイする

Databricksは、エージェントをDatabricks Appsにデプロイすることをお勧めします。これにより、エージェントコード、サーバー構成、およびGitベースのバージョン管理をフルマネージドで実行できます。または、Model Servingにデプロイします。

いずれを選択した場合でも、その MCP サーバーが依存するすべてのリソースへのアクセスをエージェントに許可します。たとえば、Genie Space 上の CAN_RUN や AI 検索インデックス上の SELECT などです。

Databricks Apps (recommended)
Model Serving

エージェントが使用する各リソース (すべてのMCPサーバーの背後にあるリソースを含む) を databricks.yml の resources.apps.<app>.resources で宣言し、バンドルをデプロイして、アプリのDatabricks サービスプリンシパルにアクセスを付与します。例えば、マネージド Genie および AI Search サーバーを使用するエージェントの場合：

YAML
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'

Bash
databricks bundle deploy
databricks bundle run my_agent_app

作成とデプロイの全体のワークフローについては、AIエージェントを作成してDatabricks Appsにデプロイするを参照してください。すべてのリソースタイプとアクセス許可の値については、AI エージェントの認証を参照してください。

エージェントをロギング時に必要なすべてのリソースと共に記録し、デプロイしてください。「生成AIアプリケーションのためのエージェントのデプロイ (Model Serving)」および「Databricksリソースの認証」を参照してください。Databricksは、MCPサーバーリソースを導出するためにdatabricks-mcpパッケージを推奨します。

マネージド MCP サーバーの場合、サーバーに必要なリソースを取得するには、databricks_mcp.DatabricksMCPClient().get_databricks_resources(<server_url>)を使用します。
Databricks アプリでホストされているカスタム MCP サーバーの場合、モデルをログに記録するときにアプリをリソースとして含めます。

例として、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

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

次のステップ

エージェントを MCP サービスでサードパーティ製ツールに接続して、Unity Catalog で外部 MCP サーバーを管理します。
Databricksで管理される MCP サーバーのメタパラメーターは、_metaパラメーターで管理サーバーのツール動作を構成します。
Claude、Cursor、およびその他のツールからMCPサーバーを使用するには、「MCPをAIアシスタントとコーディングエージェントに接続します」。
エージェントを外部サービスに接続するすべての手法の概要については、「エージェントをツールに接続する」を参照してください。

サーバーURLを取得します​

環境をセットアップする​

ツールを接続および一覧表示​

認証​

エージェントを構築する​

ノートブックの例​

LangGraph MCP ツール呼び出しエージェント

OpenAI MCPツール呼び出しエージェント

エージェント SDK MCP ツール呼び出しエージェント

エージェントをデプロイする​

次のステップ​