第24章 添加 MCP 与多智能体支持 —— Mini-Harness 的能力扩展

# MCP 客户端实现（简化版）

class MCPClient:
    """
    MCP 客户端（简化版）
    
    实际 MCP 协议需要：
    1. stdio、HTTP、WebSocket 通信
    2. JSON-RPC 协议
    3. 工具 Schema 定义
    
    这里简化为直接调用本地模拟工具
    """
    
    def __init__(self, server_name, server_config):
        """
        初始化 MCP 客户端
        
        参数:
            server_name: MCP 服务器名称
            server_config: MCP 服务器配置
        """
        self.server_name = server_name
        self.server_config = server_config
        self.tools = []
    
    def connect(self):
        """
        连接 MCP 服务器
        
        实际实现需要：
        - stdio: 启动进程，通过 stdin/stdout 通信
        - HTTP: 发送 HTTP 请求
        - WebSocket: 建立 WebSocket 连接
        
        简化版：直接加载工具列表
        """
        # 简化版：模拟加载工具
        if self.server_name == "filesystem":
            self.tools = self.load_filesystem_tools()
        elif self.server_name == "database":
            self.tools = self.load_database_tools()
        elif self.server_name == "web":
            self.tools = self.load_web_tools()
        
        return self.tools
    
    def load_filesystem_tools(self):
        """加载 filesystem MCP 工具"""
        return [
            {
                "name": "mcp_filesystem_read",
                "description": "Read file via MCP filesystem",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "path": {"type": "string", "description": "File path"}
                    },
                    "required": ["path"]
                }
            },
            {
                "name": "mcp_filesystem_write",
                "description": "Write file via MCP filesystem",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "path": {"type": "string"},
                        "content": {"type": "string"}
                    },
                    "required": ["path", "content"]
                }
            }
        ]
    
    def load_database_tools(self):
        """加载 database MCP 工具"""
        return [
            {
                "name": "mcp_database_query",
                "description": "Execute SQL query",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "sql": {"type": "string", "description": "SQL query"}
                    },
                    "required": ["sql"]
                }
            }
        ]
    
    def load_web_tools(self):
        """加载 web MCP 工具"""
        return [
            {
                "name": "mcp_web_fetch",
                "description": "Fetch webpage",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "url": {"type": "string", "description": "URL to fetch"}
                    },
                    "required": ["url"]
                }
            }
        ]
    
    def call_tool(self, tool_name, params):
        """
        调用 MCP 工具
        
        实际实现需要：
        - 发送 JSON-RPC 请求
        - 等待响应
        - 解析结果
        
        简化版：直接执行模拟操作
        """
        if tool_name == "mcp_filesystem_read":
            return self.mock_filesystem_read(params)
        elif tool_name == "mcp_filesystem_write":
            return self.mock_filesystem_write(params)
        elif tool_name == "mcp_database_query":
            return self.mock_database_query(params)
        elif tool_name == "mcp_web_fetch":
            return self.mock_web_fetch(params)
        else:
            return f"错误：工具 {tool_name} 不存在"
    
    # 模拟工具执行
    def mock_filesystem_read(self, params):
        """模拟 filesystem read"""
        import os
        path = params.get("path")
        if not os.path.exists(path):
            return f"错误：文件 {path} 不存在"
        with open(path, 'r') as f:
            return f"[MCP filesystem] 文件内容:\n{f.read()}"
    
    def mock_filesystem_write(self, params):
        """模拟 filesystem write"""
        path = params.get("path")
        content = params.get("content")
        with open(path, 'w') as f:
            f.write(content)
        return f"[MCP filesystem] 成功写入 {path}"
    
    def mock_database_query(self, params):
        """模拟 database query"""
        sql = params.get("sql")
        # 模拟返回数据
        return f"[MCP database] 查询结果:\nSELECT * FROM users LIMIT 5\n\nid | name | email\n1 | Alice | alice@example.com\n2 | Bob | bob@example.com"
    
    def mock_web_fetch(self, params):
        """模拟 web fetch"""
        url = params.get("url")
        return f"[MCP web] 网页内容:\n<title>Example Page</title>\n<p>This is an example webpage.</p>"


# 测试 MCP 客户端
def test_mcp_client():
    client = MCPClient("filesystem", {})
    tools = client.connect()
    
    print("filesystem MCP 工具列表:")
    for tool in tools:
        print(f"- {tool['name']}: {tool['description']}")
    
    # 测试调用
    result = client.call_tool("mcp_filesystem_read", {"path": "/tmp/test.txt"})
    print(f"\n调用结果:\n{result}")


# 运行测试
test_mcp_client()

# MCP 工具动态注册

class MCPRegistry:
    """
    MCP 工具注册中心
    """
    
    def __init__(self, harness):
        """
        初始化注册中心
        
        参数:
            harness: Harness 实例
        """
        self.harness = harness
        self.mcp_clients = {}
        self.mcp_tools = {}
    
    def register_mcp_server(self, server_name, server_config):
        """
        注册 MCP 服务器
        
        参数:
            server_name: MCP 服务器名称
            server_config: MCP 服务器配置
        """
        # 1. 创建 MCP 客户端
        client = MCPClient(server_name, server_config)
        
        # 2. 连接并获取工具列表
        tools = client.connect()
        
        # 3. 动态注册到 harness
        for tool in tools:
            tool_name = tool["name"]
            
            # 注册工具 Schema（添加到 System Prompt）
            self.harness.tools[tool_name] = lambda params: client.call_tool(tool_name, params)
            
            # 记录 MCP 工具
            self.mcp_tools[tool_name] = {
                "server": server_name,
                "schema": tool
            }
        
        # 4. 记录 MCP 客户端
        self.mcp_clients[server_name] = client
        
        # 5. 更新 System Prompt
        self.harness.system_prompt = self.update_system_prompt()
        
        return tools
    
    def update_system_prompt(self):
        """
        更新 System Prompt，添加 MCP 工具
        
        返回:
            system_prompt: 更新后的 System Prompt
        """
        base_prompt = self.harness.default_system_prompt()
        
        mcp_prompt = "\n\nMCP 工具：\n"
        for tool_name, tool_info in self.mcp_tools.items():
            schema = tool_info["schema"]
            mcp_prompt += f"- {schema['name']}: {schema['description']}\n"
        
        return base_prompt + mcp_prompt


# 测试 MCP 注册
def test_mcp_registry():
    # 创建 Harness（带权限控制）
    harness = MiniHarnessWithPermissions()
    
    # 创建注册中心
    registry = MCPRegistry(harness)
    
    # 注册 filesystem MCP
    tools = registry.register_mcp_server("filesystem", {})
    
    print("注册 MCP 工具:")
    for tool_name, tool_info in registry.mcp_tools.items():
        print(f"- {tool_name} (from {tool_info['server']})")
    
    print(f"\n更新后的 System Prompt 包含 {len(registry.mcp_tools)} 个 MCP 工具")


# 运行测试
test_mcp_registry()

# Agent 工具实现（子 Agent 派发）

class AgentTool:
    """
    Agent 工具：派发子 Agent
    """
    
    def __init__(self, harness, subagent_types):
        """
        初始化 Agent 工具
        
        参数:
            harness: 主 Harness 实例
            subagent_types: 子 Agent 类型映射
        """
        self.harness = harness
        self.subagent_types = subagent_types
    
    def call(self, params):
        """
        调用 Agent 工具
        
        参数:
            params: {
                "subagent_type": "code-reviewer",
                "prompt": "审查 app.ts"
            }
        
        返回:
            result: 子 Agent 执行结果
        """
        subagent_type = params.get("subagent_type")
        prompt = params.get("prompt")
        
        if not subagent_type or not prompt:
            return "错误：缺少 subagent_type 或 prompt 参数"
        
        # 1. 创建子 Agent（独立 Harness）
        sub_harness = self.create_subagent(subagent_type)
        
        # 2. 子 Agent 执行（障壁隔离）
        result = sub_harness.run(prompt, self.mock_model_api)
        
        # 3. 只返回结果（不泄露历史）
        return self.format_result(subagent_type, result)
    
    def create_subagent(self, subagent_type):
        """
        创建子 Agent
        
        参数:
            subagent_type: 子 Agent 类型
        
        返回:
            sub_harness: 子 Harness 实例
        """
        # 创建独立 Harness
        sub_harness = MiniHarnessWithPermissions()
        
        # 设置子 Agent System Prompt
        subagent_prompt = self.get_subagent_prompt(subagent_type)
        sub_harness.system_prompt = subagent_prompt
        
        return sub_harness
    
    def get_subagent_prompt(self, subagent_type):
        """
        获取子 Agent System Prompt
        
        参数:
            subagent_type: 子 Agent 类型
        
        返回:
            system_prompt: 子 Agent System Prompt
        """
        prompts = {
            "code-reviewer": """
你是代码审查 Agent。你可以使用以下工具：

工具列表：
- Read: 读取文件（审查代码）
- Grep: 搜索内容（查找模式）
- Glob: 搜索文件（查找文件）

任务：审查代码质量，生成审查报告。

输出格式：
## 审查报告
- 文件：
- 问题列表：
- 建议修改：
""",
            "tdd-guide": """
你是测试开发 Agent。你可以使用以下工具：

工具列表：
- Read: 读取代码（理解代码）
- Write: 写入测试文件
- Bash: 运行测试

任务：编写单元测试，确保覆盖率 >= 80%。

输出格式：
## 测试结果
- 测试文件：
- 覆盖率：
- 测试数量：
""",
            "doc-updater": """
你是文档生成 Agent。你可以使用以下工具：

工具列表：
- Read: 读取代码（理解代码）
- Write: 写入文档文件

任务：生成 README 和 API 文档。

输出格式：
## 文档生成完成
- README：
- API 文档：
"""
        }
        
        return prompts.get(subagent_type, "你是通用 Agent。")
    
    def format_result(self, subagent_type, result):
        """
        格式化子 Agent 结果
        
        参数:
            subagent_type: 子 Agent 类型
            result: 子 Agent 执行结果
        
        返回:
            formatted_result: 格式化后的结果
        """
        response = result["response"]
        content = response.get("content", "")
        
        return f"[{subagent_type}] {content}"
    
    def mock_model_api(self, messages):
        """模拟 Model API（子 Agent）"""
        last_message = messages[-1]["content"]
        
        if "审查" in last_message:
            return {
                "content": "## 审查报告\n\n文件：app.ts\n\n问题列表：\n1. 复杂度过高\n2. 命名不清晰\n\n建议修改：\n1. 拆分函数\n2. 改善命名",
                "stop_reason": "end_turn"
            }
        
        elif "测试" in last_message:
            return {
                "content": "## 测试结果\n\n测试文件：app.test.ts\n\n覆盖率：85%\n\n测试数量：10",
                "stop_reason": "end_turn"
            }
        
        elif "文档" in last_message:
            return {
                "content": "## 文档生成完成\n\nREADME：README.md 已创建\n\nAPI 文档：API.md 已创建",
                "stop_reason": "end_turn"
            }
        
        else:
            return {
                "content": "任务完成",
                "stop_reason": "end_turn"
            }


# 测试 Agent 工具
def test_agent_tool():
    # 创建主 Harness
    harness = MiniHarnessWithPermissions()
    
    # 创建 Agent 工具
    subagent_types = ["code-reviewer", "tdd-guide", "doc-updater"]
    agent_tool = AgentTool(harness, subagent_types)
    
    # 添加 Agent 工具到 Harness
    harness.tools["Agent"] = agent_tool.call
    
    # 测试派发子 Agent
    result = agent_tool.call({
        "subagent_type": "code-reviewer",
        "prompt": "审查 app.ts 文件"
    })
    
    print("子 Agent 结果:")
    print(result)


# 运行测试
test_agent_tool()

# Mini-Harness + MCP + 多 Agent 完整实现

class CompleteMiniHarness:
    """
    Mini-Harness 完整实现：
    - 基本工具（Read/Write/Edit/Bash/Glob/Grep）
    - 权限控制（deny/ask/allow）
    - MCP 集成（外部系统连接）
    - 多 Agent 支持（子 Agent 派发）
    """
    
    def __init__(self, permission_config=None, mcp_servers=None):
        """
        初始化
        
        参数:
            permission_config: 权限配置
            mcp_servers: MCP 服务器列表
        """
        self.system_prompt = self.default_system_prompt()
        self.tools = self.create_basic_tools()
        self.permission_system = PermissionSystem(permission_config)
        self.approval_system = ApprovalSystem()
        self.mcp_registry = MCPRegistry(self)
        self.agent_tool = None
        self.state = None
        
        # 注册 MCP 服务器
        if mcp_servers:
            for server_name, server_config in mcp_servers.items():
                self.mcp_registry.register_mcp_server(server_name, server_config)
        
        # 注册 Agent 工具
        self.register_agent_tool()
    
    def default_system_prompt(self):
        """默认 System Prompt"""
        return """
你是完整 Agent。你可以使用以下工具：

内置工具：
- Read: 读取文件
- Write: 写入文件
- Edit: 编辑文件
- Bash: 执行命令
- Glob: 搜索文件
- Grep: 搜索内容

MCP 工具：（动态加载）

Agent 工具：
- Agent: 派发子 Agent（参数：subagent_type, prompt）

约束：
- 每次只调用一个工具
- 工具调用后等待结果
- 完成任务后返回 end_turn
"""
    
    def create_basic_tools(self):
        """创建基本工具"""
        return {
            "Read": self.tool_read,
            "Write": self.tool_write,
            "Edit": self.tool_edit,
            "Bash": self.tool_bash,
            "Glob": self.tool_glob,
            "Grep": self.tool_grep
        }
    
    def register_agent_tool(self):
        """注册 Agent 工具"""
        subagent_types = ["code-reviewer", "tdd-guide", "doc-updater"]
        self.agent_tool = AgentTool(self, subagent_types)
        self.tools["Agent"] = self.agent_tool.call
    
    # 工具实现（略，与上一篇相同）
    
    def tool_read(self, params):
        """Read 工具"""
        import os
        file_path = params.get("file_path")
        if not file_path:
            return "错误：缺少 file_path"
        if not os.path.exists(file_path):
            return f"错误：文件 {file_path} 不存在"
        with open(file_path, 'r') as f:
            return f.read()
    
    def tool_write(self, params):
        """Write 工具"""
        import os
        file_path = params.get("file_path")
        content = params.get("content")
        if not file_path or not content:
            return "错误：缺少参数"
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        with open(file_path, 'w') as f:
            f.write(content)
        return f"成功写入 {file_path}"
    
    def tool_edit(self, params):
        """Edit 工具"""
        file_path = params.get("file_path")
        old_string = params.get("old_string")
        new_string = params.get("new_string")
        if not file_path or not old_string or not new_string:
            return "错误：缺少参数"
        with open(file_path, 'r') as f:
            content = f.read()
        new_content = content.replace(old_string, new_string, 1)
        with open(file_path, 'w') as f:
            f.write(new_content)
        return f"成功编辑 {file_path}"
    
    def tool_bash(self, params):
        """Bash 工具"""
        import subprocess
        command = params.get("command")
        if not command:
            return "错误：缺少 command"
        result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
        return result.stdout if result.returncode == 0 else result.stderr
    
    def tool_glob(self, params):
        """Glob 工具"""
        import glob
        pattern = params.get("pattern")
        if not pattern:
            return "错误：缺少 pattern"
        files = glob.glob(pattern, recursive=True)
        return "\n".join(files)
    
    def tool_grep(self, params):
        """Grep 工具"""
        import subprocess
        pattern = params.get("pattern")
        path = params.get("path", ".")
        if not pattern:
            return "错误：缺少 pattern"
        result = subprocess.run(f"grep -r {pattern} {path}", shell=True, capture_output=True, text=True)
        return result.stdout
    
    def initialize(self, prompt):
        """初始化"""
        self.state = {
            "messages": [
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": prompt}
            ],
            "loops": 0,
            "tool_calls": 0,
            "denied": 0,
            "approved": 0,
            "mcp_calls": 0,
            "agent_calls": 0
        }
        return self.state
    
    def agent_loop(self, model_api):
        """Agent Loop"""
        import re
        import json
        
        while True:
            messages = self.state["messages"]
            response = model_api(messages)
            stop_reason = response.get("stop_reason", "end_turn")
            
            if stop_reason == "end_turn":
                return response
            
            elif stop_reason == "tool_use":
                # 提取 tool_use
                content = response.get("content", "")
                json_match = re.search(r'\{[^}]+\}', content)
                if json_match:
                    tool_info = json.loads(json_match.group())
                    tool_name = tool_info.get("tool", "unknown")
                    tool_params = tool_info.get("params", {})
                else:
                    tool_name = "unknown"
                    tool_params = {}
                
                # 权限检查
                permission, reason = self.permission_system.check_permission(tool_name, tool_params)
                
                if permission == "deny":
                    tool_result = f"权限拒绝：{reason}"
                    self.state["denied"] += 1
                
                elif permission == "ask":
                    approved = self.approval_system.request_approval(tool_name, tool_params, reason)
                    if approved:
                        tool_result = self.execute_tool(tool_name, tool_params)
                        self.state["approved"] += 1
                    else:
                        tool_result = "用户拒绝执行"
                        self.state["denied"] += 1
                
                elif permission == "allow":
                    tool_result = self.execute_tool(tool_name, tool_params)
                
                else:
                    approved = self.approval_system.request_approval(tool_name, tool_params, {"reason": "默认审批"})
                    if approved:
                        tool_result = self.execute_tool(tool_name, tool_params)
                    else:
                        tool_result = "用户拒绝执行"
                
                # 追加结果
                self.state["messages"].append({"role": "assistant", "content": response["content"]})
                self.state["messages"].append({"role": "user", "content": tool_result})
                
                # 更新状态
                self.state["loops"] += 1
                self.state["tool_calls"] += 1
                
                if tool_name.startswith("mcp_"):
                    self.state["mcp_calls"] += 1
                elif tool_name == "Agent":
                    self.state["agent_calls"] += 1
                
                if self.state["loops"] > 20:
                    return response
    
    def execute_tool(self, tool_name, tool_params):
        """执行工具"""
        if tool_name in self.tools:
            return self.tools[tool_name](tool_params)
        else:
            return f"错误：工具 {tool_name} 不存在"
    
    def run(self, prompt, model_api):
        """完整运行"""
        self.initialize(prompt)
        response = self.agent_loop(model_api)
        
        return {
            "response": response,
            "loops": self.state["loops"],
            "tool_calls": self.state["tool_calls"],
            "denied": self.state["denied"],
            "approved": self.state["approved"],
            "mcp_calls": self.state["mcp_calls"],
            "agent_calls": self.state["agent_calls"],
            "messages": self.state["messages"]
        }


# PermissionSystem, ApprovalSystem, MCPClient, MCPRegistry, AgentTool 类（前面已定义）

# 测试完整 Mini-Harness
def test_complete_mini_harness():
    # MCP 服务器配置
    mcp_servers = {
        "filesystem": {},
        "database": {}
    }
    
    # 创建完整 Mini-Harness
    harness = CompleteMiniHarness(mcp_servers=mcp_servers)
    
    print("完整 Mini-Harness 功能:")
    print(f"- 基本工具: {len(harness.tools) - len(harness.mcp_registry.mcp_tools) - 1} 个")
    print(f"- MCP 工具: {len(harness.mcp_registry.mcp_tools)} 个")
    print(f"- Agent 工具: 1 个")
    
    # 测试运行
    prompt = "审查 app.ts 并生成文档"
    
    def mock_model_api(messages):
        last = messages[-1]["content"]
        if "审查" in last or "文档" in last:
            return {
                "content": "{\"tool\": \"Agent\", \"params\": {\"subagent_type\": \"code-reviewer\", \"prompt\": \"审查 app.ts\"}}",
                "stop_reason": "tool_use"
            }
        return {"content": "任务完成", "stop_reason": "end_turn"}
    
    result = harness.run(prompt, mock_model_api)
    
    print(f"\n运行结果:")
    print(f"- Agent 调用: {result['agent_calls']}")


# 运行测试
test_complete_mini_harness()