> ## Documentation Index > Fetch the complete documentation index at: https://docs.qoder.com/llms.txt > Use this file to discover all available pages before exploring further. # MCP 統合 MCP（Model Context Protocol）は AI エージェントが外部ツールを呼び出すためのオープンプロトコルです。Python SDK には MCP クライアント機能が組み込まれており、ホストアプリケーションは「どのような MCP サーバーがあるか」を記述するだけで、SDK が接続、ツール発見、メッセージルーティング、OAuth、状態同期などを自動的に処理します。
## アーキテクチャ概要 ``` ┌────────────────────────────────────────────────────────────┐ │ Python app (SDK Host) │ │ │ │ ┌──────────────────────────┐ │ │ │ create_sdk_mcp_server(...) │ ← in-process tools │ │ │ + @tool(...) │ no extra process │ │ └──────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────────────────┐ │ │ │ query({mcp_servers}) │── stdio ─▶ qodercli child │ │ │ / QoderSDKClient(...) │ │ │ └──────────────────────────┘ │ │ │ │ │ ├── stdio ──▶ MCP server (process) │ ├── sse ──▶ MCP server (HTTP/SSE) │ └── http ──▶ MCP server (Streamable HTTP) └────────────────────────────────────────────────────────────┘ ``` * **インプロセス**：ツールは単なる Python の async 関数で、自分のプロセス内で実行されます。`create_sdk_mcp_server` で生成されたものは SDK の control channel を通じて CLI と通信し、別途サブプロセスを起動することはありません。 * **外部サーバー**：設定でサブプロセスまたはリモート URL を宣言し、CLI が接続、発見、呼び出しを担当します。
## 3 つの接続方式 | 方式 | 設定項目 type | プロセス境界 | 適用シナリオ | | -------------- | ------------------------------------ | ------ | -------------------------------------------------- | | **インプロセス** | `'sdk'`（`create_sdk_mcp_server` で作成） | 同一プロセス | カスタム業務ツール、ホスト状態への直接アクセスが必要 | | **Stdio** | `'stdio'`（省略可能） | サブプロセス | 既存の MCP ツールパッケージ（`@modelcontextprotocol/server-*`） | | **SSE / HTTP** | `'sse'` / `'http'` | リモート | リモートサービス、SaaS ツール、OAuth が必要なサービス | 3 つの方式は**混在可能**です——同じ `query()` / `QoderSDKClient` 内で異なる種類のサーバーを複数同時に登録できます。 > 💡 `mcp_servers` には `str` / `pathlib.Path` を渡すこともできます。これは JSON 設定ファイルのパスを指し、SDK は `--mcp-config ` として CLI に透過的に渡します。
## インプロセスサーバー（推奨） インプロセスツールは最もシンプルな拡張方法です。通常の `async` 関数を定義し、デコレーターでスキーマを宣言するだけで、エージェントから呼び出し可能になります。`@tool()` / schema / handler の詳細な動作については [tools.md](/ja/cli/sdk/python/tools) を参照してください。本セクションでは MCP サーバーの組み立てに関する部分のみを扱います。
### 30 秒で開始 ```python theme={null} import asyncio from typing import Annotated from qoder_agent_sdk import ( QoderAgentOptions, create_sdk_mcp_server, query, tool, ) @tool("greet", "相手に挨拶します。", {"name": Annotated[str, "相手の名前"]}) async def greet(args): return {"content": [{"type": "text", "text": f"こんにちは、{args['name']}！"}]} server = create_sdk_mcp_server(name="my_tools", tools=[greet]) async def main(): options = QoderAgentOptions( mcp_servers={"my_tools": server}, allowed_tools=["mcp__my_tools__greet"], ) async for msg in query(prompt="用 greet 工具向 Alice 打招呼", options=options): print(msg) asyncio.run(main()) ```
### `@tool()` / `create_sdk_mcp_server()` 完全シグネチャ ```python theme={null} def tool( name: str, description: str, input_schema: type | dict[str, Any], annotations: ToolAnnotations | None = None, ) -> Callable[[Handler], SdkMcpTool[Any]]: ... def create_sdk_mcp_server( name: str, version: str = "1.0.0", tools: list[SdkMcpTool[Any]] | None = None, ) -> McpSdkServerConfig: ... ``` | パラメータ | 説明 | | -------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------- | | `name`（tool） | ツール名。完全修飾名は `mcp____` になります | | `description` | モデル向けの説明。AI がいつ呼び出すかを決定します——**What/When を明確に記述** | | `input_schema` | シンプルな dict / `TypedDict` / 完全な JSON Schema dict の 3 種類の書き方が可能。詳細は [Tools Reference - `input_schema`](/ja/cli/sdk/python/references#input_schema) を参照 | | `annotations` | MCP ツールアノテーション。詳細は下表参照 | | `name`（server） | サーバー名（ツールプレフィックス `mcp____` を決定） | | `version` | デフォルト `'1.0.0'` | | `tools` | `SdkMcpTool` のリスト | 戻り値の `McpSdkServerConfig` は `{"type": "sdk", "name": ..., "instance": ...}` の形式で、そのまま `options.mcp_servers` に渡せます。
#### annotations の実際のサポート状況 以下の 3 つのフィールドは SDK が実際に消費し、`get_mcp_status().mcpServers[i].tools[i].annotations` を通じてホスト側に返されます。 | フィールド | 動作 | ホスト側の読み方 | | ----------------- | ----------------------------------------------------------------------------------------------- | ------------------------- | | `readOnlyHint` | ツールが**読み取り専用**であることを宣言。読み取り専用ツールは並行実行可能（同一バッチ内で互いをブロックしない）。TUI のツール詳細に `[read-only]` バッジが表示されます | `annotations.readOnly` | | `destructiveHint` | ツールが**破壊的操作**を実行することを宣言。TUI のツール詳細に `[destructive]` バッジが表示されます | `annotations.destructive` | | `openWorldHint` | ツールが**外部世界**（Web 検索、サードパーティ API 呼び出しなど）に触れることを宣言。TUI のツール詳細に `[open-world]` バッジが表示されます | `annotations.openWorld` | > ホスト側のフィールド名は **`Hint` サフィックスを除いた**形式です：`readOnlyHint` → `annotations.readOnly` のように。`annotations` オブジェクトには明示的に設定されたフィールドのみが含まれます。 > > ⚠️ **これらのフィールドは auto モードのパーミッション判定には影響しません**。CLI はサーバーが自己宣言した annotation を検証不可能な参考情報として扱い（サーバーは過少・過大に申告できるため）、パーミッションパイプラインには敢えて取り込みません——取り込むとサーバーの自己評価に権威を付与してしまうためです。**特定ツールを確実にブロックしたい場合は `allowed_tools` ホワイトリストまたは hooks を使用してください**——annotation はホスト側での識別（`get_mcp_status`）と TUI 表示のためだけのものです。 `idempotentHint` と `title` は現在 SDK では消費されません——渡してもエラーにはなりませんが、CLI の動作にも影響せず、`get_mcp_status()` の戻り値にも現れません。アプリケーションでこれらの情報が必要な場合は、ホスト側で独自にマッピングを管理してください。 > 💡 **`maxResultSizeChars` について**：Python SDK は `ToolAnnotations(maxResultSizeChars=...)` を介して `anthropic/maxResultSizeChars` をツールの `_meta` に書き込み、CLI はこれに基づいてデフォルトの 50K の戻り値長さ制限を緩和します。このフィールドは Python 側の追加機能です（TS は同名の annotation で公開されており、wire は同一です）。
#### handler の戻り値 ```python theme={null} # 成功 {"content": [{"type": "text", "text": "結果"}]} # 業務上の失敗: 例外を投げずに is_error を使用 {"content": [{"type": "text", "text": "エラー説明"}], "is_error": True} ``` **業務上の失敗には例外をスローせず `is_error: True` を使用してください**。完全な content タイプの説明、Python 側と TS 側の動作の違い（`resource_link` がテキストにフォールバック、トップレベル `_meta` は透過しない、binary embedded resource はスキップ）については [Tools Reference - `CallToolResult`](/ja/cli/sdk/python/references#calltoolresult) を参照してください。 ```python theme={null} @tool( "query_db", "読み取り専用 SQL クエリ。", {"sql": Annotated[str, "SQL クエリ文"]}, annotations=ToolAnnotations(readOnlyHint=True), ) async def query_db(args): sql = args["sql"] if not sql.lstrip().upper().startswith("SELECT"): return { "is_error": True, "content": [{"type": "text", "text": "只允许 SELECT 语句"}], } rows = await db.query(sql) return {"content": [{"type": "text", "text": json.dumps(rows)}]} ```
### Handler のキャンセルシグナル handler は第 2 引数として `ToolInvocationContext` を受け取ることができ、CLI が現在の呼び出しをキャンセルした際に `extra.signal` を介して協調的に終了できます。 ```python theme={null} @tool("watch", "Watch a counter", {"max": int}) async def watch(args, extra): for i in range(args["max"]): if extra.signal.is_set(): return {"content": [{"type": "text", "text": f"aborted at {i}"}]} await asyncio.sleep(0.01) return {"content": [{"type": "text", "text": "done"}]} ``` > ⚠️ **同じサーバー設定を複数の `query()` で再利用しないでください**：各 query は独立した transport をバインドします。再利用しても副作用はありませんが、「query をまたいで状態を共有する」機能は得られません——共有状態は handler クロージャの外のモジュールスコープに配置してください。
## Stdio サーバー サブプロセスの stdin/stdout を通じて MCP サーバーと通信します。NPM の `@modelcontextprotocol/server-*` シリーズはすべて stdio 実装です。 ```python theme={null} class McpStdioServerConfig(TypedDict): type: NotRequired[Literal["stdio"]] # 省略可能。stdio がデフォルト command: str # 実行コマンド args: NotRequired[list[str]] # コマンド引数 env: NotRequired[dict[str, str]] # 環境変数 tools: NotRequired[list[McpServerToolPolicy]] ``` ```python theme={null} options = QoderAgentOptions( mcp_servers={ "fs": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/project"], }, "gh": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-github"], "env": {"GITHUB_TOKEN": os.environ["GITHUB_TOKEN"]}, }, }, ) ``` `command` が到達不能であったり起動に失敗したりしても、query 全体が落ちることはありません——該当サーバーの status は `'connected'` 以外のままになり、他のサーバーには影響しません。
## SSE / HTTP サーバー ```python theme={null} class McpSSEServerConfig(TypedDict): type: Literal["sse"] url: str headers: NotRequired[dict[str, str]] tools: NotRequired[list[McpServerToolPolicy]] class McpHttpServerConfig(TypedDict): type: Literal["http"] # Streamable HTTP url: str headers: NotRequired[dict[str, str]] tools: NotRequired[list[McpServerToolPolicy]] ``` ```python theme={null} options = QoderAgentOptions( mcp_servers={ "analytics": { "type": "http", "url": "https://analytics.example.com/mcp", "headers": {"Authorization": f"Bearer {os.environ['ANALYTICS_TOKEN']}"}, }, }, ) ``` リモート URL が到達不能でも同様に query がハングすることはなく、サーバーステータスは `'connected'` 以外となり、他のサーバーには影響しません。OAuth が必要なリモートサービスについては [OAuth 認証](#oauth-認証) を参照してください。
## ツール命名とホワイトリスト CLI がモデルに MCP ツールを公開する際、統一的にプレフィックスを付加します。 ``` mcp____ ``` 例えばサーバー名 `my_tools`、ツール名 `greet` の場合、モデルが認識するツール名は `mcp__my_tools__greet` です。サーバー名にはハイフンなどの特殊文字を含めることが許されています（`my-tools` → `mcp__my-tools__`）。
### `tools`：モデルに見えるツール集合を制限 モデルに**一部のツールだけを見せたい**場合は `tools` を使います。CLI はリストにない組み込みツールをすべて disallow リストに追加するため、実質的な「ホワイトリスト」セマンティクスとなります。 ```python theme={null} options = QoderAgentOptions( mcp_servers={"my_tools": server}, tools=[ "Read", "Grep", # built-in tools you still want "mcp__my_tools__greet", "mcp__my_tools__search_docs", ], ) ``` > ⚠️ **`tools` を渡さない＝すべて開放**：すべての組み込みツール + 接続済み MCP サーバーのツールがモデルに公開されます。本番環境では明示的に列挙して範囲を絞ることを推奨します。
### `allowed_tools`：事前承認（**可視性ホワイトリストではない**） `allowed_tools` はリストされたツールを「自動許可」ルールに追加します——呼び出し時に**承認プロンプトをスキップ**しますが、リストにないツールは**非表示にはなりません**。低リスクの MCP ツールを承認なしで使えるようにする際によく使われます。 ```python theme={null} options = QoderAgentOptions( mcp_servers={"my_tools": server}, allowed_tools=[ "mcp__my_tools__greet", # pre-approved, no prompt "mcp__my_tools__search_docs", ], ) ``` `allowed_tools` を渡さない場合は事前承認ルールが無いだけで、モデルは依然としてすべてのツールを見て呼び出せます。書き込み操作は通常の `permission_mode` 承認フローに従います。完全なセマンティクスは [Permissions ドキュメント](/ja/cli/sdk/python/permissions#ツール範囲の制御toolsallowed_toolsdisallowed_tools)を参照してください。
### `allowed_mcp_server_names`：プロセス型サーバーのホワイトリスト **プロセス型**（stdio/sse/http）サーバーのみをフィルタリングし、**インプロセスサーバーには影響しません**。`strict_mcp_config=True` と組み合わせることで、CLI がローカルの追加設定を読み込むことを拒否できます。 ```python theme={null} options = QoderAgentOptions( mcp_servers={ "keep": {"command": "..."}, "drop": {"command": "..."}, }, allowed_mcp_server_names=["keep"], # 'drop' はステータスに残るが接続しない strict_mcp_config=True, # settings.json / .mcp.json から MCP サーバーを読み込まない ) ``` > ⚠️ **`allowed_mcp_server_names` を渡さない＝すべて開放**：宣言されたすべてのプロセス型サーバーが接続されます。範囲を絞りたい場合は明示的に列挙してください。インプロセスサーバーは常にこのフィールドの影響を受けません。
## ランタイム管理（QoderSDKClient） `query()` は使い捨てのイテレータで、途中でサーバーや認証を変更することはできません。MCP のランタイム管理を行うには `QoderSDKClient` を使用する必要があります。状態照会、OAuth、サーバーの追加・削除、再接続 / トグルなどがすべて公開メソッドとして提供されています。 > ⚠️ **キャッシュの原則**：MCP サーバー設定 / 認証状態の変更は tools リストを再構築するため、**セッション途中の変更はプロンプトプレフィックスキャッシュを破壊します**。SDK は「状態の照会 + 最初のメッセージ前に認証を完了する」ためのメソッドを提供します。サーバーセット自体は `options.mcp_servers` で起動時に一度設定し、必要に応じて新しい `QoderSDKClient` を作成してください。
### 状態の照会 ```python theme={null} async with QoderSDKClient(options) as client: status = await client.get_mcp_status() # McpStatusResponse を返す: {"mcpServers": [McpServerStatus, ...]} # 各 McpServerStatus には以下が含まれる: # name, status, serverInfo?, error?, config?, scope?, tools? for server in status["mcpServers"]: print(f"{server['name']}: {server['status']}") if server["status"] == "connected": print(" tools:", [t["name"] for t in server.get("tools", [])]) ``` > 💡 MCP ハンドシェイクは CLI が `initialize` を完了した後、最初のユーザーメッセージの前に発生します。`QoderSDKClient.connect()` は initialize の戻りまで待機します。ハンドシェイク IO に数百ミリ秒かかる場合があるため、必要に応じて `get_mcp_status()` を `connected` になるまで自前でポーリングしてください。
### 状態変化の購読 次の 2 つの方式のいずれかを選択できます。 **方式 1（推奨）**：options に `on_mcp_status_change` コールバックを設定すると、状態が変化するたびに 1 回呼び出されます。 ```python theme={null} async def on_status(msg): print(f"{msg['server_name']} -> {msg['status']}") if msg.get("error"): print(" error:", msg["error"]) options = QoderAgentOptions( mcp_servers={...}, on_mcp_status_change=on_status, ) ``` **方式 2**：メッセージストリームを消費し、`system/mcp_status_change` をフィルタリングします。コールバックとメッセージストリームは同じ payload です。コールバックはフィルタリング処理を省略できる便利機能です。
### ランタイムでのサーバー追加・削除 / 再接続 / 起動・停止 | メソッド | 用途 | | ----------------------------------------- | -------------------------------------------------------------- | | `client.set_mcp_servers(servers)` | 全量の desired マッピングで現在の MCP 設定を置換。`{added, removed, errors}` を返す | | `client.reconnect_mcp_server(name)` | 指定サーバーを再接続。通常は `'failed'` 状態からの復帰に使用 | | `client.toggle_mcp_server(name, enabled)` | 特定サーバーを有効化 / 無効化。無効化すると接続が切断され、そのツールも削除される | > ⚠️ これら 3 つのメソッドはすべて tools リストの再構築を引き起こすため、プロンプトプレフィックスキャッシュを破壊します。本番環境では起動時に `mcp_servers` を完全に設定することを優先し、これらの API はデバッグやローカル開発用途に留めることを推奨します。
### コントロールリクエストのタイムアウト ```python theme={null} options = QoderAgentOptions( control_request_timeout_ms=20_000, # デフォルトは 60_000。0 を渡すと無効化 ) ``` タイムアウト後、SDK は自動的に `control_cancel_request` を書き込み、現在の Future を reject します。
## OAuth 認証 リモート MCP サーバー（HTTP/SSE）は OAuth を必要とすることがよくあります。CLI には完全な OAuth 2.0 + PKCE + Dynamic Client Registration（RFC 7591）実装が組み込まれています。Python SDK は **inbound（CLI が能動的にホストに OAuth 完了を要求）** と **outbound（ホストが能動的に OAuth をトリガー）** の 2 つのパスを公開しており、ホストの形態に応じて選択できます。 > ⚠️ **キャッシュの原則**：OAuth 完了後、CLI はサーバーに再接続して tools を再発見するため、**セッション途中での認証完了は必然的にプロンプトプレフィックスキャッシュを破壊します**。最初のユーザーメッセージを送信する**前に**認証を完了し、tools リストが安定してから対話を開始することを推奨します。 > 💡 **このセクションは CLI 主導の OAuth のみをカバーします**：CLI 自身が metadata discovery、PKCE、token 交換、token 永続化を行います。別の**サーバー主導**の認証パスもあります——サーバーが MCP `elicitation/create` を使用してクライアントを特定の URL にリダイレクトさせて認証を完了させます。2 つのパスは独立しており、同時にトリガーされることはありません。詳細は [Elicitation：サーバーによるユーザー入力要求](#elicitationサーバーによるユーザー入力要求) を参照してください。
### Inbound：`on_mcp_oauth_required` コールバック CLI がハンドシェイク中にサーバーで OAuth が必要であることを検出すると、control\_request を介して `McpOAuthRequest` を SDK に送信し、SDK がホストの `on_mcp_oauth_required` コールバックを呼び出します。ホストは以下のいずれかの resolution を返します。 | 戻り値の型 | 意味 | | ---------------------------------------- | ----------------------------------------------------------------- | | `OAuthToken` または `{"token": OAuthToken}` | ホストが OAuth フロー全体を完了し、token を直接 CLI に注入 | | `{"callbackUrl": "..."}` | ホストは完全なコールバック URL（`code` / `state` を含む）のみを取得し、CLI が解析して token に交換 | | `{"code": "...", "state": "..."}` | ホスト自身が code を解析し、CLI に直接返す | | `None` | 拒否。CLI は該当サーバーを failed としてマーク | ```python theme={null} async def handle_oauth(request: McpOAuthRequest) -> McpOAuthResolution | None: # Electron BrowserWindow またはシステムブラウザで request['auth_url'] を開く callback_url = await open_browser_and_wait_for_callback(request["auth_url"]) return {"callbackUrl": callback_url} options = QoderAgentOptions( mcp_servers={"analytics": {"type": "http", "url": "https://analytics.example.com/mcp"}}, on_mcp_oauth_required=handle_oauth, control_request_timeout_ms=120_000, # ユーザー認可には時間がかかる場合がある ) ```
### Outbound：ホストが能動的に認証を駆動 ホスト自身の UI に「Sign in」入口がある場合、能動的に呼び出すことができます。 ```python theme={null} async with QoderSDKClient(options) as client: status = await client.get_mcp_status() for server in status["mcpServers"]: if server["status"] != "needs-auth": continue result = await client.mcp_authenticate(server["name"]) if result.get("requiresUserAction"): await open_in_browser(result["authUrl"]) callback_url = await wait_for_user_paste_callback() await client.mcp_submit_oauth_callback_url(server["name"], callback_url) # サイレント経路（cached client + 有効な refresh token）: requiresUserAction=False # UI を表示せず、server は直接 connected に進む # tools リストは安定済みなので、ユーザーメッセージを送信できる await client.query("first user message") ``` | メソッド | 用途 | 呼び出しタイミング | | ---------------------------------------------------------- | ------------------------------------------------------------------------------------------- | ------------------------------- | | `client.mcp_authenticate(name, redirect_uri=None)` | OAuth を起動。`{authUrl?, requiresUserAction}` を返す。サイレント更新時は `requiresUserAction=False` で UI 不要 | **最初の user message 前** | | `client.mcp_submit_oauth_callback_url(name, callback_url)` | 完全なコールバック URL（code/state を含む）を送信 | **最初の user message 前** | | `client.inject_mcp_token(name, token)` | ホスト自身が OAuth フロー全体を完了し、`OAuthToken` を直接 CLI に注入 | **最初の user message 前** | | `client.mcp_clear_auth(name)` | CLI に保存された OAuth 認証情報を削除。「sign out」相当 | 任意のタイミング。次回のツール呼び出しで再認証がトリガーされる | `redirect_uri` はオプションで、デフォルトの OAuth コールバック先を上書きします（Electron カスタムプロトコル、社内ネットワークコールバックアドレスなど）。 CLI はデフォルトで token をシステム Keychain（macOS / Linux Secret Service）に保存し、フォールバックとして `~/.qoder/mcp-oauth-tokens.json`（パーミッション 0o600 + クロスプロセスロック）を使用します。
## Elicitation：サーバーによるユーザー入力要求 MCP `elicitation/create` は **server → client** 方向のリクエストで、クライアントにユーザーの前でインタラクションを表示させるためのものです（form モードは構造化入力を収集、url モードはユーザーを特定の URL に誘導して操作を完了させる）。 > ✅ **Python SDK は TS SDK に対応済み**：`QoderAgentOptions.on_elicitation` は `ElicitationResult` を返す async コールバックを受け取り、シグネチャは TS 版と一致しています。コールバック未設定時は、SDK はデフォルト契約に従って自動的に `{"action": "cancel"}` を応答します。`Elicitation` / `ElicitationResult` の 2 種類の hook イベントは引き続き並行してトリガーされ、読み取り専用の観察チャネルとして機能します。 > ⚠️ **現時点で CLI は `elicitation.url` capability を advertise していません**。サーバー側の `elicit({mode: 'url'})` は CLI から直接拒否されます（`MCP error -32602: Client does not support URL-mode elicitation requests`）。そのため **URL モードの elicit は SDK に到達せず、`system/elicitation_complete` 通知も現状の CLI ではトリガーされません**。CLI が URL capability を有効化すれば、このパスは自動的に接続されます。
### `on_elicitation` で elicit に応答 ```python theme={null} from qoder_agent_sdk import ElicitationRequest, ElicitationResult, QoderAgentOptions async def on_elicitation(req: ElicitationRequest) -> ElicitationResult: # form モード: req["requestedSchema"] は JSON Schema。返す content は一致する必要がある。 if req.get("mode") == "form": return {"action": "accept", "content": {"token": "xxx"}} # 処理しにくい場合は decline / cancel を返す。CLI は結果を MCP server に返送する。 return {"action": "decline"} options = QoderAgentOptions( mcp_servers={"my_server": {"type": "http", "url": "..."}}, on_elicitation=on_elicitation, ) ``` * フィールド名は TS SDK の camelCase に従います（`serverName / elicitationId / requestedSchema / displayName`）。CLI の snake\_case payload は SDK が自動的に変換します。 * `None` を返すことは `{"action": "cancel"}` と等価で、ホストはフォールバック経路で簡単にあきらめることができます。 * `mcp.types.ElicitResult` Pydantic モデルを返すこともできます（SDK が `model_dump` を呼び出します）。
### Elicitation の観察（hook チャネル） `on_elicitation` を実装した後でも、`Elicitation` / `ElicitationResult` の hook は引き続き並行してトリガーされます——これらは読み取り専用の観察チャネルであり、**意思決定には**関与しません。 | Hook イベント | タイミング | payload TypedDict | | ------------------- | ----------------- | ----------------------------------------------------------------------------------------------------------------- | | `Elicitation` | サーバーリクエスト到達時 | `ElicitationHookInput` — `mcp_server_name / message / mode / elicitation_id? / requested_schema? / url? / title?` | | `ElicitationResult` | SDK / host の応答完了後 | `ElicitationResultHookInput` — `mcp_server_name / action / mode / elicitation_id? / content?` | ```python theme={null} from qoder_agent_sdk import HookMatcher, QoderAgentOptions async def on_elicit(input, tool_use_id, context): print( "elicit from", input["mcp_server_name"], "mode=", input["mode"], "schema=", input.get("requested_schema"), ) return {"continue_": True} options = QoderAgentOptions( mcp_servers={"my_server": {"type": "http", "url": "..."}}, hooks={ "Elicitation": [HookMatcher(hooks=[on_elicit])], }, ) ```
### OAuth パスとの境界 * **CLI 主導の OAuth**（`mcp_authenticate` / `inject_mcp_token` / `on_mcp_oauth_required`）：token は qodercli Keychain に保存。`get_mcp_status()` が `needs-auth` の時に駆動。**Elicitation hook はトリガーされません**。 * **サーバー主導の elicit**：token はサーバー内部に保存。`get_mcp_status()` は `needs-auth` を示さない。`on_elicitation` コールバックで意思決定（未登録時は SDK が自動で cancel）。 2 つのパスは相互排他ではありませんが重複もしません。同じサーバーは通常どちらか一方のみを使用します。
## Options クイックリファレンス | フィールド | 型 | デフォルト | 説明 | | ---------------------------- | ------------------------------------------- | -------- | ---------------------------------------------------------------------------- | | `mcp_servers` | `dict[str, McpServerConfig] \| str \| Path` | `{}` | サーバー名 → 設定。または JSON 設定ファイルのパス | | `allowed_mcp_server_names` | `list[str]` | `[]` | プロセス型サーバーのホワイトリスト（インプロセスには影響なし）。空リスト＝すべて開放 | | `strict_mcp_config` | `bool` | `False` | CLI がユーザー設定ファイルから追加 MCP を読み込むことを禁止 | | `tools` | `list[str] \| ToolsPreset \| None` | `None` | **モデルに見えるツールのホワイトリスト**。渡さない＝すべての組み込み + MCP ツールが見える | | `allowed_tools` | `list[str]` | `[]` | **事前承認**リスト（承認プロンプトをスキップ。可視性は制御**しない**）。空リスト＝事前承認ルール無し | | `disallowed_tools` | `list[str]` | `[]` | 明示的に拒否するツール。allow よりも優先 | | `control_request_timeout_ms` | `int` | `60_000` | control リクエストのタイムアウト（mcp 系列を含む）。0 で無効化 | | `on_mcp_oauth_required` | `OnMcpOAuthRequired \| None` | `None` | CLI がサーバーの OAuth 必要を検出した際にトリガー | | `on_mcp_status_change` | `OnMcpStatusChange \| None` | `None` | サーバーステータスが変化するたびにトリガー。`system/mcp_status_change` ストリームをフィルタリングするのと等価 | | `on_elicitation` | `OnElicitation \| None` | `None` | サーバーが MCP `elicitation/create` でユーザー入力を要求した際にホストが意思決定。未設定時は SDK が自動的に cancel | | `hooks['Elicitation']` | `list[HookMatcher]` | – | サーバーがユーザー入力を要求した際の読み取り専用観察 hook（意思決定は `on_elicitation` 経由） |
### QoderSDKClient のメソッド | メソッド | 説明 | 呼び出しタイミング | | --------------------------------------------------- | ---------------------------------------------- | --------------------------- | | `get_mcp_status()` | 現在のすべての MCP サーバー状態を取得 | 任意のタイミング | | `set_mcp_servers(servers)` | MCP サーバー設定を全量置換。`{added, removed, errors}` を返す | 任意のタイミング（プレフィックスキャッシュを破壊する） | | `reconnect_mcp_server(name)` | 指定サーバーを再接続 | 任意のタイミング | | `toggle_mcp_server(name, enabled)` | サーバーを有効化 / 無効化 | 任意のタイミング | | `mcp_authenticate(name, redirect_uri=None)` | OAuth を能動的に開始 | **最初の user message 前** | | `mcp_submit_oauth_callback_url(name, callback_url)` | OAuth コールバック URL を送信 | **最初の user message 前** | | `inject_mcp_token(name, token)` | ホスト自身が OAuth フローを完了した後、token を注入 | **最初の user message 前** | | `mcp_clear_auth(name)` | 保存済みの OAuth 認証情報を削除 | 任意のタイミング |
## 型リファレンス ```python theme={null} from qoder_agent_sdk import ( # ファクトリ create_sdk_mcp_server, tool, SdkMcpTool, # 設定 McpServerConfig, McpStdioServerConfig, McpSSEServerConfig, McpHttpServerConfig, McpSdkServerConfig, McpServerToolPolicy, # ステータス McpServerStatus, McpServerStatusConfig, McpServerConnectionStatus, McpServerInfo, McpToolInfo, McpToolAnnotations, McpStatusResponse, McpStatusChangeMessage, # ランタイム変更 McpSetServersResult, # OAuth OAuthToken, McpOAuthRequest, McpOAuthResolution, McpOAuthTokenResolution, McpOAuthCallbackUrlResolution, McpOAuthCodeResolution, OnMcpOAuthRequired, OnMcpStatusChange, ) ``` `McpServerStatus.status` の列挙値（`McpServerConnectionStatus`）： | 値 | 意味 | | -------------- | ---------------------------------------- | | `'pending'` | 登録済み、接続未開始 | | `'connecting'` | ハンドシェイク中 | | `'connected'` | 接続済み、ツール呼び出し可能 | | `'failed'` | 接続失敗（`error` フィールドを参照） | | `'needs-auth'` | OAuth が必要。認証フローを実行してください | | `'disabled'` | 無効化（CLI 内部設定または `toggle_mcp_server` による） |
## ベストプラクティス 1. **説明は AI 向けに記述**：`@tool` の `description` は AI がいつそのツールを選択するかを決定します。「何をするか、いつ使うか、何に使うべきでないか」を明確に記述してください。 2. **パラメータに `Annotated` を付ける**：シンプルな dict / TypedDict のフィールドには `Annotated[type, "..."]` を記述してください。AI はこの情報を使って呼び出しパラメータを構築します。 3. **失敗には `is_error: True` を使い、例外をスローしない**：AI に結果を見せてください。完全な比較は [ツール使用ガイド - SDK が tool 戻り値のエラーを処理する方法](/ja/cli/sdk/python/tools#sdk-が-tool-戻り値のエラーを処理する方法)を参照してください。 4. **読み取り専用 + `readOnlyHint` を優先**：書き込み操作は慎重に行い、`can_use_tool` または hooks で二次確認を行ってください。 5. **サーバー名は短く**：ツールプレフィックスに含まれるため、長すぎる名前はトークンを浪費します。 6. **インプロセスの共有状態はモジュールスコープに配置**：handler はクロージャですが、各 query は同じ server インスタンスを再利用します。 7. **OAuth は最初の user message 前に完了**：`mcp_authenticate` + `mcp_submit_oauth_callback_url`、または `on_mcp_oauth_required` inbound コールバック、または `inject_mcp_token` を使用してください。セッション途中で認証を完了すると必然的にプロンプトプレフィックスキャッシュを破壊します。 8. **MCP 状態は `get_mcp_status()` または `on_mcp_status_change` で取得**：push チャネル（status change message）は引き続き利用可能です。必要に応じてどちらかを選んでください。 9. **適切な `control_request_timeout_ms` を設定**：リモートサーバーのハンドシェイクは秒単位になる可能性があります。デフォルトの 60 秒で通常は十分ですが、OAuth でユーザー操作を待つ場合は大きめに、CI 環境では明示的に指定してください。 10. **分離には `strict_mcp_config` を使用**：ユーザーローカルの `~/.qoder/settings.json` / `.mcp.json` で宣言された MCP サーバーがアプリケーションに干渉することを防ぎます。
## 完全な例 ```python theme={null} import asyncio import json import os from typing import Annotated from mcp.types import ToolAnnotations from qoder_agent_sdk import ( AssistantMessage, McpOAuthRequest, McpOAuthResolution, QoderAgentOptions, QoderSDKClient, ResultMessage, TextBlock, create_sdk_mcp_server, tool, ) # 1. 業務ツールを定義 @tool( "get_user_orders", "ユーザーの注文を検索し、ステータスで絞り込めます。", { "user_id": Annotated[str, "ユーザー UUID"], "status": Annotated[str, "注文ステータスで絞り込み: pending/paid/shipped/cancelled"], }, annotations=ToolAnnotations(readOnlyHint=True), ) async def get_user_orders(args): try: orders = await db.get_orders(args["user_id"], args.get("status")) return {"content": [{"type": "text", "text": json.dumps(orders)}]} except Exception as e: return { "is_error": True, "content": [{"type": "text", "text": f"検索に失敗しました: {e}"}], } # 2. server を組み立て crm = create_sdk_mcp_server(name="crm", tools=[get_user_orders]) # 3. inbound OAuth コールバックを準備（リモート server が認証を必要とする場合に CLI が呼び出す） async def handle_oauth(request: McpOAuthRequest) -> McpOAuthResolution | None: callback_url = await open_browser_and_wait_for_callback(request["auth_url"]) return {"callbackUrl": callback_url} # 4. client を起動し、先に認証を完了してから最初のメッセージを送信 async def main(): options = QoderAgentOptions( mcp_servers={ "crm": crm, "analytics": {"type": "http", "url": "https://analytics.example.com/mcp"}, }, allowed_tools=["mcp__crm__get_user_orders"], on_mcp_oauth_required=handle_oauth, control_request_timeout_ms=120_000, ) async with QoderSDKClient(options) as client: # outbound フォールバック: CLI が inbound を開始しない場合でも、状態を取得して自前で進められる status = await client.get_mcp_status() for server in status["mcpServers"]: if server["status"] != "needs-auth": continue result = await client.mcp_authenticate(server["name"]) if result.get("requiresUserAction"): callback_url = await open_browser_and_wait_for_callback(result["authUrl"]) await client.mcp_submit_oauth_callback_url(server["name"], callback_url) # 5. メッセージを消費（この時点で tools リストは安定し、プレフィックスキャッシュを正しく構築できる） await client.query("user-123 の最近の支払い済み注文を調べて") async for msg in client.receive_response(): if isinstance(msg, AssistantMessage): for block in msg.content: if isinstance(block, TextBlock): print(block.text) elif isinstance(msg, ResultMessage): if msg.subtype == "success": print("完了。cost=", msg.total_cost_usd) else: print("失敗:", msg.subtype) asyncio.run(main()) ```