## 概述
2026年4月8日，Anthropic同时推出两条Agent产品线：**Claude Agent SDK**（自托管自主Agent框架）和**Managed Agents**（云端托管Agent基础设施）。本教程将深入两条路径的架构差异、适用场景，并通过完整代码示例演示如何从零构建单Agent工具调用链，进而扩展到多Agent编排系统。适合已熟悉Claude Code基础操作、希望构建自主Agent系统的中高级开发者。
## 前置条件
- Python 3.10+ 与 Node.js 18+
- Claude Code CLI（最新版本）
- Anthropic API Key（需开通Agent SDK与Managed Agents权限）
- 基本了解 REST API 与异步编程
- 熟悉 Claude Code 基本命令（/help、/init、Task工具）
---
## 一、双轨架构全景：为什么是两条路
Anthropic在2026年Q1-Q2采取了一个看似矛盾但实则精妙的策略：同时提供**自托管SDK**和**云端托管API**。理解这两条路径的设计哲学，是做出正确技术选型的前提。
```mermaid
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或
自编排多Agent]
D1 & D2 & D3 & D4 --> F[会话管理 +
多Agent编排]
end
```
### 1.1 Claude Agent SDK：给Claude一台电脑
Agent SDK的设计哲学可以概括为一句话：**"Give Claude a computer"**。它提供：
- **原生Bash执行**：Agent可以直接运行shell命令
- **文件系统访问**：读写、创建、删除文件
- **自定义工具注册**：将任意Python/TS函数暴露为Agent工具
- **上下文管理**：对话历史、文件引用、环境变量的完整生命周期
核心架构是一个**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.
关键能力：
- **云端安全沙箱**：无需自建基础设施
- **会话生命周期管理**：创建、运行、暂停、恢复
- **Webhook事件流**：实时监听Agent行为
- **多Agent编排**：主Agent委托子Agent并行处理
- **持久化记忆**：跨会话知识留存
### 1.3 选型决策矩阵
| 维度 | Agent SDK | Managed Agents |
|------|-----------|----------------|
| 部署位置 | 本地/自有服务器 | Anthropic云 |
| 执行环境 | 自定义沙箱或宿主机 | 云端安全容器 |
| 成本模型 | API调用费 + 基础设施 | 按会话计费（$0.08/session起） |
| 控制粒度 | 完全控制 | 托管控制 |
| 多Agent | 自行编排 | 内置编排API |
| 适用规模 | 个人/小团队 | 企业级/大规模 |
| 上手难度 | 中等 | 较低 |
---
## 二、Agent SDK实战：构建自主代码审查Agent
### 2.1 环境搭建
```bash
# 安装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的核心能力取决于它拥有哪些工具。下面定义三个代码审查工具：
```python
# 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主循环
```python
# 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，并实现重试逻辑：
```python
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不执行危险操作？**
在工具实现中添加安全检查层：
```python
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
```bash
# 安装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 启动会话并监控
```python
# 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。
```mermaid
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审查报告
```
```python
# 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模式比轮询更高效：
```python
# 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 架构设计
```mermaid
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 完整实现
```python
# 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
```yaml
# .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带来的编排死锁、成本放大和调试复杂性往往是过度工程。判断标准：
```python
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在无约束情况下可能产生意外行为。最佳实践是设置严格的权限边界：
```yaml
# 安全策略示例
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
```
**反模式三：忽略成本控制**
```python
# 成本监控装饰器
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 会话事件流分析
```python
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架构：
```text
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的时机。
---
## 延伸阅读
- Anthropic工程博客：[Scaling Managed Agents: Decoupling the brain from the hands](https://www.anthropic.com/engineering/managed-agents) — Managed Agents的架构设计哲学
- Anthropic研究：[Building Effective Agents](https://www.anthropic.com/research/building-effective-agents) — Agent设计的最佳实践与模式
- Claude API文档：[Multiagent Sessions](https://platform.claude.com/docs/en/managed-agents/multi-agent) — 多Agent编排API官方参考
## 参考来源
1. [Claude Agent SDK & Managed Agents: Anthropic's Q2 2026 Agent Infrastructure Play](https://zylos.ai/research/2026-04-20-claude-agent-sdk-managed-agents-architecture) — Zylos Research对双轨架构的深度技术分析
2. [How to Build AI Agents with Claude Agent SDK in 2026](https://codelucky.com/build-ai-agents-claude-agent-sdk-2026/) — Agent SDK完整构建教程与代码示例
3. [Code with Claude Conference Recap: Managed Agents, SpaceX Compute](https://claudeapi.com/en/blog/news/code-with-claude-conference/) — 2026年5月6日大会要点，含Managed Agents正式发布信息
4. [Multi-Agent Orchestration with the Anthropic Agent SDK](https://www.nintecsystems.com/insights/multi-agent-orchestration-anthropic-agent-sdk) — NINtec关于多Agent决策框架的工程分析
5. [2026 AI Agent Framework Showdown](https://qubittool.com/blog/ai-agent-framework-comparison-2026) — 六大Agent框架横向对比（含Claude SDK）
## 训练营适用性评估
- **目标训练营**：AI工程高级训练营、Agent开发专项训练营
- **教学时长**：4-6小时（含实操），其中Agent SDK部分2小时、Managed Agents部分2小时、综合实战2小时
- **难度定位**：中高级，需具备Python异步编程和API开发基础
- **实验环境**：需Anthropic API Key、GitHub仓库、Python 3.12+环境
> [!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 等场景全覆盖。
