メインコンテンツまでスキップ
に最終更新

PythonエージェントでMCPツールを使用する

MCPサーバーに接続し、そのツールを検出して、それらを使用するPythonエージェントを実行します。Databricks提供のMCP、登録済みのMCP、またはDatabricks Apps上のサーバーのURLを使用します。

Claude Code、Codex、またはその他のコーディングエージェントからMCPを使用するには、サポートされているコーディングエージェントでクライアントを選択します。その他のアシスタントおよび MCP クライアントについては、その他の MCP クライアントを参照してください。

Agent Bricks CLI プロジェクトの場合は、CLI を使用して MCP ツールを追加します。以下の例は、独自の Python コードから接続する方法を示しています。

前提条件​

  • お使いのコンピュータ上の Python 3.12。
  • お使いのサーバーのMCP 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の選択と接続要件については、認証とネットワークアクセスを参照してください。

ステップ 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を検索するには、次を実行します。

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 のワークスペースへのサインインを使用します。

この例では、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

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の場合:

レガシーワークスペースサーバーまたはDatabricks Appsでホストされているサーバーの場合は、基盤となるリソースまたはアプリへのアクセス権を付与します。エージェント認証を参照してください。

その他のリソース​

このページの見出し