Claude Agent SDK 实战：从 Claude Code 到通用 Agent 运行时的进化之路

from claude_agent_sdk import ClaudeAgent, tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。
    
    Args:
        city: 城市名称，如 "北京"、"上海"
    """
    # 模拟天气 API 调用
    weather_data = {
        "北京": "晴，25°C，湿度 40%",
        "上海": "多云，22°C，湿度 65%",
        "广州": "小雨，28°C，湿度 80%",
    }
    return weather_data.get(city, f"{city}：暂无天气数据")

@tool
def search_web(query: str) -> str:
    """搜索网络获取最新信息。
    
    Args:
        query: 搜索查询
    """
    # 模拟搜索 API
    return f"搜索结果：关于 '{query}' 的最新信息..."

# 创建 Agent
agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    system_prompt="你是一个友好的 AI 助手，可以查询天气和网络信息。",
    tools=[get_weather, search_web],
    max_turns=10,  # 限制最大工具调用轮次
)

# 运行 Agent
response = agent.run("北京今天天气怎么样？顺便搜一下明天有没有什么科技新闻。")
print(response)

pip install claude-agent-sdk

export ANTHROPIC_API_KEY=your_api_key_here

export CLAUDE_CODE_USE_BEDROCK=1
export AWS_REGION=us-east-1
export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key

export CLAUDE_CODE_USE_VERTEX=1
export CLOUD_ML_REGION=us-east5
export ANTHROPIC_VERTEX_PROJECT_ID=your_project_id

from claude_agent_sdk import ClaudeAgent

# 创建 Agent 实例
agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    system_prompt="你是一个友好的 AI 助手。",
)

# 运行 Agent
response = agent.run("什么是 AI Agent？")
print(response)

from claude_agent_sdk import ClaudeAgent, Permission, tool

@tool
def read_file(path: str) -> str:
    """读取文件内容。"""
    with open(path, 'r') as f:
        return f.read()

@tool
def write_file(path: str, content: str) -> str:
    """写入文件。"""
    with open(path, 'w') as f:
        f.write(content)
    return f"已写入 {path}"

@tool
def execute_shell(command: str) -> str:
    """执行 Shell 命令。"""
    import subprocess
    result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return result.stdout

# 定义权限
permissions = [
    # 允许读取 /tmp 下的文件
    Permission(
        tool="read_file",
        condition=lambda args: args["path"].startswith("/tmp/"),
        require_approval=False,
    ),
    
    # 允许写入 /tmp/output 下的文件
    Permission(
        tool="write_file",
        condition=lambda args: args["path"].startswith("/tmp/output/"),
        require_approval=False,
    ),
    
    # Shell 命令需要批准
    Permission(
        tool="execute_shell",
        require_approval=True,
    ),
]

# 创建 Agent
agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    tools=[read_file, write_file, execute_shell],
    permissions=permissions,
    interactive=True,  # 启用交互式批准
)

# 运行 Agent
response = agent.run("读取 /tmp/data.txt，统计字数，将结果写入 /tmp/output/result.txt")
print(response)

from claude_agent_sdk import ClaudeAgent, Permission

# 定义权限规则
permissions = [
    # 允许读取 /tmp 目录下的文件
    Permission(
        tool="read_file",
        condition=lambda args: args["path"].startswith("/tmp/"),
        require_approval=False,
    ),
    
    # 允许写入 /tmp/output 目录
    Permission(
        tool="write_file",
        condition=lambda args: args["path"].startswith("/tmp/output/"),
        require_approval=False,
    ),
    
    # 执行 Shell 命令需要用户批准
    Permission(
        tool="execute_shell",
        require_approval=True,  # 每次调用都需要用户批准
    ),
    
    # 禁止删除文件
    Permission(
        tool="delete_file",
        allow=False,  # 完全禁止
    ),
]

agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    tools=[...],
    permissions=permissions,
)

# 生产环境：严格限制
production_permissions = [
    # 只允许读取特定目录
    Permission(
        tool="read_file",
        condition=lambda args: (
            args["path"].startswith("/data/input/") and
            not args["path"].endswith(".secret")
        ),
        require_approval=False,
    ),
    
    # 只允许写入特定目录
    Permission(
        tool="write_file",
        condition=lambda args: args["path"].startswith("/data/output/"),
        require_approval=False,
    ),
    
    # 禁止所有 Shell 命令
    Permission(tool="execute_shell", allow=False),
    
    # 禁止删除
    Permission(tool="delete_file", allow=False),
    
    # 只允许调用特定的 API
    Permission(
        tool="http_request",
        condition=lambda args: args["url"].startswith("https://api.example.com/"),
        require_approval=False,
    ),
]

from claude_agent_sdk import ClaudeAgent, Hook

def log_tool_call(tool_name, tool_args):
    print(f"[LOG] 调用工具: {tool_name}, 参数: {tool_args}")
    return None  # 返回 None 表示不修改工具调用

agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    tools=[...],
    hooks=[
        Hook(event="pre_tool_call", handler=log_tool_call),
    ],
)

from claude_agent_sdk import ClaudeAgent, Subagent

# 研究子 Agent
research_agent = Subagent(
    name="researcher",
    model="claude-sonnet-4-20250514",
    system_prompt="你是一个研究助理，专门收集和整理信息。",
    tools=[search_web, read_document],
)

# 写作子 Agent
writing_agent = Subagent(
    name="writer",
    model="claude-sonnet-4-20250514",
    system_prompt="你是一个写作助理，专门将信息整理成清晰的文章。",
    tools=[write_file],
)

# 主 Agent
main_agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    system_prompt="你是一个项目经理，协调研究和写作任务。",
    subagents=[research_agent, writing_agent],
)

response = main_agent.run("写一篇关于 AI Agent 可观测性的文章。")

from claude_agent_sdk import ClaudeAgent, MCPClient, tool

# 自定义工具（内置）
@tool
def calculate(expression: str) -> str:
    """计算数学表达式。"""
    return str(eval(expression))

# MCP Server 1：GitHub
github_mcp = MCPClient(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-github"],
)

# MCP Server 2：文件系统
fs_mcp = MCPClient(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
)

# 创建 Agent，同时使用内置工具和 MCP 工具
agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    tools=[calculate],  # 内置工具
    mcp_clients=[github_mcp, fs_mcp],  # MCP 工具
)

# Agent 可以混合使用所有工具
response = agent.run("""
1. 列出我的 GitHub 仓库
2. 找到 star 数最多的仓库
3. 计算 star 数的平方根
4. 将结果写入 /tmp/result.txt
""")
print(response)

from claude_agent_sdk import ClaudeAgent, MCPClient

# 连接 MCP Server
mcp_client = MCPClient(
    server_url="http://localhost:3000",  # MCP Server 地址
    # 或者使用 stdio（本地进程）
    # command="npx",
    # args=["-y", "@modelcontextprotocol/server-github"],
)

# 创建 Agent，注册 MCP 工具
agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    mcp_clients=[mcp_client],
)

# Agent 可以自动发现并使用 MCP Server 提供的工具
response = agent.run("列出我的 GitHub 仓库")

# 安装 GitHub MCP Server
npm install -g @modelcontextprotocol/server-github

# 配置 GitHub Token
export GITHUB_TOKEN=your_github_token

from claude_agent_sdk import ClaudeAgent, MCPClient

# 连接 GitHub MCP Server
github_mcp = MCPClient(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-github"],
)

agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    mcp_clients=[github_mcp],
)

# Agent 可以：
# - 列出仓库
# - 创建 Issue
# - 提交 Pull Request
# - 查看 CI 状态
# - 等等
response = agent.run("在我的 awesome-project 仓库创建一个 Issue，标题是 'Bug: 修复登录问题'")

from claude_agent_sdk import ClaudeAgent, Permission, Hook, RetryPolicy
from opentelemetry import trace
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 配置 OpenTelemetry
tracer = trace.get_tracer("production-agent")

# 审计日志 Hook
def audit_hook(event, data):
    logger.info(f"[AUDIT] {event}: {data}")
    if event == "pre_tool_call":
        span = tracer.start_span(f"tool.{data['tool_name']}")
        span.set_attribute("tool.input", str(data['tool_args']))
    elif event == "post_tool_call":
        span = trace.get_current_span()
        span.set_attribute("tool.output", str(data['tool_result']))
        span.end()

# 严格的权限规则
permissions = [
    Permission(
        tool="read_file",
        condition=lambda args: args["path"].startswith("/data/input/"),
        require_approval=False,
    ),
    Permission(
        tool="write_file",
        condition=lambda args: args["path"].startswith("/data/output/"),
        require_approval=False,
    ),
    Permission(tool="execute_shell", allow=False),
    Permission(tool="delete_file", allow=False),
]

# 生产级 Agent 配置
agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    system_prompt="你是一个生产级 AI Agent。遵循最小权限原则，避免不必要的工具调用。",
    tools=[...],
    permissions=permissions,
    max_turns=10,
    timeout=300,
    retry_policy=RetryPolicy(
        max_retries=3,
        retry_on=[TimeoutError, ConnectionError],
        backoff_factor=2,
    ),
    fallback_model="claude-haiku-3-5-20241022",
    hooks=[
        Hook(event="pre_tool_call", handler=lambda data: audit_hook("pre_tool_call", data)),
        Hook(event="post_tool_call", handler=lambda data: audit_hook("post_tool_call", data)),
    ],
    audit_log={
        "enabled": True,
        "log_file": "/var/log/agent/audit.log",
        "log_level": "INFO",
    },
)

# 成本优化示例
agent = ClaudeAgent(
    model="claude-haiku-3-5-20241022",  # 使用更便宜的模型
    max_turns=5,  # 限制工具调用轮次
    system_prompt="简洁回答，避免不必要的工具调用。",  # 引导模型节省 token
)

# 流式输出示例
for chunk in agent.run("写一篇长文", stream=True):
    print(chunk, end="", flush=True)

# 生产环境安全配置
agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    tools=[...],
    permissions=[...],  # 严格的权限规则
    max_turns=10,
    timeout=300,  # 5 分钟超时
    # 审计日志
    audit_log={
        "enabled": True,
        "log_file": "/var/log/agent/audit.log",
        "log_level": "INFO",
    },
)

from claude_agent_sdk import ClaudeAgent, Hook
from opentelemetry import trace

tracer = trace.get_tracer("agent-observability")

def otel_hook(event, data):
    if event == "pre_tool_call":
        span = tracer.start_span(f"tool.{data['tool_name']}")
        span.set_attribute("tool.input", str(data['tool_args']))
    elif event == "post_tool_call":
        span = trace.get_current_span()
        span.set_attribute("tool.output", str(data['tool_result']))
        span.end()

agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    tools=[...],
    hooks=[
        Hook(event="pre_tool_call", handler=lambda data: otel_hook("pre_tool_call", data)),
        Hook(event="post_tool_call", handler=lambda data: otel_hook("post_tool_call", data)),
    ],
)

from claude_agent_sdk import ClaudeAgent, RetryPolicy

agent = ClaudeAgent(
    model="claude-sonnet-4-20250514",
    tools=[...],
    retry_policy=RetryPolicy(
        max_retries=3,
        retry_on=[TimeoutError, ConnectionError],
        backoff_factor=2,
    ),
    fallback_model="claude-haiku-3-5-20241022",  # 备用模型
)