当AI开始带团队：Claude Code 的 Team Agent 编程方式彻底改变了我写代码的方式

你有没有想过，编程不再是你一个人对着 AI 聊天，而是你是 CEO，AI 是你的整个工程团队？

这不是科幻。这是 Claude Agent SDK 的 Team Agent 编程模式，现在就能用。

先说说你现在在做什么

你的工作流大概是这样：

打开终端 → 问 Claude 一个问题 → 等回答 → 复制代码 → 继续问。

一来一往，像审问犯人。

问题在哪？这个模式本质上是串行的。你问一句，它答一句，你每次只用了 AI 的 1/10 的脑子。

而 Team Agent 模式完全反过来：你扔一个任务，一群 AI 同时干活。

什么是 Team Agent 模式？

Claude Agent SDK 的 Team Agent，简单说就是 一个 Agent 指挥一群 Agent。

技术上叫 orchestrator-subagent 架构：

你
│
└── Orchestrator（主 Agent）
      ├── Subagent A → 负责写测试
      ├── Subagent B → 负责查文档
      └── Subagent C → 负责修 Bug

主 Agent 理解你的意图，分解任务，把子任务派给不同的子 Agent 并行执行，最后汇总结果。

这不是比喻，这是真实的代码执行流程。

实战：一个真实的例子

假设你要做这件事：“把这个 Express 应用迁移到 TypeScript，顺便补全单元测试，然后更新 README”

串行模式（你现在的方式）：

问 Claude 怎么配置 tsconfig → 等 → 复制
问怎么改 app.js → 等 → 改
继续改其他文件…
写测试…
写 README…

几个来回下来，你已经精疲力竭。

Team Agent 模式：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# 安装：pip install claude-agent-sdk
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Edit", "Bash", "Glob"],
        max_turns=50,
    )

    async for message in query(
        prompt="""
        将当前 Express 项目迁移到 TypeScript：
        1. 配置 TypeScript 环境
        2. 转换所有 .js 文件为 .ts，保持功能不变
        3. 为核心模块编写单元测试（覆盖率 > 80%）
        4. 更新 README，记录新的构建流程

        这四个任务可以并行执行，互不干扰的部分请同时进行。
        """,
        options=options,
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

Claude Agent SDK 收到这个任务后：

自动识别哪些步骤可以并行
按依赖关系编排执行顺序
每个工具调用形成完整的决策-执行循环

你只需要描述目标，执行细节交给 Agent 安排。

核心机制：Agent Loop

理解 Team Agent 之前，先搞清楚单个 Agent 怎么工作的。

每个 Claude Agent 都在跑一个循环：

接收任务
    │
    ▼
思考下一步（Claude 判断）
    │
    ├──→ 调工具（Bash、Read、Write...）→ 拿结果
    │         └──→ 回到"思考下一步"
    │
    └──→ 输出最终答案（没有工具调用了）→ 结束

关键点：Claude 自己决定调几次工具，循环几轮。你只需要告诉它目标，剩下的它自己安排。

每一轮有多少 turn 可以用 max_turns 控制；烧多少钱可以用 max_budget_usd 控制。

实战代码：完整的 Team Agent 示例

下面是一个实际可用的 Python 示例，展示如何用 Claude Agent SDK 跑一个多 Agent 工作流：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
import asyncio
import json
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def run_single_agent(task: str, model: str = "claude-sonnet-4-6") -> str:
    """运行单个子 Agent，返回最终结果"""
    options = ClaudeAgentOptions(
        model=model,  # 支持按任务选择不同模型（Opus/Sonnet/Haiku）
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        max_turns=10,
        max_budget_usd=0.50,  # 单个子 Agent 最多花 50 美分
    )

    async for message in query(prompt=task, options=options):
        if isinstance(message, ResultMessage):
            return message.result or ""
    return ""


async def run_parallel_agents(tasks: list[str]) -> list[str]:
    """并行运行多个子 Agent"""
    # query() 本身是 async generator，直接用 asyncio.gather 并行
    results = await asyncio.gather(
        *[run_single_agent(task) for task in tasks]
    )
    return list(results)


async def orchestrate_project(project_description: str):
    """主编排 Agent：分解任务 → 并行执行 → 汇总"""

    # Step 1: 主 Agent（Opus）分解任务
    decompose_task = f"""
    你是项目经理。分析以下需求，将其分解为可并行执行的独立子任务。
    每个子任务应该：
    1. 明确、具体、可独立完成
    2. 相互之间没有强依赖（可以同时进行）
    3. 只输出 JSON，格式：{{"tasks": ["任务1", "任务2", ...]}}

    需求：{project_description}
    """

    decompose_result = await run_single_agent(decompose_task)
    tasks_data = json.loads(decompose_result)
    tasks = tasks_data["tasks"]

    print(f"📋 任务分解完成，共 {len(tasks)} 个子任务，开始并行执行...")
    for i, task in enumerate(tasks):
        print(f"  [{i+1}] {task[:60]}...")

    # Step 2: 并行执行所有子任务（Sonnet，性价比高）
    results = await run_parallel_agents(tasks)

    # Step 3: 汇总 Agent（Opus）整合结果
    summary_task = f"""
    以下是各个子任务的执行结果，请整合成一份完整的项目报告：

    {chr(10).join([f"任务{i+1}结果：{r}" for i, r in enumerate(results)])}

    生成最终总结，包括：完成的工作、遇到的问题、后续建议。
    """
    final_result = await run_single_agent(summary_task)
    return final_result


# 使用示例
if __name__ == "__main__":
    result = asyncio.run(orchestrate_project(
        "重构 src/ 目录下的 API 层，添加错误处理，更新单元测试，并生成 API 文档"
    ))
    print("\n✅ 项目完成！")
    print(result)

进阶：用 Hooks 守住你的代码边界

多 Agent 并行有个隐患：你不知道哪个 Agent 在什么时候碰了哪个文件。

Hooks 就是你的安全网——在每次工具调用前后插入检查逻辑，拦危险操作、记操作日志，一行代码都不放过。

注意：不是继承类，而是注册回调函数：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import asyncio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    ClaudeSDKClient,
    HookMatcher,
    AssistantMessage,
    ResultMessage,
)


# Hook 回调函数：保护 .env 文件不被修改
async def protect_env_files(input_data, tool_use_id, context):
    file_path = input_data["tool_input"].get("file_path", "")
    file_name = file_path.split("/")[-1]

    if file_name == ".env":
        print(f"⚠️  拦截：禁止修改 .env 文件")
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Cannot modify .env files",
            }
        }
    return {}  # 空对象 = 放行


# Hook 回调函数：记录所有文件写入操作
async def log_file_writes(input_data, tool_use_id, context):
    file_path = input_data["tool_input"].get("file_path", "unknown")
    print(f"📝 Agent 正在写入：{file_path}")
    return {}  # 只记录，不拦截


async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        hooks={
            # PreToolUse：在工具执行前触发
            "PreToolUse": [
                HookMatcher(matcher="Write|Edit", hooks=[protect_env_files, log_file_writes])
            ],
            # PostToolUse：在工具执行后触发
            "PostToolUse": [],
        },
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("修复 src/ 目录下所有 TypeScript 类型错误")
        async for message in client.receive_response():
            if isinstance(message, (AssistantMessage, ResultMessage)):
                print(message)


asyncio.run(main())

成本控制：别让 AI 把你的钱烧光

多 Agent 并行 = 成本叠加风险。务必设置预算上限，并实时追踪：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage


async def run_with_budget(task: str):
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        max_turns=15,
        max_budget_usd=0.50,  # 单个 Agent 最多花 50 美分
    )

    async for message in query(prompt=task, options=options):
        if isinstance(message, ResultMessage):
            # total_cost_usd：Python SDK ResultMessage 的累计估算成本字段
            # （官方文档：code.claude.com/docs/en/agent-sdk/cost-tracking）
            print(f"本次消耗（估算）：${message.total_cost_usd:.4f}")
            # message.usage 包含 input_tokens / output_tokens 明细
            if message.usage:
                print(f"  输入 tokens：{message.usage.get('input_tokens', 0)}")
                print(f"  输出 tokens：{message.usage.get('output_tokens', 0)}")
            print(f"结果：{message.result}")


asyncio.run(run_with_budget("分析 src/ 目录的代码结构，生成依赖关系报告"))

最佳实践（成本分层）：

Orchestrator 用 claude-opus-4-7（最聪明，做规划和汇总）
Subagents 用 claude-sonnet-4-6（性价比高，做具体执行）
简单查询用 claude-haiku-4-5（最便宜，做信息检索）

⚠️ 注意：total_cost_usd 是 SDK 根据本地价格表在客户端估算的，SDK 升级或定价变动时可能偏差。精确计费请用 Claude Console Usage 页面或官方 Usage & Cost API，不要用这个字段做财务决策。

什么任务最适合 Team Agent？

✅ 适合：

大型代码重构（多文件、多模块）
全栈功能开发（前端 + 后端 + 测试同时进行）
代码审查 + 自动修复
批量文档生成
多语言国际化

❌ 不适合：

有强依赖顺序的任务（A 必须在 B 之前完成）
简单的一问一答
需要高度创意判断的任务（AI 各说各话容易矛盾）

一句话总结

Team Agent 不是让 AI 更快地回答你的问题，而是让 AI 替你管理整个项目。

你的角色从"程序员"变成"工程总监"——你定目标，AI 做执行。

这不是遥远的未来，这是现在的 Claude Agent SDK，现在就能装，现在就能用。

1
2
pip install claude-agent-sdk
export ANTHROPIC_API_KEY=your-api-key

三行代码，开始你的第一个 Team Agent。

想看更多 AI 编程实战案例？欢迎关注。

先说说你现在在做什么#

什么是 Team Agent 模式？#

实战：一个真实的例子#

核心机制：Agent Loop#

实战代码：完整的 Team Agent 示例#

进阶：用 Hooks 守住你的代码边界#

成本控制：别让 AI 把你的钱烧光#

什么任务最适合 Team Agent？#

一句话总结#