Learn Claude Code

# =============================================================================
# Agent 循环 - 这是一切的核心
# =============================================================================
def agent_loop(messages: list) -> list:
    """
    完整的 Agent 在一个函数里。

    所有编程 Agent 共享的模式：
        while True:
            response = model(messages, tools)
            if no tool calls: return
            execute tools, append results, continue
    """
    while True:
        # 1. 调用模型
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            messages=messages,
            tools=TOOLS,
            max_tokens=8000,
        )

        # 2. 打印文本输出
        tool_calls = []
        for block in response.content:
            if hasattr(block, "text"):
                print(block.text)
            if block.type == "tool_use":
                tool_calls.append(block)

        # 3. 如果没有工具调用，任务完成
        if response.stop_reason != "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            return messages

        # 4. 执行工具，收集结果
        results = []
        for tc in tool_calls:
            print(f"\n> {tc.name}: {tc.input}")
            output = execute_tool(tc.name, tc.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": tc.id,
                "content": output,
            })

        # 5. 追加到对话，继续循环
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": results})


# 工具实现示例：安全的文件读取
def safe_path(p: str) -> Path:
    """确保路径在工作区内（安全措施）"""
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"Path escapes workspace: {p}")
    return path

def run_read(path: str, limit: int = None) -> str:
    """读取文件内容，可选行数限制"""
    try:
        text = safe_path(path).read_text()
        lines = text.splitlines()
        if limit and limit < len(lines):
            lines = lines[:limit]
            lines.append(f"... ({len(text.splitlines()) - limit} more lines)")
        return "\n".join(lines)[:50000]
    except Exception as e:
        return f"Error: {e}"

def run_edit(path: str, old_text: str, new_text: str) -> str:
    """精确替换文件中的文本"""
    try:
        fp = safe_path(path)
        content = fp.read_text()
        if old_text not in content:
            return f"Error: Text not found in {path}"
        # 只替换第一个匹配（安全）
        new_content = content.replace(old_text, new_text, 1)
        fp.write_text(new_content)
        return f"Edited {path}"
    except Exception as e:
        return f"Error: {e}"

class TodoManager:
    """带约束的任务列表管理器"""

    def __init__(self):
        self.items = []

    def update(self, items: list) -> str:
        """验证并更新任务列表"""
        validated = []
        in_progress_count = 0

        for i, item in enumerate(items):
            content = str(item.get("content", "")).strip()
            status = str(item.get("status", "pending")).lower()
            active_form = str(item.get("activeForm", "")).strip()

            # 验证检查
            if not content:
                raise ValueError(f"Item {i}: content required")
            if status not in ("pending", "in_progress", "completed"):
                raise ValueError(f"Item {i}: invalid status '{status}'")
            if not active_form:
                raise ValueError(f"Item {i}: activeForm required")

            if status == "in_progress":
                in_progress_count += 1

            validated.append({
                "content": content,
                "status": status,
                "activeForm": active_form
            })

        # 强制约束
        if len(validated) > 20:
            raise ValueError("Max 20 todos allowed")
        if in_progress_count > 1:
            raise ValueError("Only one task can be in_progress at a time")

        self.items = validated
        return self.render()

    def render(self) -> str:
        """渲染为人类可读的文本"""
        if not self.items:
            return "No todos."

        lines = []
        for item in self.items:
            if item["status"] == "completed":
                lines.append(f"[x] {item['content']}")
            elif item["status"] == "in_progress":
                lines.append(f"[>] {item['content']} <- {item['activeForm']}")
            else:
                lines.append(f"[ ] {item['content']}")

        completed = sum(1 for t in self.items if t["status"] == "completed")
        lines.append(f"\n({completed}/{len(self.items)} completed)")

        return "\n".join(lines)


# TodoWrite 工具定义
TODO_TOOL = {
    "name": "TodoWrite",
    "description": "Update the task list. Use to plan and track progress.",
    "input_schema": {
        "type": "object",
        "properties": {
            "items": {
                "type": "array",
                "description": "Complete list of tasks (replaces existing)",
                "items": {
                    "type": "object",
                    "properties": {
                        "content": {"type": "string", "description": "Task description"},
                        "status": {
                            "type": "string",
                            "enum": ["pending", "in_progress", "completed"]
                        },
                        "activeForm": {
                            "type": "string",
                            "description": "Present tense action, e.g. 'Reading files'"
                        },
                    },
                    "required": ["content", "status", "activeForm"],
                },
            }
        },
        "required": ["items"],
    },
}


# 系统提醒（软约束）
INITIAL_REMINDER = "Use TodoWrite for multi-step tasks."
NAG_REMINDER = "10+ turns without todo update. Please update todos."

# 代理类型注册表
AGENT_TYPES = {
    "explore": {
        "description": "Read-only agent for exploring code",
        "tools": ["bash", "read_file"],  # 不能写
        "prompt": "Search and analyze. Return concise summary."
    },
    "code": {
        "description": "Full agent for implementing features",
        "tools": "*",  # 所有工具
        "prompt": "Implement the requested changes efficiently."
    },
    "plan": {
        "description": "Planning agent for design",
        "tools": ["bash", "read_file"],  # 只读
        "prompt": "Output a numbered plan. Do NOT make changes."
    }
}


def run_task(description: str, prompt: str, agent_type: str) -> str:
    """
    执行子代理任务，带隔离的上下文。

    这是子代理机制的核心：
    1. 创建隔离的消息历史（关键：没有父上下文！）
    2. 使用代理特定的系统提示词
    3. 根据代理类型过滤可用工具
    4. 运行同样的查询循环
    5. 只返回最终文本（不是中间细节）
    """
    if agent_type not in AGENT_TYPES:
        return f"Error: Unknown agent type '{agent_type}'"

    config = AGENT_TYPES[agent_type]

    # 代理特定的系统提示词
    sub_system = f"""You are a {agent_type} subagent at {WORKDIR}.
{config["prompt"]}
Complete the task and return a clear, concise summary."""

    # 过滤后的工具
    sub_tools = get_tools_for_agent(agent_type)

    # 隔离的消息历史 - 这是关键！
    # 子代理从全新的历史开始，不会看到父代理的对话
    sub_messages = [{"role": "user", "content": prompt}]

    # 进度显示
    print(f"  [{agent_type}] {description}")
    start = time.time()
    tool_count = 0

    # 运行同样的 Agent 循环（静默执行）
    while True:
        response = client.messages.create(
            model=MODEL,
            system=sub_system,
            messages=sub_messages,
            tools=sub_tools,
            max_tokens=8000,
        )

        if response.stop_reason != "tool_use":
            break

        tool_calls = [b for b in response.content if b.type == "tool_use"]
        results = []

        for tc in tool_calls:
            tool_count += 1
            output = execute_tool(tc.name, tc.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": tc.id,
                "content": output
            })

        sub_messages.append({"role": "assistant", "content": response.content})
        sub_messages.append({"role": "user", "content": results})

    # 只返回最终文本 - 父代理看到简洁的摘要
    elapsed = time.time() - start
    print(f"  [{agent_type}] {description} - done ({tool_count} tools, {elapsed:.1f}s)")

    for block in response.content:
        if hasattr(block, "text"):
            return block.text

    return "(subagent returned no text)"

class SkillLoader:
    """加载和管理 SKILL.md 文件"""

    def __init__(self, skills_dir: Path):
        self.skills_dir = skills_dir
        self.skills = {}
        self.load_skills()

    def parse_skill_md(self, path: Path) -> dict:
        """解析 YAML 前置 + Markdown 正文"""
        content = path.read_text()
        match = re.match(r'^---\s*\n(.*?)\n---\s*\n(.*)$', content, re.DOTALL)
        if not match:
            return None

        frontmatter, body = match.groups()
        metadata = {}
        for line in frontmatter.strip().split("\n"):
            if ":" in line:
                key, value = line.split(":", 1)
                metadata[key.strip()] = value.strip().strip("\"'")

        return {
            "name": metadata["name"],
            "description": metadata["description"],
            "body": body.strip(),
            "path": path,
            "dir": path.parent,
        }

    def get_descriptions(self) -> str:
        """Layer 1: 只返回元数据（放入 system prompt）"""
        return "\n".join(
            f"- {name}: {skill['description']}"
            for name, skill in self.skills.items()
        )

    def get_skill_content(self, name: str) -> str:
        """Layer 2: 返回完整内容（触发时注入）"""
        skill = self.skills[name]
        content = f"# Skill: {skill['name']}\n\n{skill['body']}"

        # Layer 3: 列出可用资源
        resources = []
        for folder in ["scripts", "references", "assets"]:
            folder_path = skill["dir"] / folder
            if folder_path.exists():
                files = list(folder_path.glob("*"))
                if files:
                    resources.append(f"{folder}: {', '.join(f.name for f in files)}")

        if resources:
            content += f"\n\n**Available resources:**\n" + "\n".join(f"- {r}" for r in resources)

        return content


def run_skill(skill_name: str) -> str:
    """
    加载 Skill 并注入到对话中。

    关键机制：
    1. 获取 skill 内容（SKILL.md 主体 + 资源提示）
    2. 包装在  标签中返回
    3. 模型作为 tool_result 接收（user message）
    4. 模型现在"知道"如何完成任务

    为什么用 tool_result 而不是 system prompt？
    - System prompt 变化会使缓存失效（20-50x 成本增加）
    - Tool results 追加到末尾（前缀不变，缓存命中）
    """
    content = SKILLS.get_skill_content(skill_name)

    if content is None:
        available = ", ".join(SKILLS.list_skills()) or "none"
        return f"Error: Unknown skill '{skill_name}'. Available: {available}"

    # 包装在标签中，让模型知道这是 skill 内容
    return f"""
{content}


Follow the instructions in the skill above to complete the user's task."""


# SKILL.md 示例
"""
---
name: pdf
description: Process PDF files. Use when reading, creating, or merging PDFs.
---

# PDF Processing Skill

## Reading PDFs

Use pdftotext for quick extraction:
```bash
pdftotext input.pdf -
```

For complex layouts, use PyMuPDF:
```python
import fitz
doc = fitz.open("input.pdf")
for page in doc:
    text = page.get_text()
```
"""