PythonエージェントからMCPツールを呼び出す

LangGraph、OpenAI Agents SDK、または OpenAI クライアントで構築されたカスタム Python エージェントで MCP ツールを使用します。これらの例は、サインイン、ツールの検索、および独自の Databricks 資格情報を使用してエージェントをローカルでランする方法を示しています。

サインインしたユーザーに代わってツールを呼び出すアプリについては、「アプリをMCPツールに接続する」を参照してください。

他の接続方法については、サポートされているコーディングエージェント、 その他のMCPクライアント、または Agent Bricks CLIでMCPツールを追加する方法をご覧ください。

前提条件

• お使いのコンピュータ上の Python 3.12。
• Databricksが提供するMCP、登録済みMCP、またはDatabricks Apps上のサーバーのURL。誰かがMCPをあなたと共有した場合は、その人があなたにアクセス権を付与する必要があります。ユーザーごとのOAuthを使用している場合は、プロバイダーのサインインを完了します。Databricks Appsのサーバーを使用するには、アプリに対する CAN USE 権限が必要です。
• ツール呼び出しをサポートする Databricks モデルEndpointへのアクセス。この例ではdatabricks-claude-sonnet-4-5を使用します。ワークスペースで使用可能なEndpointに置き換えます。

system.ai.dbsql、system.ai.sandbox、または system.ai.web_search を使用するには、アカウント管理者がアカウントコンソールの [ プレビュー ] ページから [ Unity Gateway ] ベータ版を有効にする必要があります。アカウントプレビューの管理を参照してください。

ID の選択については、「サポートされている送信認証」を参照してください。接続要件については、「MCP ネットワーキング」を参照してください。

ステップ 1：インストールとサインイン

1. Databricks CLIをインストールしていない場合は、macOSまたはLinuxで次のコマンドを実行します。

   Bash

   curl -fsSL https://raw.githubusercontent.com/databricks/setup-cli/main/install.sh | sh

   Windows またはその他のインストール方法については、「Install the Databricks CLI」を参照してください。

2. ワークスペースにサインインします。

   Bash

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

3. Python ライブラリをインストールします。

   Bash

   pip install --upgrade databricks-mcp databricks-sdk "mcp>=1.24,<2"

この例では、以下のエージェント フレームワークと互換性のあるMCP Python 1.xを使用しています。

ステップ 2: サーバーへの接続

次のコードを mcp_agent.py として保存します。<mcp-server-url> をサーバーの URL に置き換えます。

• Databricks提供または登録済みMCP: https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<service>。
• Databricks Apps上のサーバー： 概要ページからアプリのURLをコピーし、/mcpを追加します。

Python

from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient

workspace_client = WorkspaceClient()
server_url = "<mcp-server-url>"
mcp_client = DatabricksMCPClient(
    server_url=server_url,
    workspace_client=workspace_client,
)

tools = mcp_client.list_tools()
for tool in tools:
    print(tool.name, tool.description, tool.inputSchema, sep="\n")

スクリプトをランします。

Bash

python mcp_agent.py

サーバーのツール、説明、入力スキーマが表示されます。次のステップとして、これらのツールのいずれかがサポートする読み取り専用タスクを選択します。リストが空の場合、または接続に失敗した場合は、MCP 認証を参照してください。

別のサーバーを使用する

• 組み込みの MCP (built-in MCP) については、MCP URL 内に system.ai.github などの完全な名前を使用します。
• ワークスペースMCPサーバーとの既存の統合については、server_urlをそのワークスペースEndpoint URLに置き換えてください。

カタログとスキーマでアクセスできるMCPを検索するには、次を実行します。

Bash

databricks ai-gateway list-mcp-services --parent schemas/system.ai

登録済みのMCPの一覧を表示するには、 system.ai を <catalog>.<schema> に置き換えます。CLI がページネーションを処理します。

ステップ3：これらのツールを使用してエージェントをランする

フレームワークを選択し、そのパッケージをインストールし、Pythonの例を mcp_agent.py に追加します。各例では、同じ server_url とステップ 2 のワークスペースへのサインインを使用します。

LangGraph

この例では、MCP コンテンツ ブロックを convert_to_openai_messages を使用してモデルのチャット メッセージ形式に変換します。

Bash

pip install --upgrade databricks-langchain langgraph

Python

import asyncio
from databricks_langchain import (
    ChatDatabricks,
    DatabricksMCPServer,
    DatabricksMultiServerMCPClient,
)
from langchain_core.messages import convert_to_openai_messages
from langgraph.prebuilt import create_react_agent

async def main():
    client = DatabricksMultiServerMCPClient([
        DatabricksMCPServer(
            name="my-mcp-server",
            url=server_url,
            workspace_client=workspace_client,
        ),
    ])
    agent = create_react_agent(
        ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
        tools=await client.get_tools(),
        prompt=lambda state: convert_to_openai_messages(state["messages"]),
    )
    task = input("Ask the agent to use a tool: ")
    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": task}],
    })
    for message in result["messages"]:
        print(message)

asyncio.run(main())

デプロイメント ノートブック (オプション)

Model Serving のデプロイメントでは、このノートブックを適宜変更してサーバー URL を使用します。

LangGraph MCP tool-calling agent

ノートブックを新しいタブで開くOpen in Databricks

OpenAI Agents SDK

Bash

pip install --upgrade databricks-openai openai-agents

Python

import asyncio
from agents import Agent, Runner, set_default_openai_api, set_default_openai_client
from agents.mcp import MCPServerStreamableHttpParams
from agents.tracing import set_trace_processors
from databricks_openai import AsyncDatabricksOpenAI
from databricks_openai.agents.mcp_server import McpServer

set_default_openai_client(AsyncDatabricksOpenAI())
set_default_openai_api("chat_completions")
set_trace_processors([])

async def main():
    async with McpServer(
        name="my-mcp-server",
        params=MCPServerStreamableHttpParams(url=server_url),
        workspace_client=workspace_client,
    ) as server:
        agent = Agent(
            name="Tool-using agent",
            instructions="Use the available tools to answer the user's question.",
            model="databricks-claude-sonnet-4-5",
            mcp_servers=[server],
        )
        task = input("Ask the agent to use a tool: ")
        result = await Runner.run(agent, task)
        for item in result.new_items:
            print(item.to_input_item())

asyncio.run(main())

デプロイメント ノートブック (オプション)

Model Serving のデプロイメントでは、このノートブックを適宜変更してサーバー URL を使用します。

Agents SDK MCP tool-calling agent

ノートブックを新しいタブで開くOpen in Databricks

OpenAI client

Bash

pip install --upgrade databricks-openai

この例では、OpenAI互換クライアントを使用してツール呼び出しループを明示的に実行します。

Python

import json
from databricks_openai import DatabricksOpenAI, McpServerToolkit

toolkit = McpServerToolkit(url=server_url, workspace_client=workspace_client)
tools_by_name = {tool.name: tool for tool in toolkit.get_tools()}
model_client = DatabricksOpenAI()
messages = [{"role": "user", "content": input("Ask the agent to use a tool: ")}]

for _ in range(10):
    response = model_client.chat.completions.create(
        model="databricks-claude-sonnet-4-5",
        messages=messages,
        tools=[tool.spec for tool in tools_by_name.values()],
    )
    message = response.choices[0].message
    messages.append(message.model_dump(exclude_none=True))
    if not message.tool_calls:
        print(message.content)
        break
    for call in message.tool_calls:
        try:
            tool = tools_by_name.get(call.function.name)
            if tool is None:
                raise ValueError(f"Unknown tool: {call.function.name}")
            arguments = json.loads(call.function.arguments or "{}")
            if not isinstance(arguments, dict):
                raise ValueError("Tool arguments must be a JSON object.")
            output = tool.execute(**arguments)
        except Exception as error:
            output = json.dumps({"error": str(error)})
        print(call.function.name, output)
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": str(output),
        })
else:
    raise RuntimeError("The agent reached the tool-calling limit.")

デプロイメント ノートブック (オプション)

Model Serving のデプロイメントでは、このノートブックを適宜変更してサーバー URL を使用します。

OpenAI MCP tool-calling agent

ノートブックを新しいタブで開くOpen in Databricks

python mcp_agent.pyをもう一度ランします。プロンプトが表示されたら、必要な入力を含めて、選択した読み取り専用タスクを要求します。たとえば、サーバーにチケット検索ツールがある場合は、特定のプロジェクトでオープンなチケットを検索するように依頼します。

出力された会話から、ツール呼び出し、その結果、およびエージェントの回答を確認してください。ツール呼び出しを伴わない回答では、MCP サーバーが使用されたことは確認されません。

トラブルシューティングのためにツールを直接呼び出す

ステップ 2 で出力されたツール名と入力スキーマを使用します。この例では、サーバーのツールで動作するように名前と引数の入力を求めます。ステップ 2 の接続コードの後にランします。

Python

import json

tool_name = input("Read-only tool name: ")
arguments = json.loads(input("Tool arguments as a JSON object: "))
result = mcp_client.call_tool(tool_name, arguments)
print(result)

結果にツールエラーがなく、期待したデータが含まれていることを確認します。

ツールを呼び出す前に、list_tools()を使用してツール名と入力スキーマを見つけてください。結果のフォーマットはツールによって異なります：

• structuredContentが存在する場合は、その構造化結果を直接使用してください。ツールは、outputSchemaを使用してその形状を記述できます。
• それ以外の場合は、content ブロックを検査します。ツールが JSON を返す場合にのみ、テキスト ブロックを JSON として解析します。MCP は、プレーン テキストやその他のコンテンツ タイプもサポートしています。
• 特定の出力フィールドに依存する前に、isError を確認し、応答のサンプルを検査してください。

準備ができたらデプロイして共有する

ローカルの例はユーザーとして実行されます。Databricks Apps でエージェントをデプロイする際、使用するアイデンティティを選択します。共有アクセスの場合はアプリの Service Principal、ユーザーごとのアクセスの場合は呼び出し元のユーザーを選択します。

Databricks提供または登録済みのMCPの場合:

• 呼び出し元にアクセス権を付与し、MCP とその親カタログおよびスキーマへのアクセスを許可します。
• エージェントがユーザーに代わって動作する場合は、ユーザーごとのアクセスを構成してください。アプリ独自のIDプロバイダーでユーザーをサインインさせる場合は、アプリをMCPツールに接続する方法を参照してください。
• MCPをガバナンス管理して、ツールの制限、ポリシーの適用、レート制限の設定、および呼び出しの監視を行います。

ワークスペースのMCP EndpointまたはDatabricks Appsでホストされているサーバーの場合は、基になるリソースまたはアプリへのアクセスを許可します。エージェント認証を参照してください。

その他のリソース

• Model Serving でのエージェントのデプロイ。
• Model Servingエージェントの認証を構成します。