カスタムエージェントで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 サーバーをホストする

サーバータイプ	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のプロファイル名は 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 ツールをランするには、ワークスペースで Serverless コンピュートが有効になっている必要があります。

認証

エージェントのランする場所と一致する認証方法を選択します。外部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のService Principalの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 Agent 上の CAN_RUN や、AI Search インデックス上の SELECT などがあります。

Databricks Apps (recommended)
Model Serving

エージェントが使用する各リソース（すべての MCP サーバーの背後にあるリソースを含む）を databricks.yml 内の resources.apps.<app>.resources で宣言し、バンドルをデプロイしてアプリの Databricks Service Principal にアクセス権を付与します。たとえば、マネージド 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 アプリケーション向けエージェントのデプロイ (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,
)

次のステップ

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

サーバーURLを取得する​

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

ツールの接続と一覧表示​

認証​

エージェントの構築​

ノートブックの例​

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

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

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

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

次のステップ​