MCP 集成

┌────────────────────────────────────────────────────────────┐
│  你的 Python 应用 (SDK Host)                               │
│                                                            │
│   ┌──────────────────────────┐                             │
│   │ create_sdk_mcp_server(...) │ ← In-Process 工具         │
│   │  + @tool(...)             │     直接定义、零进程        │
│   └──────────────────────────┘                             │
│                  │                                         │
│                  ▼                                         │
│   ┌──────────────────────────┐                             │
│   │  query({mcp_servers})    │── stdio ─▶ qodercli 子进程  │
│   │  / 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())