MCP 統合

┌────────────────────────────────────────────────────────────┐
│  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)
└────────────────────────────────────────────────────────────┘

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())

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: ...

# 成功
{"content": [{"type": "text", "text": "結果"}]}

# 業務上の失敗: 例外を投げずに is_error を使用
{"content": [{"type": "text", "text": "エラー説明"}], "is_error": True}

@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)}]}

@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"}]}

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]    # 省略可能。stdio がデフォルト
    command: str                           # 実行コマンド
    args: NotRequired[list[str]]           # コマンド引数
    env: NotRequired[dict[str, str]]       # 環境変数
    tools: NotRequired[list[McpServerToolPolicy]]

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"]},
        },
    },
)

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]]

options = QoderAgentOptions(
    mcp_servers={
        "analytics": {
            "type": "http",
            "url": "https://analytics.example.com/mcp",
            "headers": {"Authorization": f"Bearer {os.environ['ANALYTICS_TOKEN']}"},
        },
    },
)

mcp__<server_name>__<tool_name>

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",
    ],
)

options = QoderAgentOptions(
    mcp_servers={"my_tools": server},
    allowed_tools=[
        "mcp__my_tools__greet",          # pre-approved, no prompt
        "mcp__my_tools__search_docs",
    ],
)

options = QoderAgentOptions(
    mcp_servers={
        "keep": {"command": "..."},
        "drop": {"command": "..."},
    },
    allowed_mcp_server_names=["keep"],   # 'drop' はステータスに残るが接続しない
    strict_mcp_config=True,              # settings.json / .mcp.json から MCP サーバーを読み込まない
)

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", [])])

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,
)

options = QoderAgentOptions(
    control_request_timeout_ms=20_000,   # デフォルトは 60_000。0 を渡すと無効化
)

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,   # ユーザー認可には時間がかかる場合がある
)

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")

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,
)

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])],
    },
)

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,
)

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())