Claude Agent SDK

概述

Claude Agent SDK是Anthropic推出的智能体开发框架，围绕Claude模型的独特能力（如extended thinking、tool_use协议、MCP集成）构建。本文深入介绍其架构和使用方法。

相关内容

关于MCP协议的详细介绍，参见 MCP与工具协议。

核心架构

graph TD
    subgraph Claude Agent SDK
        A[Agent] --> B[System Prompt<br/>系统提示]
        A --> C[Tools<br/>工具定义]
        A --> D[MCP Servers<br/>MCP服务器]
        A --> E[Model Config<br/>模型配置]
    end

    subgraph Tool Use Protocol
        C --> C1[function<br/>自定义函数]
        C --> C2[computer_use<br/>计算机操作]
        C --> C3[text_editor<br/>文本编辑]
        C --> C4[bash<br/>命令行]
    end

    subgraph MCP Integration
        D --> D1[File System<br/>文件系统]
        D --> D2[Database<br/>数据库]
        D --> D3[API Gateway<br/>API网关]
        D --> D4[Custom Server<br/>自定义服务]
    end

    subgraph Execution
        F[Agent Loop] --> G[Message]
        G --> H{Response Type}
        H -->|text| I[Text Output]
        H -->|tool_use| J[Execute Tool]
        J --> G
        H -->|end_turn| K[Complete]
    end

tool_use协议

Claude的tool_use是一个结构化的工具调用协议，比简单的function calling更加规范。

工具定义

import anthropic

client = anthropic.Anthropic()

# 定义工具
tools = [
    {
        "name": "get_weather",
        "description": "Get current weather information for a location.",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City name, e.g., 'San Francisco, CA'"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit"
                }
            },
            "required": ["location"]
        }
    },
    {
        "name": "search_database",
        "description": "Search the product database.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "limit": {"type": "integer", "default": 10}
            },
            "required": ["query"]
        }
    }
]

Agent Loop实现

def run_agent_loop(system_prompt, user_message, tools, max_turns=10):
    """Claude Agent的核心循环"""
    messages = [{"role": "user", "content": user_message}]

    for turn in range(max_turns):
        # 调用Claude
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=system_prompt,
            tools=tools,
            messages=messages,
        )

        # 检查是否结束
        if response.stop_reason == "end_turn":
            # 提取最终文本回复
            text_blocks = [b.text for b in response.content 
                          if b.type == "text"]
            return "\n".join(text_blocks)

        # 处理tool_use
        if response.stop_reason == "tool_use":
            # 将assistant的回复加入消息历史
            messages.append({
                "role": "assistant",
                "content": response.content
            })

            # 执行所有工具调用
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result)
                    })

            # 将工具结果加入消息历史
            messages.append({
                "role": "user",
                "content": tool_results
            })

    return "Max turns reached"

Extended Thinking

Claude的Extended Thinking功能让模型在回复前进行深度思考，对复杂Agent任务特别有用。

启用Extended Thinking

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000  # 思考token预算
    },
    messages=[{
        "role": "user", 
        "content": "Analyze this complex system architecture and suggest improvements..."
    }]
)

# 响应包含thinking blocks
for block in response.content:
    if block.type == "thinking":
        print(f"[Thinking]: {block.thinking}")
    elif block.type == "text":
        print(f"[Response]: {block.text}")

在Agent Loop中使用

def agent_with_thinking(task, tools, thinking_budget=5000):
    """带extended thinking的Agent循环"""
    messages = [{"role": "user", "content": task}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=8000,
            thinking={"type": "enabled", "budget_tokens": thinking_budget},
            tools=tools,
            messages=messages,
        )

        # thinking blocks提供了决策过程的可解释性
        for block in response.content:
            if block.type == "thinking":
                logger.info(f"Agent思考过程: {block.thinking[:200]}...")

        if response.stop_reason == "end_turn":
            return extract_text(response)

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

MCP集成

Model Context Protocol (MCP) 是Anthropic推出的开放标准，用于连接LLM与外部数据源和工具。

MCP服务器连接

from anthropic import Anthropic
from anthropic.types import MCPServerConfig

# 配置MCP服务器
mcp_servers = [
    MCPServerConfig(
        name="filesystem",
        command="npx",
        args=["-y", "@anthropic/mcp-server-filesystem", "/path/to/workspace"]
    ),
    MCPServerConfig(
        name="database",
        command="npx",
        args=["-y", "@anthropic/mcp-server-postgres", "postgresql://..."]
    ),
]

# Agent自动发现MCP服务器提供的工具
agent = Agent(
    model="claude-sonnet-4-20250514",
    mcp_servers=mcp_servers,
    system_prompt="You have access to filesystem and database tools."
)

MCP工具协议

graph LR
    A[Claude Agent] -->|tool_use request| B[MCP Client]
    B -->|JSON-RPC| C[MCP Server]
    C -->|Execute| D[External Resource<br/>文件/数据库/API]
    D -->|Result| C
    C -->|JSON-RPC response| B
    B -->|tool_result| A

MCP的优势：

• 标准化: 统一的工具连接协议
• 可发现性: Agent自动发现可用工具
• 安全性: 内置的权限控制
• 可复用性: MCP服务器可跨Agent复用

Claude Code内部机制

Claude Code是基于Claude Agent SDK构建的CLI工具，展示了Agent SDK的实际应用。

工作原理

graph TD
    A[用户输入] --> B[System Prompt<br/>含项目上下文]
    B --> C[Claude Agent Loop]
    C --> D{需要工具?}
    D -->|是| E[工具执行]
    E --> E1[Read File]
    E --> E2[Edit File]
    E --> E3[Bash Command]
    E --> E4[Search/Grep]
    E1 --> C
    E2 --> C
    E3 --> C
    E4 --> C
    D -->|否| F[输出回复]

核心设计模式

1. 项目上下文注入: CLAUDE.md文件提供项目特定指令
2. 多工具协调: 文件读写、搜索、命令执行的协调
3. 权限模型: 用户确认高风险操作
4. 会话管理: 多轮对话中的上下文保持

多轮Agent Loop

对话管理

class ClaudeAgent:
    def __init__(self, system_prompt, tools, model="claude-sonnet-4-20250514"):
        self.client = anthropic.Anthropic()
        self.system_prompt = system_prompt
        self.tools = tools
        self.model = model
        self.conversation_history = []

    def chat(self, user_message):
        """多轮对话"""
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })

        # 运行Agent循环直到获得最终回复
        while True:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=4096,
                system=self.system_prompt,
                tools=self.tools,
                messages=self.conversation_history,
            )

            # 将assistant回复加入历史
            self.conversation_history.append({
                "role": "assistant",
                "content": response.content
            })

            if response.stop_reason == "end_turn":
                return extract_text(response)

            if response.stop_reason == "tool_use":
                tool_results = self._execute_tools(response)
                self.conversation_history.append({
                    "role": "user",
                    "content": tool_results
                })

    def _execute_tools(self, response):
        """执行工具调用并返回结果"""
        results = []
        for block in response.content:
            if block.type == "tool_use":
                try:
                    result = self.tool_map[block.name](block.input)
                    results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result)
                    })
                except Exception as e:
                    results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": f"Error: {str(e)}",
                        "is_error": True
                    })
        return results

上下文窗口管理

当对话历史过长时，需要管理上下文窗口：

def manage_context(messages, max_tokens=180000):
    """上下文窗口管理策略"""
    total_tokens = estimate_tokens(messages)

    if total_tokens > max_tokens:
        # 策略1: 压缩早期消息
        summary = summarize_early_messages(messages[:len(messages)//2])
        messages = [
            {"role": "user", "content": f"Previous conversation summary: {summary}"},
            *messages[len(messages)//2:]
        ]

    return messages

实际案例: 代码审查Agent

# 代码审查Agent
code_review_tools = [
    {
        "name": "read_file",
        "description": "Read the contents of a file.",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "File path"}
            },
            "required": ["path"]
        }
    },
    {
        "name": "search_codebase",
        "description": "Search for patterns in the codebase.",
        "input_schema": {
            "type": "object",
            "properties": {
                "pattern": {"type": "string"},
                "file_glob": {"type": "string", "default": "**/*"}
            },
            "required": ["pattern"]
        }
    },
    {
        "name": "run_tests",
        "description": "Run test suite.",
        "input_schema": {
            "type": "object",
            "properties": {
                "test_path": {"type": "string", "default": "tests/"}
            }
        }
    }
]

review_agent = ClaudeAgent(
    system_prompt="""You are an expert code reviewer. 
    When asked to review code:
    1. Read the changed files
    2. Search for related code to understand context
    3. Run tests to verify correctness
    4. Provide detailed feedback on:
       - Code quality and readability
       - Potential bugs
       - Performance considerations
       - Security concerns
    Be constructive and specific.""",
    tools=code_review_tools,
)

# 使用
feedback = review_agent.chat("Please review the changes in src/auth.py")

与其他SDK对比

特性	Claude Agent SDK	OpenAI Agents SDK	LangChain
模型支持	Claude系列	GPT系列	多家
工具协议	tool_use + MCP	function calling	多种适配
Extended Thinking	原生支持	不支持	不支持
多智能体	手动实现	Handoff原生	LangGraph
可观测性	日志	Trace	LangSmith
Computer Use	原生支持	不支持	不支持

最佳实践

1. Prompt工程

# 结构化的系统提示
system_prompt = """You are a specialized agent for {domain}.

## Capabilities
You have access to the following tools:
{tool_descriptions}

## Guidelines
1. Always verify before making changes
2. Explain your reasoning before acting
3. Handle errors gracefully

## Constraints
- Never execute destructive operations without confirmation
- Stay within the scope of your role
"""

2. 错误处理

# 工具执行的错误处理
def safe_tool_execution(tool_name, tool_input):
    try:
        result = tool_map[tool_name](**tool_input)
        return {"status": "success", "result": result}
    except PermissionError:
        return {"status": "error", "message": "Permission denied"}
    except TimeoutError:
        return {"status": "error", "message": "Operation timed out"}
    except Exception as e:
        return {"status": "error", "message": str(e)}

3. 成本优化

• 使用 claude-haiku 处理简单的路由和分类任务
• 使用 claude-sonnet 处理大多数Agent任务
• 仅在需要深度推理时使用 claude-opus 或 extended thinking
• 合理设置 thinking.budget_tokens

总结

Claude Agent SDK的核心特点：

1. tool_use协议: 结构化、可靠的工具调用
2. MCP集成: 标准化的外部资源连接
3. Extended Thinking: 复杂任务的深度推理
4. Computer Use: 独特的GUI操作能力
5. 安全设计: 内置的安全考量和权限模型

---

评论