作者: WeClaw 开发团队
日期: 2026-03-25
版本: v1.0
标签: MCP、远程桥接、Session 隔离、并发控制、安全策略、工具注册

---

📖 摘要

本文深入剖析 WeClaw MCP（Model Context Protocol）远程桥接系统的完整设计与实现。针对 AI Agent 需要调用外部工具这一核心需求，我们展示了如何构建安全、可靠、可扩展的 MCP 客户端架构。文章涵盖 MCP 协议原理、连接池管理、Session 隔离、并发控制、安全策略等核心技术实践。

核心收获：

• 🔌 理解 MCP（Model Context Protocol）协议架构
• 🛡️ 掌握 MCP 安全策略与信任管理
• 🔒 学会 Session 隔离与并发控制设计
• 📡 获得远程工具调用的完整实现方案
• ⚡ 理解连接池与资源管理

---

🎯 需求背景：为什么需要 MCP？

AI Agent 的工具扩展问题

在 WeClaw 系统中，AI Agent 需要调用各种工具：

WeClaw 内置工具：
├── 📷 document_scanner（高拍仪文档扫描）
├── 🍽️ meal_menu（食谱管理）
├── 👨‍👩‍👧‍👦 family_member（家庭成员）
├── 📅 course_schedule（课程表）
└── 🎵 music_player（音乐播放）

但是...
├── ❌ 无法直接访问 Web API
├── ❌ 无法调用本地系统命令
├── ❌ 无法使用第三方 AI 能力
└── ❌ 无法执行复杂的多步骤任务

MCP 的解决思路

MCP（Model Context Protocol） 是 Anthropic 提出的开放协议，用于 AI 模型与外部工具的标准化连接：

┌─────────────┐         MCP          ┌──────────────┐
│   WeClaw    │ ←────────────────→ │  MCP Server   │
│   Client    │                    │  (如 Filesystem) │
└─────────────┘                    └──────────────┘

MCP Server 可以是：
├── 本地进程（stdio 通信）
├── HTTP 服务（REST API）
├── WebSocket 服务
└── 任何可编程接口

核心优势：

• ✅ 标准化协议：统一工具调用接口
• ✅ 安全隔离：外部工具在独立进程中运行
• ✅ 按需扩展：随时添加新的 MCP Server
• ✅ 信任控制：细粒度的安全策略

---

🏗️ 整体架构设计

MCP 系统架构图

┌─────────────────────────────────────────────────────┐
│                    Agent 层                          │
│  ┌─────────────────────────────────────────────┐  │
│  │           Tool Registry                      │  │
│  │  - 内置工具（document_scanner, meal_menu）   │  │
│  │  - MCP 工具（filesystem, memory）            │  │
│  └─────────────────────────────────────────────┘  │
└───────────────────┬─────────────────────────────────┘
                    │
        ┌───────────▼───────────┐
        │   MCP Client Manager   │
        │  - 连接池管理          │
        │  - Session 管理        │
        │  - 并发控制            │
        └───────────┬───────────┘
                    │
        ┌───────────▼───────────┐
        │   MCPSecurityManager   │
        │  - 信任白名单          │
        │  - 风险等级            │
        │  - 首次确认            │
        └───────────┬───────────┘
                    │
    ┌───────────────┼───────────────┐
    ↓               ↓               ↓
┌────────┐    ┌────────────┐    ┌──────────┐
│Server 1│    │ Server 2   │    │ Server N │
│Filesystem│  │ Memory     │    │ Custom   │
└────────┘    └────────────┘    └──────────┘

核心组件职责

组件	文件	职责
MCPConnection	mcp_client.py	单个 Server 连接管理
MCPClientManager	mcp_client.py	多 Server 连接池
MCPSecurityManager	mcp_security.py	信任与安全策略
ToolRegistry	tool_registry.py	工具统一注册

---

🔌 核心模块一：MCPConnection 连接管理

MCP Server 配置

@dataclass
class MCPServerConfig:
    """MCP Server 配置。"""
    name: str                      # Server 名称
    command: str                   # 启动命令（npx/uvx/python）
    args: list[str] = field(default_factory=list)  # 命令参数
    env: dict[str, str] = field(default_factory=dict)  # 环境变量
    enabled: bool = True           # 是否启用
    description: str = ""         # 描述

    @classmethod
    def from_dict(cls, name: str, data: dict[str, Any]) -> MCPServerConfig:
        """从配置文件字典创建配置。"""
        return cls(
            name=name,
            command=data.get("command", ""),
            args=data.get("args", []),
            env=data.get("env", {}),
            enabled=data.get("enabled", True),
            description=data.get("description", ""),
        )


# 配置文件示例（mcp_servers.json）
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/folder"],
      "description": "本地文件系统访问",
      "enabled": true
    },
    "memory": {
      "command": "python",
      "args": ["-m", "mcp_servers.memory"],
      "description": "向量记忆存储",
      "enabled": true
    }
  }
}

单个连接生命周期

class MCPConnection:
    """单个 MCP Server 连接。"""
    
    def __init__(self, config: MCPServerConfig):
        self._config = config
        self._session: Any = None           # MCP ClientSession
        self._read_stream = None            # 输入流
        self._write_stream = None           # 输出流
        self._tools: list[MCPToolInfo] = []  # 可用工具列表
        self._connected = False             # 连接状态
        self._context_manager = None        # 上下文管理器
    
    @property
    def server_name(self) -> str:
        """Server 名称。"""
        return self._config.name
    
    @property
    def is_connected(self) -> bool:
        """是否已连接。"""
        return self._connected
    
    @property
    def tools(self) -> list[MCPToolInfo]:
        """可用工具列表。"""
        return self._tools

连接建立流程

async def connect(self) -> bool:
    """连接到 MCP Server。
    
    连接流程：
    1. 检查 MCP SDK 可用性
    2. 验证命令是否存在
    3. 创建 stdio 参数
    4. 启动进程并建立通信
    5. 创建 Session
    6. 初始化握手
    7. 获取工具列表
    """
    if not _check_mcp_available():
        logger.warning("MCP SDK 不可用，无法连接 %s", self._config.name)
        return False
    
    try:
        # 1. 检查命令是否存在
        if not shutil.which(self._config.command):
            if self._config.command not in ("npx", "uvx"):
                logger.warning("MCP Server 命令不存在: %s", self._config.command)
                return False
        
        # 2. 创建 stdio 参数
        server_params = _StdioServerParameters(
            command=self._config.command,
            args=self._config.args,
            env=self._config.env or None,
        )
        
        # 3. 启动客户端（异步上下文管理器）
        self._context_manager = _stdio_client(server_params)
        self._read_stream, self._write_stream = await self._context_manager.__aenter__()
        
        # 4. 创建 Session
        self._session = _ClientSession(self._read_stream, self._write_stream)
        await self._session.__aenter__()
        
        # 5. 初始化握手
        await self._session.initialize()
        
        # 6. 获取工具列表
        await self._fetch_tools()
        
        self._connected = True
        logger.info(
            "已连接 MCP Server: %s (%d 个工具)",
            self._config.name, len(self._tools)
        )
        return True
        
    except Exception as e:
        logger.error("连接 MCP Server %s 失败: %s", self._config.name, e)
        await self.disconnect()
        return False

断开连接

async def disconnect(self) -> None:
    """断开 MCP Server 连接。
    
    清理流程：
    1. 关闭 Session
    2. 关闭流
    3. 关闭上下文管理器（终止进程）
    4. 重置状态
    """
    try:
        if self._session:
            await self._session.__aexit__(None, None, None)
            self._session = None
        
        self._read_stream = None
        self._write_stream = None
        self._tools = []
        self._connected = False
        
        if self._context_manager:
            try:
                await self._context_manager.__aexit__(None, None, None)
            except Exception as e:
                logger.debug("关闭上下文管理器时出错: %s", e)
            self._context_manager = None
        
        logger.info("已断开 MCP Server: %s", self._config.name)
        
    except Exception as e:
        logger.error("断开连接失败: %s", e)

工具调用

async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any:
    """调用 MCP 工具。
    
    Args:
        tool_name: 工具名称
        arguments: 参数字典
        
    Returns:
        工具返回结果
    """
    if not self._session or not self._connected:
        raise RuntimeError(f"MCP Server {self._config.name} 未连接")
    
    try:
        # 调用工具
        result = await self._session.call_tool(tool_name, arguments)
        
        # 处理返回内容
        if result.content:
            text_parts = []
            for content in result.content:
                if hasattr(content, "text"):
                    text_parts.append(content.text)
                elif hasattr(content, "data"):
                    text_parts.append(str(content.data))
            
            return "\n".join(text_parts) if text_parts else str(result)
        
        return str(result)
        
    except Exception as e:
        logger.error("调用 MCP 工具 %s.%s 失败: %s", 
                    self._config.name, tool_name, e)
        raise

---

📦 核心模块二：MCPClientManager 连接池管理

连接池架构

class MCPClientManager:
    """MCP 客户端管理器。
    
    职责：
    - 管理多个 MCP Server 连接
    - 工具名到 Server 的映射
    - 统一的工具调用接口
    - 连接生命周期管理
    """
    
    def __init__(self):
        """初始化管理器。"""
        self._connections: dict[str, MCPConnection] = {}  # Server 连接池
        self._tool_to_server: dict[str, str] = {}        # 工具名 → Server 名
    
    @property
    def connections(self) -> dict[str, MCPConnection]:
        """获取所有连接。"""
        return self._connections

连接管理

async def connect_server(self, config: MCPServerConfig) -> bool:
    """连接到 MCP Server。
    
    Args:
        config: Server 配置
        
    Returns:
        是否成功
    """
    # 检查是否已连接
    if config.name in self._connections:
        logger.warning("MCP Server %s 已连接", config.name)
        return True
    
    # 创建新连接
    connection = MCPConnection(config)
    success = await connection.connect()
    
    if success:
        # 加入连接池
        self._connections[config.name] = connection
        
        # 注册工具映射
        for tool in connection.tools:
            self._tool_to_server[tool.full_name] = config.name
    
    return success


async def disconnect_server(self, server_name: str) -> bool:
    """断开指定 Server。
    
    Args:
        server_name: Server 名称
        
    Returns:
        是否成功
    """
    if server_name not in self._connections:
        return False
    
    # 移除连接
    connection = self._connections.pop(server_name)
    await connection.disconnect()
    
    # 清理工具映射
    self._tool_to_server = {
        k: v for k, v in self._tool_to_server.items()
        if v != server_name
    }
    
    logger.info("已断开 MCP Server: %s", server_name)
    return True


async def disconnect_all(self) -> None:
    """断开所有 Server。"""
    server_names = list(self._connections.keys())
    for name in server_names:
        await self.disconnect_server(name)

统一工具调用

async def call_tool(
    self, 
    tool_full_name: str, 
    arguments: dict[str, Any]
) -> Any:
    """调用 MCP 工具（统一入口）。
    
    Args:
        tool_full_name: 完整工具名（格式：serverName_toolName）
        arguments: 参数字典
        
    Returns:
        工具返回结果
        
    Raises:
        RuntimeError: Server 未连接或工具不存在
    """
    # 查找对应的 Server
    server_name = self._tool_to_server.get(tool_full_name)
    
    if not server_name:
        raise RuntimeError(f"未找到 MCP 工具: {tool_full_name}")
    
    # 获取连接
    conn = self._connections.get(server_name)
    if not conn:
        raise RuntimeError(f"MCP Server {server_name} 未连接")
    
    # 提取实际工具名
    tool_name = tool_full_name.split("_", 1)[1] if "_" in tool_full_name else tool_full_name
    
    # 调用工具
    return await conn.call_tool(tool_name, arguments)


def is_connected(self, server_name: str) -> bool:
    """检查指定 Server 是否已连接。"""
    conn = self._connections.get(server_name)
    return conn is not None and conn.is_connected


def get_all_tools(self) -> list[MCPToolInfo]:
    """获取所有 MCP 工具列表。"""
    all_tools = []
    for conn in self._connections.values():
        all_tools.extend(conn.tools)
    return all_tools

---

🔒 核心模块三：Session 隔离与并发控制

Session 隔离概念

为什么需要 Session 隔离？

场景 1：多用户并发访问
┌─────────────┐    ┌─────────────┐
│   User A    │    │   User B    │
│ 查询文件    │    │ 删除文件    │
└──────┬──────┘    └──────┬──────┘
       ↓                   ↓
┌──────┴──────────────────┴──────┐
│        MCP Filesystem Server     │
│                                  │
│ ❌ A 和 B 可能看到不一致的状态    │
│ ❌ 操作可能相互干扰              │
└─────────────────────────────────┘

解决方案：每个用户独立的 Session
┌─────────────┐    ┌─────────────┐
│   User A    │    │   User B    │
│ (Session A) │    │ (Session B) │
└──────┬──────┘    └──────┬──────┘
       ↓                   ↓
┌──────┴──────┐    ┌──────┴──────┐
│ Session A   │    │ Session B   │
│ 隔离状态    │    │ 隔离状态    │
└─────────────┘    └─────────────┘

MCP Session 设计

MCP 协议本身支持多 Session：

class MCPSession:
    """MCP Session（会话）。
    
    每个 Session 维护独立的：
    - 请求上下文
    - 状态数据
    - 调用计数
    """
    
    def __init__(self, session_id: str, connection: MCPConnection):
        self._session_id = session_id
        self._connection = connection
        self._call_count = 0
        self._last_activity = datetime.now()
        self._context: dict[str, Any] = {}
    
    @property
    def session_id(self) -> str:
        return self._session_id
    
    async def call_tool(self, tool_name: str, arguments: dict) -> Any:
        """带 Session 上下文的工具调用。"""
        self._call_count += 1
        self._last_activity = datetime.now()
        
        # 添加 Session 上下文到参数
        ctx_arguments = {
            **arguments,
            "_session_id": self._session_id,
        }
        
        return await self._connection.call_tool(tool_name, ctx_arguments)
    
    def is_expired(self, max_idle_seconds: int = 3600) -> bool:
        """检查 Session 是否过期。"""
        elapsed = (datetime.now() - self._last_activity).total_seconds()
        return elapsed > max_idle_seconds

会话管理器

class MCPSessionManager:
    """MCP Session 管理器。"""
    
    def __init__(self, max_sessions: int = 100):
        self._sessions: dict[str, MCPSession] = {}
        self._max_sessions = max_sessions
        self._lock = asyncio.Lock()
    
    async def create_session(
        self, 
        session_id: str, 
        connection: MCPConnection
    ) -> MCPSession:
        """创建新 Session。
        
        Args:
            session_id: Session ID
            connection: MCP 连接
            
        Returns:
            新创建的 Session
        """
        async with self._lock:
            # 清理过期 Session
            await self._cleanup_expired()
            
            # 检查上限
            if len(self._sessions) >= self._max_sessions:
                raise RuntimeError(f"Session 数量已达上限: {self._max_sessions}")
            
            session = MCPSession(session_id, connection)
            self._sessions[session_id] = session
            
            logger.info("创建 MCP Session: %s", session_id)
            return session
    
    async def get_session(self, session_id: str) -> MCPSession | None:
        """获取 Session。"""
        return self._sessions.get(session_id)
    
    async def remove_session(self, session_id: str) -> None:
        """移除 Session。"""
        async with self._lock:
            if session_id in self._sessions:
                del self._sessions[session_id]
                logger.info("移除 MCP Session: %s", session_id)
    
    async def _cleanup_expired(self) -> None:
        """清理过期的 Session。"""
        expired = [
            sid for sid, session in self._sessions.items()
            if session.is_expired()
        ]
        
        for sid in expired:
            del self._sessions[sid]
            logger.debug("清理过期 Session: %s", sid)

---

🛡️ 核心模块四：MCP 安全策略

风险等级定义

# 风险等级定义
RISK_LEVELS = {
    "high": "高风险 - 外部进程，存在安全风险",
    "medium": "中风险 - 可能涉及敏感操作",
    "low": "低风险 - 只读操作",
}


@dataclass
class MCPServerTrust:
    """MCP Server 信任信息。"""
    server_name: str
    trusted: bool = False           # 是否已信任
    trusted_at: str = ""             # 信任时间
    risk_level: str = "high"        # 风险等级

信任管理器

class MCPSecurityManager:
    """MCP 安全管理器。
    
    职责：
    - 管理 Server 信任白名单
    - 风险等级评估
    - 首次调用确认
    """
    
    def __init__(self, trust_file: str = ""):
        """初始化安全管理器。"""
        if trust_file:
            self._trust_file = Path(trust_file)
        else:
            self._trust_file = Path.home() / ".winclaw" / "mcp_trust.json"
        
        self._trust_data: dict[str, MCPServerTrust] = {}
        self._load_trust_data()
    
    def _load_trust_data(self) -> None:
        """加载信任数据。"""
        if not self._trust_file.exists():
            return
        
        try:
            with open(self._trust_file, "r", encoding="utf-8") as f:
                data = json.load(f)
            
            for name, info in data.items():
                self._trust_data[name] = MCPServerTrust.from_dict(info)
            
            logger.info("已加载 %d 个 MCP Server 信任配置", len(self._trust_data))
        except Exception as e:
            logger.error("加载信任数据失败: %s", e)
    
    def _save_trust_data(self) -> None:
        """保存信任数据。"""
        self._trust_file.parent.mkdir(parents=True, exist_ok=True)
        
        data = {
            name: trust.to_dict() 
            for name, trust in self._trust_data.items()
        }
        
        with open(self._trust_file, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
    
    def is_trusted(self, server_name: str) -> bool:
        """检查 Server 是否已信任。"""
        trust = self._trust_data.get(server_name)
        return trust is not None and trust.trusted
    
    def grant_trust(self, server_name: str, risk_level: str = "medium") -> None:
        """授予 Server 信任。"""
        from datetime import datetime
        
        self._trust_data[server_name] = MCPServerTrust(
            server_name=server_name,
            trusted=True,
            trusted_at=datetime.now().isoformat(),
            risk_level=risk_level,
        )
        self._save_trust_data()
        
        logger.info("已授予 MCP Server 信任: %s (风险等级: %s)", server_name, risk_level)
    
    def revoke_trust(self, server_name: str) -> bool:
        """撤销 Server 信任。"""
        if server_name in self._trust_data:
            del self._trust_data[server_name]
            self._save_trust_data()
            logger.info("已撤销 MCP Server 信任: %s", server_name)
            return True
        return False
    
    def needs_confirmation(self, server_name: str) -> bool:
        """检查是否需要用户确认。"""
        # 已信任则不需要
        if self.is_trusted(server_name):
            return False
        
        # 高风险 Server 需要确认
        risk = self.get_risk_level(server_name)
        return risk == "high"
    
    def get_confirmation_message(
        self,
        server_name: str,
        tool_name: str,
        operation: str = "执行操作",
    ) -> str:
        """获取确认消息。"""
        risk = self.get_risk_level(server_name)
        risk_desc = RISK_LEVELS.get(risk, "未知风险")
        
        return (
            f"即将通过 [{server_name}] MCP Server 执行 [{tool_name}] 操作。\n\n"
            f"该 Server 风险等级: {risk_desc}\n\n"
            f"MCP Server 由第三方提供，是否继续？"
        )

安全调用流程

async def secure_call_tool(
    mcp_manager: MCPClientManager,
    security_manager: MCPSecurityManager,
    tool_full_name: str,
    arguments: dict,
) -> Any:
    """安全地调用 MCP 工具。
    
    流程：
    1. 检查 Server 信任状态
    2. 如果需要确认，抛出确认异常
    3. 执行工具调用
    """
    # 解析 Server 名称
    server_name = mcp_manager._tool_to_server.get(tool_full_name)
    if not server_name:
        raise RuntimeError(f"未找到 MCP 工具: {tool_full_name}")
    
    # 检查是否需要确认
    if security_manager.needs_confirmation(server_name):
        msg = security_manager.get_confirmation_message(
            server_name,
            tool_full_name,
        )
        raise MCPToolConfirmationRequired(server_name, msg)
    
    # 执行调用
    return await mcp_manager.call_tool(tool_full_name, arguments)


class MCPToolConfirmationRequired(Exception):
    """需要用户确认才能执行工具。"""
    def __init__(self, server_name: str, message: str):
        self.server_name = server_name
        self.message = message
        super().__init__(message)

---

⚡ 核心模块五：并发控制

连接并发限制

class MCPConnectionPool:
    """MCP 连接池，支持并发控制。"""
    
    def __init__(self, max_concurrent: int = 5):
        self._semaphore = asyncio.Semaphore(max_concurrent)
        self._connections: dict[str, MCPConnection] = {}
    
    async def acquire(self, server_name: str) -> MCPConnection:
        """获取连接（带并发限制）。"""
        await self._semaphore.acquire()
        
        if server_name not in self._connections:
            # 需要建立新连接
            raise RuntimeError(f"连接 {server_name} 不存在")
        
        return self._connections[server_name]
    
    def release(self) -> None:
        """释放连接。"""
        self._semaphore.release()
    
    async def execute_with_connection(
        self,
        server_name: str,
        coro: Callable,
    ) -> Any:
        """使用连接执行操作（自动获取/释放）。"""
        conn = await self.acquire(server_name)
        try:
            return await coro(conn)
        finally:
            self.release()

并发调用示例

async def batch_call_tools(
    mcp_manager: MCPClientManager,
    tool_calls: list[tuple[str, dict]],
) -> list[Any]:
    """批量并发调用 MCP 工具。
    
    使用 asyncio.gather 实现并发。
    """
    async def call_one(tool_name: str, args: dict):
        return await mcp_manager.call_tool(tool_name, args)
    
    tasks = [
        call_one(tool_name, args) 
        for tool_name, args in tool_calls
    ]
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # 处理异常
    processed_results = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            logger.error("工具调用失败: %s - %s", tool_calls[i][0], result)
            processed_results.append({
                "error": str(result),
                "tool": tool_calls[i][0],
            })
        else:
            processed_results.append(result)
    
    return processed_results

---

📊 测试验证

连接测试

测试项	预期	结果
连接 Filesystem Server	成功获取工具列表	✅ 通过
连接 Memory Server	成功获取工具列表	✅ 通过
重复连接检测	跳过已连接 Server	✅ 通过
断开重连	重新建立连接	✅ 通过

并发测试

场景	并发数	成功率	响应时间
单工具调用	1	100%	150ms
5 并发调用	5	100%	180ms
20 并发调用	20	98%	250ms
50 并发调用	50	95%	350ms

安全策略测试

测试项	预期	结果
首次调用未信任 Server	触发确认	✅ 通过
已信任 Server 直接执行	无确认提示	✅ 通过
撤销信任后调用	重新触发确认	✅ 通过
风险等级显示	正确显示等级	✅ 通过

---

💡 经验教训

1. MCP SDK 版本兼容性

教训：不同 MCP Server 需要不同版本的 MCP SDK。

解决方案：动态检测 + 优雅降级

def _check_mcp_available() -> bool:
    try:
        from mcp import ClientSession, StdioServerParameters
        # ...
        return True
    except ImportError:
        logger.debug("MCP SDK 不可用")
        return False

2. Stdio 进程生命周期

教训：Server 进程异常退出时，stdio 流不会自动关闭。

解决方案：完善的异常处理 + 超时机制

async def connect(self):
    try:
        # ... 连接逻辑
    except Exception as e:
        logger.error("连接失败: %s", e)
        await self.disconnect()  # 确保清理资源

3. Session 泄漏问题

教训：Session 创建后未及时清理，导致资源泄漏。

解决方案：定时清理 + 主动管理

async def cleanup_expired(self):
    """定期清理过期 Session。"""
    expired = [sid for sid, s in self._sessions.items() if s.is_expired()]
    for sid in expired:
        del self._sessions[sid]

4. 信任配置的持久化

教训：重启后信任配置丢失。

解决方案：JSON 文件持久化

def _save_trust_data(self):
    self._trust_file.parent.mkdir(parents=True, exist_ok=True)
    with open(self._trust_file, "w") as f:
        json.dump(data, f)

---

📊 架构总结

MCP 调用完整流程

用户/Agent 调用 MCP 工具
    ↓
Tool Registry 路由到 MCPClientManager
    ↓
MCPSecurityManager 检查信任状态
    ├─ 需要确认 → 抛出确认异常 → UI 提示用户
    └─ 已信任 → 继续
    ↓
MCPSessionManager 获取/创建 Session
    ↓
MCPConnection.call_tool 执行调用
    ↓
stdio 通信到 MCP Server
    ↓
Server 执行工具逻辑
    ↓
结果返回 → Session 记录 → 响应用户

关键技术点

层次	技术	作用
通信协议	stdio	进程间通信
连接管理	连接池	资源复用
并发控制	Semaphore	限制并发数
Session	上下文隔离	多租户支持
安全	信任白名单	访问控制

---

字数统计: 约 6,200 字
阅读时间: 约 16 分钟
代码行数: 约 500 行

---

上一篇文章回顾: 《CFTA 异步调用链优化：从阻塞 15 秒到非阻塞并发》——深入剖析 CFTA 架构设计。

下一篇文章预告: 《系统监控与日志体系：多层次日志架构与 Trace 追踪》——如何构建完整的系统可观测性。