MCP Client

实现一个极简的 MCP 客户端：以 JSON-RPC 2.0 与服务端通信，支持 initialize、 list_tools、call_tool，并演示工具调用的人工审批（human-in-the-loop）机制。 本示例中客户端直接持有服务端引用（进程内通信）；真实场景可替换为 stdio/SSE 传输层。 对应教程：第7卷《MCP》第3章 MCP Client。

文件路径: examples/volume-07-mcp\mcp_client.py

---

源代码

"""MCP Client 概念实现示例。

实现一个极简的 MCP 客户端：以 JSON-RPC 2.0 与服务端通信，支持 ``initialize``、
``list_tools``、``call_tool``，并演示工具调用的**人工审批（human-in-the-loop）**机制。

本示例中客户端直接持有服务端引用（进程内通信）；真实场景可替换为 stdio/SSE 传输层。

对应教程：第7卷《MCP》第3章 MCP Client。
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any

from mcp_server import JSONRPC_VERSION, McpServer

# 审批回调：接收 (tool_name, arguments) 返回 bool
ApprovalCallback = Callable[[str, dict[str, Any]], bool]


@dataclass
class McpClient:
    """MCP 客户端（概念实现）。

    Attributes:
        server: 连接的服务端（进程内引用）。
        require_approval: 是否对工具调用启用人工审批。
        approval_callback: 审批回调，``None`` 时默认拒绝。
    """

    server: McpServer
    require_approval: bool = False
    approval_callback: ApprovalCallback | None = None
    _request_id: int = 0
    _initialized: bool = False
    _call_history: list[dict[str, Any]] = field(default_factory=list)

    def _next_id(self) -> int:
        self._request_id += 1
        return self._request_id

    def _send(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        """向服务端发送 JSON-RPC 请求并返回响应。"""
        request: dict[str, Any] = {
            "jsonrpc": JSONRPC_VERSION,
            "method": method,
            "id": self._next_id(),
        }
        if params is not None:
            request["params"] = params
        return self.server.handle_request(request)

    def initialize(self) -> dict[str, Any]:
        """初始化连接。

        Returns:
            服务端返回的 serverInfo 与 capabilities。
        """
        resp = self._send("initialize")
        if "error" in resp:
            raise RuntimeError(resp["error"]["message"])
        self._initialized = True
        return resp["result"]

    def list_tools(self) -> list[dict[str, Any]]:
        """列出服务端工具。

        Returns:
            工具信息列表。
        """
        resp = self._send("tools/list")
        if "error" in resp:
            raise RuntimeError(resp["error"]["message"])
        return resp["result"]["tools"]

    def call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
        """调用工具，必要时触发审批。

        Args:
            name: 工具名。
            arguments: 参数。

        Returns:
            工具返回内容。

        Raises:
            PermissionError: 审批未通过。
            RuntimeError: 调用失败。
        """
        if self.require_approval:
            approved = self.approval_callback(name, arguments) if self.approval_callback else False
            self._call_history.append({"name": name, "arguments": arguments, "approved": approved})
            if not approved:
                raise PermissionError(f"工具调用 {name} 未获审批")
        else:
            self._call_history.append({"name": name, "arguments": arguments, "approved": True})

        resp = self._send("tools/call", {"name": name, "arguments": arguments})
        if "error" in resp:
            raise RuntimeError(resp["error"]["message"])
        return resp["result"]["content"][0]["text"]

    @property
    def call_history(self) -> list[dict[str, Any]]:
        """工具调用历史记录（便于审计）。"""
        return list(self._call_history)


def connect(server: McpServer, require_approval: bool = False) -> McpClient:
    """便捷工厂：创建客户端并完成初始化。

    Args:
        server: MCP 服务端。
        require_approval: 是否启用审批。

    Returns:
        已初始化的 ``McpClient``。
    """
    client = McpClient(server=server, require_approval=require_approval)
    client.initialize()
    return client


def main() -> None:
    """演示 MCP 客户端调用与审批。"""
    from mcp_server import build_demo_server

    server = build_demo_server()
    client = connect(server)
    print("工具列表:")
    for t in client.list_tools():
        print(f"  - {t['name']}: {t['description']}")

    print("\n调用 add_numbers:")
    print(" ", client.call_tool("add_numbers", {"a": 10, "b": 20}))

    print("\n启用审批（只允许 get_time）:")
    approved_client = McpClient(
        server=server,
        require_approval=True,
        approval_callback=lambda name, args: name == "get_time",
    )
    approved_client.initialize()
    print("  get_time:", approved_client.call_tool("get_time", {"timezone": "Asia/Shanghai"}))
    try:
        approved_client.call_tool("add_numbers", {"a": 1, "b": 2})
    except PermissionError as exc:
        print(f"  被拒绝: {exc}")


if __name__ == "__main__":
    main()