首页 › tutorials › Claude Agent双轨编排:SDK到托管Agent实战

概述

2026年4月8日,Anthropic同时推出两条Agent产品线:Claude Agent SDK(自托管自主Agent框架)和Managed Agents(云端托管Agent基础设施)。本教程将深入两条路径的架构差异、适用场景,并通过完整代码示例演示如何从零构建单Agent工具调用链,进而扩展到多Agent编排系统。适合已熟悉Claude Code基础操作、希望构建自主Agent系统的中高级开发者。

前置条件


一、双轨架构全景:为什么是两条路

Anthropic在2026年Q1-Q2采取了一个看似矛盾但实则精妙的策略:同时提供自托管SDK和云端托管API。理解这两条路径的设计哲学,是做出正确技术选型的前提。

graph TB
    subgraph "双轨架构全景"
        A[开发者需求] --> B{部署模式?}
        B -->|本地/自建| C[Claude Agent SDK]
        B -->|云端托管| D[Managed Agents API]

        C --> C1[Python/TS SDK]
        C --> C2[本地沙箱执行]
        C --> C3[自定义工具链]
        C --> C4[完全控制]

        D --> D1[REST API]
        D --> D2[云端安全沙箱]
        D --> D3[内置MCP连接器]
        D --> D4[弹性扩缩容]

        C1 & C2 & C3 & C4 --> E[单Agent或<br/>自编排多Agent]
        D1 & D2 & D3 & D4 --> F[会话管理 +<br/>多Agent编排]
    end

1.1 Claude Agent SDK:给Claude一台电脑

Agent SDK的设计哲学可以概括为一句话:"Give Claude a computer"。它提供:

核心架构是一个agentic loop——模型思考、调用工具、观察结果、继续思考,直到任务完成或达到停止条件。

1.2 Managed Agents:解耦大脑与双手

Managed Agents的设计哲学是"Decoupling the brain from the hands"。Anthropic工程博客明确指出:

Harnesses encode assumptions that go stale as models improve. Managed Agents is built around interfaces that stay stable as harnesses change.

关键能力:

1.3 选型决策矩阵

维度 Agent SDK Managed Agents
部署位置 本地/自有服务器 Anthropic云
执行环境 自定义沙箱或宿主机 云端安全容器
成本模型 API调用费 + 基础设施 按会话计费($0.08/session起)
控制粒度 完全控制 托管控制
多Agent 自行编排 内置编排API
适用规模 个人/小团队 企业级/大规模
上手难度 中等 较低

二、Agent SDK实战:构建自主代码审查Agent

2.1 环境搭建

# 安装Agent SDK
pip install claude-agent-sdk anthropic

# 验证安装
python -c "import claude_agent_sdk; print(claude_agent_sdk.__version__)"

# 设置API Key
export ANTHROPIC_API_KEY="sk-ant-..."

2.2 定义工具集

一个Agent的核心能力取决于它拥有哪些工具。下面定义三个代码审查工具:

# agent_tools.py
import subprocess
import json
from pathlib import Path

def run_linter(file_path: str) -> dict:
    """对指定文件运行linter检查"""
    ext = Path(file_path).suffix
    linters = {
        ".py": ["ruff", "check", "--output-format=json", file_path],
        ".js": ["eslint", "--format=json", file_path],
        ".ts": ["eslint", "--format=json", file_path],
    }

    if ext not in linters:
        return {"status": "skipped", "reason": f"No linter for {ext}"}

    result = subprocess.run(
        linters[ext], capture_output=True, text=True, timeout=30
    )

    if result.returncode == 0:
        return {"status": "clean", "issues": []}

    try:
        issues = json.loads(result.stdout)
        return {"status": "issues_found", "issues": issues}
    except json.JSONDecodeError:
        return {"status": "error", "raw": result.stderr}

def read_file_content(file_path: str) -> dict:
    """读取文件内容,返回源码和元信息"""
    path = Path(file_path)
    if not path.exists():
        return {"error": f"File not found: {file_path}"}

    stat = path.stat()
    content = path.read_text(encoding="utf-8", errors="replace")

    return {
        "path": str(path),
        "size_bytes": stat.st_size,
        "lines": content.count("\n") + 1,
        "content": content[:50000],  # 限制大小
    }

def run_tests(test_path: str = "tests/") -> dict:
    """运行测试套件并返回结果"""
    result = subprocess.run(
        ["pytest", test_path, "-v", "--tb=short", "-q"],
        capture_output=True, text=True, timeout=120
    )

    return {
        "exit_code": result.returncode,
        "passed": "passed" in result.stdout.lower(),
        "output": result.stdout[-2000:],  # 最后2000字符
        "errors": result.stderr[-1000:],
    }

2.3 构建Agent主循环

# review_agent.py
import anthropic
import json
from agent_tools import run_linter, read_file_content, run_tests

client = anthropic.Anthropic()

# 工具定义(OpenAPI格式)
tools = [
    {
        "name": "run_linter",
        "description": "对指定源码文件运行静态分析检查,返回代码质量问题",
        "input_schema": {
            "type": "object",
            "properties": {
                "file_path": {
                    "type": "string",
                    "description": "要检查的文件路径"
                }
            },
            "required": ["file_path"]
        }
    },
    {
        "name": "read_file_content",
        "description": "读取文件内容,用于理解代码逻辑",
        "input_schema": {
            "type": "object",
            "properties": {
                "file_path": {
                    "type": "string",
                    "description": "要读取的文件路径"
                }
            },
            "required": ["file_path"]
        }
    },
    {
        "name": "run_tests",
        "description": "运行项目测试套件,验证修改后的正确性",
        "input_schema": {
            "type": "object",
            "properties": {
                "test_path": {
                    "type": "string",
                    "description": "测试目录路径,默认为tests/"
                }
            }
        }
    }
]

# 工具执行映射
TOOL_MAP = {
    "run_linter": run_linter,
    "read_file_content": read_file_content,
    "run_tests": run_tests,
}

SYSTEM_PROMPT = """你是一个专业的代码审查Agent。你的工作流程:
1. 首先读取目标文件,理解代码结构
2. 运行linter检查静态问题
3. 运行测试验证功能正确性
4. 综合分析结果,给出结构化的审查报告

审查报告格式:
- 严重程度:🔴 严重 / 🟡 警告 / 🟢 建议
- 问题描述
- 修复建议(含代码示例)
"""

def run_review_agent(target_files: list[str], max_turns: int = 10) -> str:
    """运行代码审查Agent的主循环"""

    messages = [{
        "role": "user",
        "content": f"请审查以下文件的代码质量:{', '.join(target_files)}。"
                   f"按照工作流程执行完整的代码审查。"
    }]

    for turn in range(max_turns):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=SYSTEM_PROMPT,
            tools=tools,
            messages=messages,
        )

        # 检查是否停止
        if response.stop_reason == "end_turn":
            # 提取最终文本回复
            final_text = ""
            for block in response.content:
                if block.type == "text":
                    final_text += block.text
            return final_text

        # 处理工具调用
        assistant_content = response.content
        messages.append({"role": "assistant", "content": assistant_content})

        # 执行所有工具调用
        tool_results = []
        for block in assistant_content:
            if block.type == "tool_use":
                tool_name = block.name
                tool_input = block.input

                print(f"  [Turn {turn+1}] 调用工具: {tool_name}")
                print(f"    参数: {json.dumps(tool_input, ensure_ascii=False)}")

                # 执行工具
                tool_func = TOOL_MAP.get(tool_name)
                if tool_func:
                    result = tool_func(**tool_input)
                else:
                    result = {"error": f"Unknown tool: {tool_name}"}

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result, ensure_ascii=False),
                })

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

    return "Agent达到最大轮次限制,审查未完成。"


# 执行审查
if __name__ == "__main__":
    import sys
    files = sys.argv[1:] or ["src/main.py"]
    report = run_review_agent(files)
    print("\n" + "="*60)
    print("代码审查报告")
    print("="*60)
    print(report)

2.4 预期输出

  [Turn 1] 调用工具: read_file_content
    参数: {"file_path": "src/main.py"}
  [Turn 2] 调用工具: run_linter
    参数: {"file_path": "src/main.py"}
  [Turn 3] 调用工具: run_tests
    参数: {"test_path": "tests/"}

============================================================
代码审查报告
============================================================
## 审查结果:src/main.py

### 🔴 严重问题(1)
- **SQL注入风险**(第42行):直接拼接用户输入到SQL查询
  ```python
  # 修复建议
  cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
  ```

### 🟡 警告(2)
- **未处理的异常**(第78行):数据库连接缺少try/except
- **硬编码配置**(第15行):数据库URL应从环境变量读取

### 🟢 建议(1)
- **类型注解缺失**:建议添加函数签名类型标注

测试结果:全部通过(12/12)

2.5 常见问题与解决

Q: Agent工具调用超时怎么办?

设置合理的timeout,并实现重试逻辑:

import time

def call_tool_with_retry(func, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(**kwargs)
        except subprocess.TimeoutExpired:
            if attempt == max_retries - 1:
                return {"error": "Tool execution timed out"}
            time.sleep(2 ** attempt)
        except Exception as e:
            return {"error": str(e)}

Q: 如何控制Agent不执行危险操作?

在工具实现中添加安全检查层:

ALLOWED_EXTENSIONS = {".py", ".js", ".ts", ".go", ".rs", ".java"}
BLOCKED_PATHS = {".env", "credentials", "secret", ".ssh"}

def safe_read_file(file_path: str) -> dict:
    path = Path(file_path).resolve()

    # 路径安全检查
    if any(blocked in str(path).lower() for blocked in BLOCKED_PATHS):
        return {"error": "Access denied: sensitive path"}

    if path.suffix not in ALLOWED_EXTENSIONS:
        return {"error": f"Unsupported file type: {path.suffix}"}

    return read_file_content(file_path)

三、Managed Agents实战:云端托管多Agent系统

3.1 创建第一个托管Agent

# 安装CLI工具
pip install anthropic[managed-agents]

# 创建Agent配置文件
cat > agent-config.yaml << 'EOF'
agent:
  name: "code-reviewer"
  description: "自动代码审查Agent,支持静态分析和测试验证"
  model: "claude-sonnet-4-6"

  tools:
    - name: "run_linter"
      description: "运行代码静态分析"
      type: "bash"
      command: "ruff check --output-format=json {{file_path}}"

    - name: "run_tests"
      description: "运行测试套件"
      type: "bash"
      command: "pytest {{test_path}} -v --tb=short"

  environment:
    runtime: "python:3.12"
    packages:
      - "ruff>=0.4"
      - "pytest>=8.0"

  permissions:
    allowed_commands: ["ruff", "pytest", "cat", "ls"]
    max_execution_time: 300
    network_access: false
EOF

# 注册Agent
anthropic agents create --config agent-config.yaml

3.2 启动会话并监控

# managed_session.py
import anthropic
import time

client = anthropic.Anthropic()

# 创建Agent会话
session = client.beta.agents.sessions.create(
    agent_id="code-reviewer",
    initial_message="审查 src/api/handlers.py 文件的安全性和代码质量",
)

print(f"会话ID: {session.id}")
print(f"状态: {session.status}")

# 轮询事件流
events = client.beta.agents.sessions.events.list(
    session_id=session.id,
)

for event in events:
    match event.type:
        case "agent.thinking":
            print(f"🧠 思考中: {event.thinking[:100]}...")
        case "agent.tool_use":
            print(f"🔧 调用工具: {event.tool_name}({event.tool_input})")
        case "agent.tool_result":
            print(f"📋 工具结果: {event.content[:200]}")
        case "agent.message":
            print(f"💬 Agent回复: {event.content}")
        case "session.complete":
            print("✅ 会话完成")
            break

3.3 多Agent编排:核心模式

Managed Agents的真正威力在于多Agent编排。Anthropic在2026年5月6日的"Code with Claude"大会上正式发布了Multiagent Sessions API。

sequenceDiagram
    participant User as 开发者
    participant Orch as 编排Agent
    participant Lint as Linter Agent
    participant Sec as 安全Agent
    participant Test as 测试Agent
    participant Sum as 摘要Agent

    User->>Orch: 审查PR #42
    Orch->>Lint: 并行 - 静态分析5个文件
    Orch->>Sec: 并行 - 安全漏洞扫描
    Lint-->>Orch: 12个代码质量问题
    Sec-->>Orch: 2个安全隐患
    Orch->>Test: 执行测试套件
    Test-->>Orch: 8/10 passed, 2 failed
    Orch->>Sum: 汇总审查报告
    Sum-->>User: 结构化PR审查报告
# multi_agent_orchestrator.py
import anthropic
import asyncio
from dataclasses import dataclass

client = anthropic.Anthropic()

@dataclass
class AgentSpec:
    id: str
    role: str
    tools: list[str]

# 定义专业Agent团队
TEAM = {
    "linter": AgentSpec(
        id="code-linter-v1",
        role="静态代码分析专家,专注于代码风格、复杂度、可维护性",
        tools=["ruff_check", "eslint", "complexity_analyzer"],
    ),
    "security": AgentSpec(
        id="security-scanner-v1",
        role="安全审计专家,检测OWASP Top 10漏洞、依赖风险",
        tools=["bandit_scan", "dependency_check", "secret_detector"],
    ),
    "tester": AgentSpec(
        id="test-runner-v1",
        role="测试执行专家,运行并分析测试结果",
        tools=["pytest_run", "coverage_report", "mutation_test"],
    ),
}

async def run_parallel_agents(files: list[str]) -> dict:
    """并行启动多个专业Agent"""

    tasks = {
        "lint": client.beta.agents.sessions.create(
            agent_id=TEAM["linter"].id,
            initial_message=f"对以下文件执行完整的静态分析:{files}",
        ),
        "security": client.beta.agents.sessions.create(
            agent_id=TEAM["security"].id,
            initial_message=f"扫描以下文件的安全漏洞:{files}",
        ),
    }

    # 并行等待第一批Agent完成
    lint_session, sec_session = await asyncio.gather(
        tasks["lint"], tasks["security"]
    )

    # 等待所有结果
    lint_result = await wait_for_completion(lint_session.id)
    sec_result = await wait_for_completion(sec_session.id)

    # 根据前序结果决定是否运行测试
    if lint_result["severity"] in ["high", "critical"] or \
       sec_result["vulnerabilities"]:
        test_session = await client.beta.agents.sessions.create(
            agent_id=TEAM["tester"].id,
            initial_message=(
                f"运行完整测试套件。已知问题:\n"
                f"静态分析: {lint_result['summary']}\n"
                f"安全扫描: {sec_result['summary']}\n"
                f"请重点检查与以上问题相关的测试用例。"
            ),
        )
        test_result = await wait_for_completion(test_session.id)
    else:
        test_result = {"status": "skipped", "reason": "无严重问题"}

    return {
        "lint": lint_result,
        "security": sec_result,
        "test": test_result,
    }

async def wait_for_completion(session_id: str, timeout: int = 300) -> dict:
    """等待Agent会话完成,提取结构化结果"""
    start = time.time()

    while time.time() - start < timeout:
        session = client.beta.agents.sessions.retrieve(
            session_id=session_id
        )

        if session.status == "completed":
            # 从会话消息中提取最终结果
            messages = client.beta.agents.sessions.messages.list(
                session_id=session_id
            )
            return parse_agent_output(messages)

        elif session.status == "failed":
            return {
                "status": "error",
                "error": session.error_message,
            }

        await asyncio.sleep(2)

    return {"status": "timeout"}

def parse_agent_output(messages) -> dict:
    """解析Agent输出为结构化数据"""
    for msg in reversed(messages):
        if msg.role == "assistant" and msg.content:
            return {
                "status": "completed",
                "summary": msg.content[:500],
                "raw": msg.content,
            }
    return {"status": "no_output"}

3.4 Webhook驱动的异步编排

对于长时间运行的任务,Webhook模式比轮询更高效:

# webhook_orchestrator.py
from flask import Flask, request
import anthropic

app = Flask(__name__)
client = anthropic.Anthropic()

# 注册Webhook
client.beta.agents.webhooks.create(
    url="https://your-server.com/webhook/agent-events",
    events=["session.completed", "session.failed", "tool.executed"],
)

@app.route("/webhook/agent-events", methods=["POST"])
def handle_agent_event():
    event = request.json

    match event["type"]:
        case "session.completed":
            session_id = event["session_id"]
            result = event["result"]

            # 触发下游Agent
            if event["agent_id"] == "code-linter-v1":
                # Linter完成,启动安全扫描
                client.beta.agents.sessions.create(
                    agent_id="security-scanner-v1",
                    initial_message=f"基于Linter发现的问题,深入扫描:{result['summary']}",
                )

        case "tool.executed":
            print(f"工具执行: {event['tool_name']} -> {event['status']}")

        case "session.failed":
            # 错误处理:通知开发者
            send_alert(
                f"Agent {event['agent_id']} 失败: {event['error']}"
            )

    return {"status": "ok"}, 200

四、实战案例:GitHub PR自动审查系统

将Agent SDK与Managed Agents结合,构建一个完整的PR自动审查流水线。

4.1 架构设计

graph LR
    subgraph "GitHub Webhook"
        A[PR Created] --> B[Webhook触发]
    end

    subgraph "编排层 (Agent SDK)"
        B --> C[主控Agent]
        C --> D[文件分类]
        D --> E{文件类型?}
    end

    subgraph "执行层 (Managed Agents)"
        E -->|Python| F[Linter Agent]
        E -->|TypeScript| G[ESLint Agent]
        E -->|通用| H[安全Agent]
        F & G & H --> I[结果聚合]
    end

    subgraph "输出层"
        I --> J[生成PR Review评论]
        J --> K[GitHub API]
        K --> L[PR页面展示]
    end

4.2 完整实现

# pr_review_pipeline.py
import anthropic
import subprocess
import json
from github import Github

client = anthropic.Anthropic()
gh = Github()

def on_pr_created(repo_name: str, pr_number: int):
    """PR创建时的入口函数"""

    # 1. 获取PR变更文件
    repo = gh.get_repo(repo_name)
    pr = repo.get_pull(pr_number)
    files = [f.filename for f in pr.get_files()]

    print(f"PR #{pr_number}: {len(files)} files changed")

    # 2. 用Agent SDK分类文件(快速本地推理)
    classification = classify_files(files)

    # 3. 分派到Managed Agents执行专业审查
    review_results = {}

    # Python文件 -> Linter Agent
    if classification["python"]:
        review_results["python"] = dispatch_managed_agent(
            agent_id="python-reviewer-v1",
            message=f"审查Python文件: {classification['python']}\n"
                    f"PR描述: {pr.title}\n"
                    f"变更摘要: {pr.body}",
        )

    # 安全关键文件 -> 安全Agent
    if classification["security_sensitive"]:
        review_results["security"] = dispatch_managed_agent(
            agent_id="security-scanner-v1",
            message=f"安全审查: {classification['security_sensitive']}",
        )

    # 4. 聚合结果,生成PR评论
    review_body = format_review_comment(review_results)

    # 5. 发布评论
    pr.create_review(
        body=review_body,
        event="COMMENT",
    )

    print(f"Review posted on PR #{pr_number}")

def classify_files(files: list[str]) -> dict:
    """使用Agent SDK本地快速分类文件"""
    result = {"python": [], "typescript": [], "security_sensitive": []}

    SECURITY_PATTERNS = {
        "auth", "login", "password", "secret", "credential",
        "token", "crypto", "ssl", "certificate",
    }

    for f in files:
        if f.endswith(".py"):
            result["python"].append(f)
        elif f.endswith((".ts", ".tsx", ".js", ".jsx")):
            result["typescript"].append(f)

        path_lower = f.lower()
        if any(p in path_lower for p in SECURITY_PATTERNS):
            result["security_sensitive"].append(f)

    return result

def dispatch_managed_agent(agent_id: str, message: str) -> dict:
    """向Managed Agent派发任务并等待结果"""
    session = client.beta.agents.sessions.create(
        agent_id=agent_id,
        initial_message=message,
    )

    # 同步等待(适合短任务)
    import time
    for _ in range(60):  # 最多等2分钟
        s = client.beta.agents.sessions.retrieve(session_id=session.id)
        if s.status in ("completed", "failed"):
            break
        time.sleep(2)

    if s.status == "failed":
        return {"error": s.error_message}

    messages = client.beta.agents.sessions.messages.list(
        session_id=session.id
    )

    return {"summary": messages[-1].content if messages else "No output"}

def format_review_comment(results: dict) -> str:
    """格式化PR审查评论"""
    sections = ["## 🤖 AI Code Review\n"]

    if "python" in results:
        r = results["python"]
        sections.append("### 🐍 Python静态分析")
        sections.append(r.get("summary", "审查完成"))
        sections.append("")

    if "security" in results:
        r = results["security"]
        sections.append("### 🔒 安全审查")
        sections.append(r.get("summary", "审查完成"))
        sections.append("")

    sections.append("---")
    sections.append("*此审查由Claude Agent自动生成,请结合人工判断。*")

    return "\n".join(sections)

4.3 部署为GitHub Actions

# .github/workflows/ai-review.yml
name: AI Code Review

on:
  pull_request:
    types: [opened, synchronize]

jobs:
  ai-review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: |
          pip install anthropic PyGithub

      - name: Run AI Review
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        run: |
          python pr_review_pipeline.py \
            --repo ${{ github.repository }} \
            --pr ${{ github.event.pull_request.number }}

五、高级编排模式与反模式

5.1 三种成熟的多Agent模式

模式一:MapReduce式并行

适用于独立的、可并行的分析任务。每个Agent处理不同的文件或维度,最后由一个聚合Agent汇总。

适用场景:大规模代码库的静态分析、多维度质量扫描
优势:高并行度、低延迟
风险:上下文重复、成本线性增长

模式二:流水线式串行

Agent按顺序处理,前一个Agent的输出是后一个的输入。

适用场景:需要渐进式深入的场景(先静态分析 → 再安全审计 → 最后测试)
优势:上下文传递精准、每步可做决策
风险:总延迟为各步之和、单点失败

模式三:分层委托

主Agent负责理解任务和分配,子Agent负责执行,结果回传主Agent做最终决策。

适用场景:复杂任务需要"理解-分解-执行-综合"
优势:最灵活、可处理非结构化任务
风险:主Agent成为瓶颈、编排复杂度高

5.2 关键反模式

反模式一:为简单任务引入多Agent

NINtec的研究指出:单Agent循环可以处理绝大多数生产负载。多Agent带来的编排死锁、成本放大和调试复杂性往往是过度工程。判断标准:

def should_use_multi_agent(task_description: str) -> bool:
    """判断是否需要多Agent"""
    indicators = [
        # 任务涉及3个以上独立维度
        count_dimensions(task_description) > 3,
        # 单次执行需要超过5分钟
        estimated_duration(task_description) > 300,
        # 需要不同的专业工具集
        requires_distinct_tools(task_description),
        # 任务可以自然分解
        is_decomposable(task_description),
    ]
    return sum(indicators) >= 3

反模式二:无限制的Agent自主权

Agent在无约束情况下可能产生意外行为。最佳实践是设置严格的权限边界:

# 安全策略示例
permissions:
  # 文件系统:只读 + 指定目录写入
  filesystem:
    read: ["src/**", "tests/**", "docs/**"]
    write: ["reports/**"]

  # 网络:完全禁止
  network: false

  # 命令执行:白名单
  commands:
    allowed: ["ruff", "pytest", "git diff", "git log"]
    blocked: ["rm", "sudo", "curl", "wget"]

  # 资源限制
  limits:
    max_turns: 15
    max_tokens_per_turn: 4096
    total_timeout: 600

反模式三:忽略成本控制

# 成本监控装饰器
import functools

def cost_guard(max_cost_usd: float = 0.50):
    """限制单次Agent调用的成本上限"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            session = func(*args, **kwargs)
            usage = client.beta.agents.sessions.usage(
                session_id=session.id
            )
            actual_cost = usage.total_cost_usd

            if actual_cost > max_cost_usd:
                # 终止会话
                client.beta.agents.sessions.cancel(
                    session_id=session.id,
                    reason=f"Cost limit exceeded: ${actual_cost:.2f} > ${max_cost_usd:.2f}"
                )
                return {"error": "Cost limit exceeded"}

            return session
        return wrapper
    return decorator

六、调试与可观测性

6.1 会话事件流分析

def analyze_session_events(session_id: str):
    """分析Agent会话的完整事件流"""
    events = client.beta.agents.sessions.events.list(
        session_id=session_id,
    )

    metrics = {
        "total_turns": 0,
        "tool_calls": 0,
        "total_input_tokens": 0,
        "total_output_tokens": 0,
        "tool_breakdown": {},
        "errors": [],
    }

    for event in events:
        if event.type == "agent.thinking":
            metrics["total_turns"] += 1
        elif event.type == "agent.tool_use":
            metrics["tool_calls"] += 1
            name = event.tool_name
            metrics["tool_breakdown"][name] = (
                metrics["tool_breakdown"].get(name, 0) + 1
            )
        elif event.type == "session.error":
            metrics["errors"].append(event.message)

    metrics["efficiency"] = (
        metrics["tool_calls"] / max(metrics["total_turns"], 1)
    )

    return metrics

6.2 在Claude Code中直接调试Agent

使用Claude Code的Task工具可以在开发阶段快速测试Agent行为:

# 在Claude Code中输入
请使用Task工具启动一个Explore子Agent,
分析 src/ 目录下所有Python文件的代码复杂度,
返回cyclomatic complexity超过10的函数列表。

这等价于在CLI中手动构建了一个轻量级的单Agent编排。


七、从Claude Code到Agent的迁移路径

如果你是Claude Code用户,可以按以下路径逐步迁移到Agent架构:

graph TD
    A[Step 1: Claude Code交互式] --> B[Step 2: -p模式批处理]
    B --> C[Step 3: Hooks自动化]
    C --> D[Step 4: Agent SDK单Agent]
    D --> E[Step 5: Managed Agents托管]
    E --> F[Step 6: 多Agent编排]

    A --- "日常开发"
    B --- "CI/CD集成"
    C --- "事件驱动"
    D --- "自定义工具链"
    E --- "弹性扩缩容"
    F --- "复杂工作流"

关键转折点:当你发现自己在Claude Code中重复执行相同的多步骤工作流时,就是迁移到Agent SDK的时机。当你需要7x24无人值守运行、或者需要同时处理数十个任务时,就是迁移到Managed Agents的时机。


延伸阅读

参考来源

  1. Claude Agent SDK & Managed Agents: Anthropic's Q2 2026 Agent Infrastructure Play — Zylos Research对双轨架构的深度技术分析
  2. How to Build AI Agents with Claude Agent SDK in 2026 — Agent SDK完整构建教程与代码示例
  3. Code with Claude Conference Recap: Managed Agents, SpaceX Compute — 2026年5月6日大会要点,含Managed Agents正式发布信息
  4. Multi-Agent Orchestration with the Anthropic Agent SDK — NINtec关于多Agent决策框架的工程分析
  5. 2026 AI Agent Framework Showdown — 六大Agent框架横向对比(含Claude SDK)

训练营适用性评估

[!tip] 51学通信知识星球 星主:爱卫生(微信:gprshome201101)——公众号「51学通信」创办者,19 年通信行业运维与讲师经验,著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书(各大电商平台有售)。 星球定位:深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域,主打运维岗位与信令分析。 当前规模:5200 篇主题 / 990 个回答,198 元/年。 加入福利:加入星球后请加爱卫生微信,拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题,另有群内分享。 AI 加持:51学通信知识星球已作为 skill 接入 WorkBuddy,使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例,做总结、找文档和 PPT 等场景全覆盖。

51学通信知识星球海报-扫码加入