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

利用可能なMCPサーバーとツールを検索する

エージェントコードを作成する前に、使用可能なサーバーとツールを確認してください。サーバー名、ツール名、または引数の形状を記憶からハードコーディングしないでください。ワークスペースから確認してください。

ワークスペース内のサーバーを参照します。 利用可能なMCPサーバーを確認するには、 AI Gateway > MCPs に移動します。Databricksは、すぐに使用できる組み込みサーバーを提供しています。独自のUnity Catalogデータおよび関数（Genie spaces、ベクトル検索インデックス、Unity Catalog関数）用のマネージドMCPサーバーと、Slack、GitHub、Google Drive、Google Calendar、Gmail、Microsoft 365などのサードパーティSaaSツール用の組み込みsystem.ai MCPサービスです。
MCP サービスをプログラムで一覧表示します。 Unity Catalog REST API を使用して、任意のカタログおよびスキーマ内の MCP サービスを一覧表示します。たとえば、組み込みサービスは次のとおりです。
Bash
```
databricks api get "/api/2.1/unity-catalog/mcp-services?parent=schemas/system.ai&page_size=100"
```
登録したサービスを見つけるには、system.ai を独自の <catalog>.<schema> に置き換えます。page_size は最大 100 に制限されており、さらにサービスが存在する場合は応答に next_page_token が含まれます。スキーマ内のすべてのサービスを列挙するには、応答にトークンが返されなくなるまで page_token=<next_page_token> を指定してリクエストを繰り返します。
Bash
```
token=""
while :; do
  page=$(databricks api get "/api/2.1/unity-catalog/mcp-services?parent=schemas/system.ai&page_size=100&page_token=$token")
  echo "$page"
  token=$(echo "$page" | jq -r '.next_page_token // empty')
  [ -z "$token" ] && break
done
```
コードからサーバーのツールをリスト化します。 DatabricksMCPClient を任意のサーバー URL に向け、list_tools() を呼び出すことで、ツールの接続とリスト化で示されているように、実行時に各ツールの名前、説明、入力スキーマを取得できます。これが、サーバーの正確なツールと引数を把握するための確実な方法です。

環境をセットアップする

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"
```
以下のエージェントフレームワークの例には、独自のSDKが必要です。OpenAI Agents SDKにはopenai-agents databricks-openaiを、LangGraphにはdatabricks-langchain langgraphを追加します。

ツールの接続と一覧表示

サーバー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サービスの場合は、代わりに ai-gateway ユーザー API スコープ (user_api_scopes: [ai-gateway]) を追加し、呼び出し元のユーザーにそのサービスに対する EXECUTE を付与してください。「MCPサービスへの認証」を参照してください。

エージェントの構築

エージェントフレームワークを使用して、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にデプロイするを参照してください。すべてのリソースタイプと権限値については、エージェントの認証を参照してください。

注記

エージェントTemplateから開始します。これには、MLflow AgentServer エントリーポイント（uv run start-app で実行）、get_user_workspace_client() ユーザー代理ヘルパー、および databricks.yml が用意されています。requires-python = ">=3.12,<3.13" でインタープリターをピン留めし、uv.lock をcommitすることで、Databricks Appsのビルドイメージが、一部のエージェント依存関係の事前構築済みwheelを欠く新しいPython（例：3.14）を解決しないようにします。MCPサービスの場合、呼び出し元に帯域外アクセス権も付与してください（ユーザーごとのアクセス（ユーザー代理アクセス）の有効化を参照）。これがないと bundle validate は通過しません。

ログ記録時に必要なすべてのリソースと共にエージェントをログに記録し、その後デプロイします。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サービスを使用してエージェントをツールに接続する」を参照してください。
_metaパラメーターを使用して管理対象サーバーツールの動作を設定するには、「AI Search MCPサーバー」および「Databricks SQL MCPサーバー」を参照してください。
MCP を AI アシスタントやコーディングエージェントに接続して、Claude、Cursor、その他のツールから MCP サーバーを使用します。
エージェントを外部サービスに接続するすべてのアプローチの概要については、「エージェントを外部MCPおよびツールに接続する」を参照してください。

サーバーURLを取得する​

利用可能なMCPサーバーとツールを検索する​

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

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

認証​

エージェントの構築​

ノートブックの例​

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

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

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

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

次のステップ​