アプリをMCPツールに接続する
system.aiスキーマのMCPおよび登録するMCPを含む、Model Context Protocol (MCP) サーバー上に独自のアプリを構築します。アプリは、サインインしたユーザーとしてツールを呼び出します。Unity Gateway ガバナンスは、各ユーザーが呼び出すことができるMCPとツールを制御します。
アプリを Databricks Appsでランする場合は、ユーザー認可を使用してください。アプリが外部IDプロバイダーを使用してユーザーをサインインさせる場合、OAuthトークンフェデレーションを使用してDatabricksトークンを取得してください。このガイドは、複数のユーザーにサービスを提供するアプリを対象としています。独自の資格情報を使用してローカルでエージェントを構築およびテストするには、「PythonエージェントからMCPツールを呼び出す」を参照してください。
ユーザーごとの OAuth を使用する MCP の場合、ユーザーは外部プロバイダーアカウントも認証します。以下のステップでは、Databricks トークンの取得とプロバイダーサインインの完了の両方について説明します。
認証とアクセスの設定
コードを追加する前に、ユーザーに MCP へのアクセス権を付与し、アプリの認証を構成してください。
MCPへのユーザーアクセスを許可する
これらの要件は、両方の認証方法に適用されます。
- 各ユーザーを Databricks ワークスペースに追加します。コンシューマーアクセスは、Unity Gateway を介して MCP を呼び出すのに十分です。
- 各ユーザーにMCP上の
EXECUTEを付与し、さらにその親カタログとスキーマ上のUSE CATALOGとUSE SCHEMAを付与します。MCPの共有を参照してください。
アプリの認証を設定
アプリに一致するオプションを選択します:
- Databricks Apps
- External identity provider
- アプリのユーザー認証を有効にします。アプリは、トークン交換やアカウントフェデレーションポリシーを使用せずに、転送されたDatabricksユーザートークンを直接使用します。
- ユーザー認可スコープを追加します:ツール呼び出しには
ai-gateway、検出およびサインインステータスAPIにはunity-catalogです。 - 各ユーザーにアプリの使用権 を付与します。
アプリがユーザーのDatabricks OAuthトークンを既に持っている場合は、それを直接使用してフェデレーション設定とトークン交換をスキップします。トークンには、このガイドのAPIsのai-gatewayとunity-catalogスコープが必要です。それ以外の場合は、次のステップを実行します。
- アカウント管理者に依頼して、アイデンティティプロバイダーのアカウントフェデレーションポリシーを構成してください。ポリシーでは、許容される発行者、対象者、およびユーザーマッピングを定義します。
- IDプロバイダーからサインインしたユーザーのトークンを取得します。最初のコードステップで、このガイドで使用されるAPIsをカバーする
all-apisを持つDatabricks OAuthトークンと交換します。
アプリのサーバーもワークスペースに到達できる必要があります。IPアクセスリストが有効な場合、そのアウトバウンドIPを許可してください。MCPネットワークを参照してください。
system.ai.dbsql、system.ai.sandbox、またはsystem.ai.web_searchを使用するには、アカウント管理者またはアカウントレベルのメタストア管理者が、アカウントコンソールの プレビュー ページから Unity Gateway プレビューを有効にする必要があります。アカウントのプレビューを管理を参照してください。
アプリをセットアップする
これらのステップを通して、トークンとクライアントをサインインしたユーザー固有のものにしてください。
ユーザーのDatabricksトークンを取得します
アプリがユーザーを認証する方法に一致するオプションを選択してください。どちらのオプションでも、残りのステップでdatabricks_tokenが提供されます。
- Databricks Apps
- External identity provider
Databricks Appsは、ユーザーのDatabricksトークンを x-forwarded-access-token リクエストヘッダーで提供します。そのトークンを直接使用します。アプリでトークン交換を実行する必要はありません。
現在のリクエスト内のトークンを読み取ります。たとえば、Flaskのリクエストハンドラーの場合:
from flask import request
databricks_token = request.headers.get("x-forwarded-access-token")
if not databricks_token:
raise ValueError("A Databricks user token is required.")
その他のフレームワークについては、「ユーザー認証資格情報の取得」を参照してください。
ユーザーのDatabricks OAuthトークンを既にお持ちの場合は、それを直接使用してください。そうでない場合は、ユーザーのIDプロバイダトークンをDatabricks OAuthトークンと交換してください。
import requests
response = requests.post(
"https://<workspace-hostname>/oidc/v1/token",
data={
"subject_token": idp_token,
"subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"scope": "all-apis",
},
)
response.raise_for_status()
databricks_token = response.json()["access_token"]
以下のステップで databricks_token を使用します。交換では更新トークンは返されません。応答から expires_in を使用して有効期限を追跡し、必要に応じて有効なIDプロバイダトークンで交換を繰り返してください。トークン交換の要件と制限を参照してください。
MCP の一覧表示
ユーザーがスキーマ内でアクセスできるMCPをリストします。Databricks提供のMCPについては、schemas/system.aiを使用します。登録済みMCPについては、schemas/main.defaultのように登録されているスキーマを使用します。
- Databricks SDK
- Python
SDKをインストールする:
pip install --upgrade databricks-sdk
ユーザーのトークンでクライアントを作成し、MCPを一覧表示します。
from databricks.sdk import WorkspaceClient
workspace_client = WorkspaceClient(
host="https://<workspace-hostname>",
token=databricks_token,
auth_type="pat",
)
for service in workspace_client.ai_gateway.list_mcp_services(parent="schemas/system.ai"):
print(service.name)
SDK設定auth_type="pat"は、アプリの環境認証情報ではなく、提供されたトークンを選択します。提供されたトークンは、依然としてユーザーのOAuthトークンです。各ユーザーに対してWorkspaceClientを作成し、すべての呼び出しでそのユーザーのトークンが使用されるようにします。以下のステップでworkspace_clientを再利用します。SDKはページネーションを自動的に処理します。
ユーザーのトークンを使用して REST API を呼び出し、MCP をリストします:
import requests
url = "https://<workspace-hostname>/api/2.1/unity-catalog/mcp-services"
headers = {"Authorization": f"Bearer {databricks_token}"}
params = {"parent": "schemas/system.ai"}
while True:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
body = response.json()
for service in body.get("mcp_services", []):
print(service["name"])
if not body.get("next_page_token"):
break
params["page_token"] = body["next_page_token"]
MCPサービスのリストを参照してください。
ユーザーのサインインステータスを確認する
system.ai.githubのようなユーザーごとのOAuthを使用するMCPについては、ユーザーが外部プロバイダーにサインインしているかどうかを確認します。
- Databricks SDK
- Python
MCPのユーザー資格情報を取得します:
from databricks.sdk.errors import NotFound
from databricks.sdk.service.catalog import ProvisioningInfoState
try:
credential = workspace_client.ai_gateway.get_mcp_service_user_mapped_credential(
name="mcp-services/system.ai.github"
)
signed_in = (
credential.provisioning_info is not None
and credential.provisioning_info.state == ProvisioningInfoState.ACTIVE
)
except NotFound:
signed_in = False
ユーザーのトークンでREST APIを呼び出します:
import requests
response = requests.get(
"https://<workspace-hostname>/api/2.1/unity-catalog/mcp-services/system.ai.github/user-credentials",
headers={"Authorization": f"Bearer {databricks_token}"},
)
if response.status_code == 404:
signed_in = False
else:
response.raise_for_status()
signed_in = response.json().get("provisioning_info", {}).get("state") == "ACTIVE"
provisioning_info.stateがACTIVEの場合、ユーザーはサインインしています。ツールを呼び出すに移動します。- リクエストが
NOT_FOUNDを返すか、状態がACTIVE以外の場合は、MCPのサインインボタンを表示します。
DatabricksデータおよびツールMCPなど、ユーザーごとのOAuthを使用しないMCPには、このステップは必要ありません。呼び出し元の資格情報の取得を参照してください。
ユーザーをプロバイダーにサインインさせる
ユーザーがサインインボタンをクリックしたら、MCPのサインインページをポップアップで開きます。ユーザーはそのページでサインインし、プロバイダーの同意画面でアクセスを承認します。
const signInUrl = `https://${workspaceHostname}/mcp-service-login?name=${encodeURIComponent(mcpName)}`;
window.open(signInUrl, 'mcp-sign-in', 'popup,width=600,height=700');
サインインページは、アプリにメッセージを返しません。サインインが終了したタイミングを検出するには、ユーザーがアプリに戻ったときにサインインステータスをもう一度確認します。
サインインページは、ユーザーの Databricks ブラウザーセッションを使用します。ユーザーがそのブラウザーでワークスペースにサインインしていない場合、 Databricks は最初にサインインするように要求します。プロバイダークレデンシャルは、ページにサインインしているユーザーに対して保存されるため、そのユーザーはアプリが動作するユーザーと同じである必要があります。
ツールを呼び出す
ユーザーのDatabricksトークンを使用して、MCPのURLhttps://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<service>にMCPリクエストを送信します。
アプリは、このステップで示すようにツールを直接呼び出すか、どのツールを呼び出すかを決定するエージェントにそれらを与えることができます。エージェントをランするには、 ステップ3: これらのツールを使ってエージェントをランするに従い、 MCPの一覧表示でユーザーのトークンを使って作成した workspace_client を渡します。これにより、エージェントはそのユーザーとしてツールを呼び出します。エージェントフレームワークは MCP Python SDK のバージョン 1 を使用するため、そのページにリストされているパッケージをインストールします。
- Databricks SDK
- Python
パッケージをインストールします:
pip install --upgrade databricks-mcp
MCPの一覧表示からユーザーの workspace_client に接続し、ツールを一覧表示して呼び出します:
from databricks_mcp import DatabricksMCPClient
mcp_client = DatabricksMCPClient(
server_url="https://<workspace-hostname>/ai-gateway/mcp-services/system.ai.github",
workspace_client=workspace_client,
)
for tool in mcp_client.list_tools():
print(f"{tool.name}: {tool.description}")
result = mcp_client.call_tool("<tool-name>", arguments={})
print(result)
この例では、MCP Python SDKのバージョン2を使用しています:
pip install --upgrade "mcp>=2"
ユーザーのトークンでセッションを開き、ツールを一覧表示して呼び出します:
import asyncio
import httpx2
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
MCP_URL = "https://<workspace-hostname>/ai-gateway/mcp-services/system.ai.github"
async def main(databricks_token):
http_client = httpx2.AsyncClient(headers={"Authorization": f"Bearer {databricks_token}"})
async with http_client, streamable_http_client(MCP_URL, http_client=http_client) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
result = await session.list_tools()
for tool in result.tools:
print(f"{tool.name}: {tool.description}")
result = await session.call_tool("<tool-name>", arguments={})
print(result)
asyncio.run(main(databricks_token))
ツールを呼び出す前に list_tools() を使用してツール名と入力スキーマを発見し、各結果の isError を確認します。結果の読み方については、ツールを直接呼び出してトラブルシューティングするを参照してください。ツールリストが空であるか、接続が失敗した場合は、MCP認証を参照してください。