子代理：分工协作处理复杂任务

from claude import Agent
 
parent = Agent()
 
# 创建子代理
auth_agent = parent.create_subagent(
    name="auth-refactor",
    instructions="专注重构认证模块，保持向后兼容",
    files=["src/auth/*", "tests/test_auth.py"]
)
 
# 并行执行
result = auth_agent.run("将 session 认证改为 JWT")

# 认证模块重构规范
 
## 约束
- 不修改数据库 schema
- 保持现有 API 接口不变
- 所有修改必须有对应测试
 
## 检查清单
- [ ] 单元测试通过率 100%
- [ ] 集成测试通过
- [ ] 性能不下降

subagent = parent.create_subagent(
    name="safe-refactor",
    mode="ask",           # 子代理必须询问确认
    allowed_commands=["pytest", "black", "git diff"],  # 只允许这些命令
    denied_commands=["git push", "rm -rf", "docker"]
)

# workflow.py
workflow = {
    "steps": [
        {
            "agent": "analyzer",
            "task": "分析代码库，确定需要重构的模块",
            "output": "refactor_plan.json"
        },
        {
            "parallel": [
                {"agent": "refactor-a", "input": "refactor_plan.json", "filter": "module=A"},
                {"agent": "refactor-b", "input": "refactor_plan.json", "filter": "module=B"},
            ]
        },
        {
            "agent": "integrator",
            "task": "合并所有重构结果，解决冲突",
            "depends": ["refactor-a", "refactor-b"]
        },
        {
            "agent": "tester",
            "task": "运行全量测试，验证重构",
            "depends": ["integrator"]
        }
    ]
}

# refactor_workflow.py
from claude import Agent
import asyncio
 
async def main():
    parent = Agent()
    
    # 创建三个并行的子代理
    auth_agent = parent.create_subagent(
        name="auth-refactor",
        instructions="""
        你负责重构认证模块。约束：
        - 不修改数据库 schema
        - 保持现有 API 接口不变
        - 所有修改必须有对应测试
        - 使用 JWT 替代 session，但保留原有 session 的兼容层
        """,
        files=["src/auth/*", "tests/test_auth.py", "src/models/user.py"],
        mode="auto-edits",  # 自动编辑，但限制在指定文件
        timeout_minutes=30
    )
    
    api_agent = parent.create_subagent(
        name="api-adapter",
        instructions="""
        你负责更新 API 层以适配新的认证方式。
        约束：
        - 所有接口的 request/response 格式不变
        - 在 header 中支持 Authorization: Bearer <token>
        - 同时兼容旧的 session cookie（双轨运行 1 个版本）
        """,
        files=["src/routes/*", "src/middleware/auth.py", "tests/test_api.py"],
        mode="auto-edits",
        timeout_minutes=30
    )
    
    test_agent = parent.create_subagent(
        name="test-updater",
        instructions="""
        你负责更新测试用例以覆盖新的 JWT 认证逻辑。
        要求：
        - 原有测试必须继续通过（兼容层保证）
        - 新增 JWT 专门的测试场景
        - 测试覆盖率不下降
        """,
        files=["tests/*", "src/auth/jwt.py"],
        mode="auto-edits",
        timeout_minutes=20
    )
    
    # 并行执行
    results = await asyncio.gather(
        auth_agent.run("将 session 认证核心改为 JWT，保留兼容层"),
        api_agent.run("在 API 路由中支持 JWT Bearer Token"),
        test_agent.run("补充 JWT 测试并确保旧测试通过"),
        return_exceptions=True
    )
    
    # 检查结果
    for agent, result in zip([auth_agent, api_agent, test_agent], results):
        if isinstance(result, Exception):
            print(f"❌ {agent.name} 失败: {result}")
        else:
            print(f"✅ {agent.name} 完成")
    
    # 合并前审查
    print("\n--- 变更摘要 ---")
    for agent in [auth_agent, api_agent, test_agent]:
        diff = agent.get_diff()
        print(f"\n{agent.name}:\n{diff[:500]}...")
    
    # 用户确认后合并（或手动处理冲突）
    # parent.merge_subagent(auth_agent)
    # parent.merge_subagent(api_agent)
    # parent.merge_subagent(test_agent)
 
if __name__ == "__main__":
    asyncio.run(main())

# 子代理任务规范：代码重构
 
## 任务信息
- 目标模块：{{module}}
- 重构类型：{{type}}
- 预计影响文件数：{{affected_files}}
 
## 强制约束
- [ ] 不修改数据库 schema（除非任务描述明确允许）
- [ ] 所有公开 API 保持向后兼容
- [ ] 修改前先运行测试，记录基线通过率
- [ ] 修改后测试通过率必须 >= 基线
- [ ] 单函数行数不超过 50 行（重构后）
 
## 工作流程
1. 读取相关文件，理解当前逻辑
2. 编写/更新测试用例（先写测试再重构）
3. 执行重构
4. 运行全量测试
5. 生成变更摘要（改了哪些文件、为什么改、风险点）
 
## 输出格式

subagent = parent.create_subagent(
    name="order-refactor",
    instructions_template=".claude/subagent-templates/refactor.md",
    template_vars={
        "module": "订单模块",
        "type": "拆分 fat service",
        "affected_files": "5-8"
    }
)

{
  "workflow": "大型重构流水线",
  "steps": [
    {
      "id": "analyze",
      "agent": "analyzer",
      "type": "sequential",
      "task": "分析 src/monolith.py，输出重构计划 refactor_plan.json",
      "output_files": ["refactor_plan.json"],
      "timeout": 600
    },
    {
      "id": "refactor-parallel",
      "type": "parallel",
      "branches": [
        {
          "id": "module-a",
          "agent": "refactor-a",
          "task": "按 refactor_plan.json 重构模块 A",
          "input": "refactor_plan.json",
          "filter": "module=A",
          "files": ["src/module_a/*"],
          "timeout": 900
        },
        {
          "id": "module-b",
          "agent": "refactor-b",
          "task": "按 refactor_plan.json 重构模块 B",
          "input": "refactor_plan.json",
          "filter": "module=B",
          "files": ["src/module_b/*"],
          "timeout": 900
        },
        {
          "id": "module-c",
          "agent": "refactor-c",
          "task": "按 refactor_plan.json 重构模块 C",
          "input": "refactor_plan.json",
          "filter": "module=C",
          "files": ["src/module_c/*"],
          "timeout": 900
        }
      ]
    },
    {
      "id": "integrate",
      "agent": "integrator",
      "type": "sequential",
      "depends_on": ["module-a", "module-b", "module-c"],
      "task": "合并所有重构结果，解决模块间接口冲突",
      "timeout": 600
    },
    {
      "id": "test",
      "agent": "tester",
      "type": "sequential",
      "depends_on": ["integrate"],
      "task": "运行全量测试 + 性能基准测试，输出测试报告",
      "timeout": 1200
    }
  ]
}