WorkBuddy 工程化实战：从 Agent Loop 到 Harness 的分层解耦

# 最简 Agent Loop —— 能跑，但不可控
def naive_agent_loop(task, tools, max_steps=20):
    messages = [{"role": "user", "content": task}]
    for _ in range(max_steps):
        resp = llm.chat(messages, tools=tools)
        if resp.tool_calls:
            for tc in resp.tool_calls:
                result = execute_tool(tc)  # 无权限检查
                messages.append({"role": "tool", "content": result})
        else:
            return resp.content
    return "达到最大步数，任务未完成"  # 无中间状态保存

from dataclasses import dataclass, field
from enum import Enum

class PermissionMode(Enum):
    READ_ONLY = "read_only"       # 只读，禁止写入
    WORKSPACE_WRITE = "workspace" # 仅工作空间可写
    FULL_ACCESS = "full"          # 无限制

@dataclass
class HarnessLoop:
    """带 Harness 约束的 Agent Loop"""
    tools: dict
    permission: PermissionMode = PermissionMode.WORKSPACE_WRITE
    max_steps: int = 20
    state: dict = field(default_factory=dict)

    def _check_permission(self, tool_name: str, args: dict) -> bool:
        """权限边界检查"""
        if self.permission == PermissionMode.READ_ONLY:
            return tool_name in ("read_file", "search", "glob")
        if self.permission == PermissionMode.WORKSPACE_WRITE:
            path = args.get("path", "")
            return not path.startswith("/etc") and not path.startswith("C:\\Windows")
        return True

    def _trim_context(self, messages: list, max_tokens: int = 8000) -> list:
        """上下文裁剪：只保留最近 N 轮 + 系统提示词"""
        system = [m for m in messages if m["role"] == "system"]
        recent = [m for m in messages if m["role"] != "system"][-6:]
        return system + recent

    def run(self, task: str) -> str:
        messages = [{"role": "system", "content": "你是 WorkBuddy"},
                    {"role": "user", "content": task}]
        for step in range(self.max_steps):
            trimmed = self._trim_context(messages)
            resp = llm.chat(trimmed, tools=list(self.tools.keys()))
            if not resp.tool_calls:
                return resp.content
            for tc in resp.tool_calls:
                if not self._check_permission(tc.name, tc.args):
                    messages.append({"role": "tool",
                                     "content": "权限拒绝：当前模式不允许此操作"})
                    continue
                result = self.tools[tc.name](**tc.args)
                self.state[f"step_{step}_{tc.name}"] = result  # 状态持久化
                messages.append({"role": "tool", "content": str(result)})
        return "任务未在限定步数内完成"

from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx

server = Server("order-query")
ORDER_API = "http://internal.example.com/api/orders"

@server.list_tools()
async def list_tools():
    return [Tool(
        name="query_order",
        description="根据订单号查询订单状态、金额和物流信息",
        inputSchema={
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "订单号"}
            },
            "required": ["order_id"]
        }
    )]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_order":
        async with httpx.AsyncClient() as c:
            resp = await c.get(f"{ORDER_API}/{arguments['order_id']}",
                               timeout=10)
            data = resp.json()
        return [TextContent(
            type="text",
            text=f"状态：{data['status']}，金额：{data['amount']}元，"
                 f"当前位置：{data['location']}")]

if __name__ == "__main__":
    import mcp.server.stdio
    mcp.server.stdio.run(server)

{
  "mcpServers": {
    "order-query": {
      "command": "python",
      "args": ["/path/to/order_server.py"],
      "env": {}
    }
  }
}