深入 Agent Harness：概念解析與實作指南

TOOLS = [
    {
        "name": "read_file",
        "description": "Read the contents of a file",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "File path"}
            },
            "required": ["path"]
        },
        "_safety": "read_only"
    },
    {
        "name": "bash",
        "description": "Execute a bash command",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {"type": "string"}
            },
            "required": ["command"]
        },
        "_safety": "dangerous"
    },
    # ... write_file, list_files, dispatch_agent ...
]

ReadOnly         ──►  讀檔、查詢、搜尋（自動允許）
WorkspaceWrite   ──►  在 workspace 內寫檔（可設定自動）
DangerFullAccess ──►  全系統存取（需明確授權）
Prompt           ──►  每次都詢問使用者
Allow            ──►  使用者已預先批准

if current_level >= required_level:
    allow()
elif one_level_gap:
    ask_user()
else:
    deny()

PreToolUse      ──► 攔截 tool 呼叫、加 policy check
PostToolUse     ──► 記錄 audit log
PreCompaction   ──► 在壓縮前 flush 狀態到檔案
SessionStart    ──► 載入專案特定設定
SessionEnd      ──► 寫入摘要、計算 cost

pip install anthropic
export ANTHROPIC_API_KEY="your-key-here"

"""
Minimal Agent Harness - 一個可運行的 agent harness
展示了 agent loop、tool system、permission gate、context 管理的核心概念
"""

import json
import os
import subprocess
from pathlib import Path
from anthropic import Anthropic

TOOLS = [
    {
        "name": "read_file",
        "description": "Read the contents of a file",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "File path"}
            },
            "required": ["path"]
        },
        "_safety": "read_only"
    },
    {
        "name": "write_file",
        "description": "Write content to a file (creates or overwrites)",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"]
        },
        "_safety": "workspace_write"
    },
    {
        "name": "bash",
        "description": "Execute a bash command",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {"type": "string"}
            },
            "required": ["command"]
        },
        "_safety": "dangerous"
    },
    {
        "name": "list_files",
        "description": "List files in a directory",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "default": "."}
            }
        },
        "_safety": "read_only"
    }
]

class ToolExecutor:
    def __init__(self, workspace: Path):
        self.workspace = workspace.resolve()

    def _check_in_workspace(self, path: str) -> Path:
        """確保路徑在 workspace 內，防止 path traversal 攻擊"""
        full_path = (self.workspace / path).resolve()
        if not str(full_path).startswith(str(self.workspace)):
            raise ValueError(f"Path {path} is outside workspace")
        return full_path

    def execute(self, tool_name: str, tool_input: dict) -> str:
        try:
            if tool_name == "read_file":
                path = self._check_in_workspace(tool_input["path"])
                return path.read_text()

            elif tool_name == "write_file":
                path = self._check_in_workspace(tool_input["path"])
                path.parent.mkdir(parents=True, exist_ok=True)
                path.write_text(tool_input["content"])
                return f"Wrote {len(tool_input['content'])} bytes to {path}"

            elif tool_name == "bash":
                result = subprocess.run(
                    tool_input["command"],
                    shell=True,
                    capture_output=True,
                    text=True,
                    cwd=self.workspace,
                    timeout=30
                )
                output = result.stdout + result.stderr
                # 截斷大 output（避免 context 爆炸）
                if len(output) > 5000:
                    output = output[:5000] + "\n... [truncated]"
                return output

            elif tool_name == "list_files":
                path = self._check_in_workspace(tool_input.get("path", "."))
                return "\n".join(str(p.relative_to(self.workspace))
                                 for p in path.iterdir())
            else:
                return f"Unknown tool: {tool_name}"

        except Exception as e:
            return f"Error: {type(e).__name__}: {e}"

class PermissionGate:
    def __init__(self, mode: str = "ask"):
        # mode: "auto" (read_only auto-allow), "ask" (always prompt), "yolo" (allow all)
        self.mode = mode
        self.approved_tools = set()

    def check(self, tool_name: str, tool_input: dict, safety: str) -> bool:
        if self.mode == "yolo":
            return True

        if safety == "read_only" and self.mode == "auto":
            return True

        if tool_name in self.approved_tools:
            return True

        # Ask user
        print(f"\n🔒 Tool request: {tool_name}({json.dumps(tool_input)})")
        print(f"   Safety level: {safety}")
        choice = input("   Allow? [y/n/a (always allow this tool)]: ").strip().lower()

        if choice == "a":
            self.approved_tools.add(tool_name)
            return True
        return choice == "y"

class ContextManager:
    def __init__(self, client: Anthropic, max_tokens: int = 100_000):
        self.client = client
        self.max_tokens = max_tokens
        self.threshold = int(max_tokens * 0.85)  # 85% 觸發壓縮

    def estimate_tokens(self, messages: list) -> int:
        """粗略估算 token 數（實際應用 tokenizer）"""
        text = json.dumps(messages)
        return len(text) // 4  # 約 4 chars per token

    def maybe_compact(self, messages: list, system: str) -> list:
        """超過 threshold 就壓縮"""
        if self.estimate_tokens(messages) < self.threshold:
            return messages

        print("\n📦 Compacting context...")

        # 保留最近 5 個訊息，壓縮其餘
        recent = messages[-5:]
        to_compact = messages[:-5]

        if not to_compact:
            return messages

        summary_prompt = (
            "Summarize the following conversation history concisely, "
            "preserving key decisions, file paths, and intermediate results:\n\n"
            + json.dumps(to_compact, ensure_ascii=False)
        )

        response = self.client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=2000,
            messages=[{"role": "user", "content": summary_prompt}]
        )

        summary = response.content[0].text
        return [
            {"role": "user", "content": f"[Previous conversation summary]\n{summary}"}
        ] + recent

class AgentClient:
    def __init__(self, workspace: str = ".", permission_mode: str = "auto"):
        self.client = Anthropic()
        self.workspace = Path(workspace)
        self.executor = ToolExecutor(self.workspace)
        self.permission = PermissionGate(mode=permission_mode)
        self.context_mgr = ContextManager(self.client)
        self.messages = []
        self.system_prompt = self._build_system_prompt()

    def _build_system_prompt(self) -> str:
        base = (
            "You are a helpful coding agent running in a terminal. "
            "Use the available tools to accomplish the user's tasks. "
            "Be concise. When done, respond with plain text (no tool calls)."
        )
        memory_file = self.workspace / "CLAUDE.md"
        if memory_file.exists():
            base += f"\n\n# Project Memory (from CLAUDE.md)\n{memory_file.read_text()}"
        return base

    def _tools_for_api(self) -> list:
        """剝掉 _safety 欄位，因為 Anthropic API 不接受"""
        return [
            {k: v for k, v in tool.items() if not k.startswith("_")}
            for tool in TOOLS
        ]

    def _get_tool_safety(self, name: str) -> str:
        for t in TOOLS:
            if t["name"] == name:
                return t.get("_safety", "dangerous")
        return "dangerous"

    def run(self, user_prompt: str, max_turns: int = 20):
        """主 agent loop"""
        self.messages.append({"role": "user", "content": user_prompt})

        for turn in range(max_turns):
            # 1. Compact if needed
            self.messages = self.context_mgr.maybe_compact(
                self.messages, self.system_prompt
            )

            # 2. Call LLM
            response = self.client.messages.create(
                model="claude-sonnet-4-5",
                max_tokens=4096,
                system=self.system_prompt,
                tools=self._tools_for_api(),
                messages=self.messages
            )

            # 3. Add assistant message to history
            self.messages.append({
                "role": "assistant",
                "content": response.content
            })

            # 4. Check stop reason
            if response.stop_reason == "end_turn":
                for block in response.content:
                    if hasattr(block, "text"):
                        print(f"\n🤖 {block.text}")
                return

            # 5. Process tool calls
            if response.stop_reason == "tool_use":
                tool_results = []
                for block in response.content:
                    if hasattr(block, "text") and block.text:
                        print(f"\n💭 {block.text}")

                    if block.type == "tool_use":
                        safety = self._get_tool_safety(block.name)

                        allowed = self.permission.check(
                            block.name, block.input, safety
                        )

                        if not allowed:
                            result = "User denied this tool call."
                        else:
                            print(f"\n🔧 {block.name}({json.dumps(block.input)[:80]}...)")
                            result = self.executor.execute(block.name, block.input)
                            print(f"   ↳ {result[:200]}...")

                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result
                        })

                self.messages.append({
                    "role": "user",
                    "content": tool_results
                })

        print(f"\n⚠️  Reached max_turns ({max_turns}). Stopping.")

def main():
    print("🚀 Mini Agent Client (type 'exit' to quit)")
    print(f"   Workspace: {os.getcwd()}")
    print(f"   Permission mode: auto (read_only auto-allow, others ask)\n")

    agent = AgentClient(workspace=".", permission_mode="auto")

    while True:
        try:
            user_input = input("\n👤 You: ").strip()
            if not user_input or user_input.lower() in ("exit", "quit"):
                break
            agent.run(user_input)
        except KeyboardInterrupt:
            print("\n\n👋 Interrupted. Bye!")
            break

if __name__ == "__main__":
    main()

mkdir agent-test && cd agent-test
echo "# My Project" > README.md
python agent.py

# 試試這些 prompt：
# - "What files are in this directory?"
# - "Create a hello.py that prints 'Hello, Agent World!'"
# - "Run the hello.py and show me the output"
# - "Refactor hello.py to accept a name argument"

# 在 TOOLS 加上 dispatch_agent
{
    "name": "dispatch_agent",
    "description": (
        "Dispatch a sub-agent to handle a focused sub-task in an isolated "
        "context window. Use for broad searches or exploring alternatives. "
        "The sub-agent returns a final report; its conversation is discarded."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "task": {"type": "string", "description": "Detailed task description"}
        },
        "required": ["task"]
    },
    "_safety": "read_only"
}

# 在 ToolExecutor 加上對應處理
def _dispatch_sub_agent(self, task: str) -> str:
    """開一個新的 agent，限制只能用 read-only tool"""
    sub_agent = AgentClient(
        workspace=str(self.workspace),
        permission_mode="auto"
    )
    sub_agent._allowed_tools = [
        t["name"] for t in TOOLS if t.get("_safety") == "read_only"
    ]
    sub_agent.run(task, max_turns=10)

    for msg in reversed(sub_agent.messages):
        if msg["role"] == "assistant":
            for block in msg["content"]:
                if hasattr(block, "text"):
                    return f"Sub-agent report:\n{block.text}"
    return "Sub-agent finished with no report."

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MCPIntegration:
    def __init__(self, server_command: list):
        self.params = StdioServerParameters(
            command=server_command[0],
            args=server_command[1:]
        )

    async def list_tools(self) -> list:
        """從 MCP server 拿到 tool list，併入 agent 的 TOOLS"""
        async with stdio_client(self.params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                tools = await session.list_tools()
                return [
                    {
                        "name": f"mcp__{t.name}",
                        "description": t.description,
                        "input_schema": t.inputSchema,
                        "_safety": "ask"  # 預設詢問
                    }
                    for t in tools.tools
                ]