Claude Agent SDK for Python：用内置 Claude Code 构建编程式 AI Agent

import anyio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="创建一个包含 /api/users 端点的 Flask 应用"):
        print(message)

anyio.run(main)

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient

options = ClaudeAgentOptions(
    system_prompt="你是一位资深 Python 架构师。",
    max_turns=5,
    permission_mode='autoApprove'
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("为该仓库设计微服务架构")
    async for msg in client.receive_response():
        print(msg)

from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient

@tool("deploy", "部署应用到生产环境", {
    "environment": str,
    "branch": str
})
async def deploy_app(args):
    env = args["environment"]
    branch = args["branch"]
    # 在此处运行部署逻辑
    return {"status": f"已将 {branch} 部署到 {env}"}

server = create_sdk_mcp_server(
    name="deployment-tools",
    tools=[deploy_app]
)

options = ClaudeAgentOptions(
    mcp_servers={"deploy": server},
    allowed_tools=["mcp__tools__deploy"]
)

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Bash"],       # 自动批准
    disallowed_tools=["WebFetch"],                   # 始终阻止
    permission_mode='acceptEdits',                   # 自动批准文件编辑
    can_use_tool=lambda tool_name, context: ...     # 自定义逻辑
)

from claude_agent_sdk import HookMatcher

async def block_destructive_commands(input_data, tool_use_id, context):
    if input_data["tool_name"] == "Bash":
        command = input_data["tool_input"].get("command", "")
        dangerous_patterns = ["rm -rf", "drop table", "> /dev/sda"]
        for pattern in dangerous_patterns:
            if pattern in command:
                return {
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": "deny",
                        "permissionDecisionReason": f"检测到危险模式: {pattern}"
                    }
                }
    return {}

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Bash", hooks=[block_destructive_commands])
        ]
    }
)

from claude_agent_sdk import query, ClaudeAgentOptions

async def review_code(pr_diff):
    options = ClaudeAgentOptions(
        system_prompt="""你是一位专家级代码审查员。重点关注：
        - 安全漏洞
        - 性能问题
        - 代码风格和规范
        - 测试覆盖缺口""",
        allowed_tools=["Read", "Grep"],
        permission_mode='editFiles',
        max_turns=3
    )
    
    results = []
    async for message in query(
        prompt=f"审查以下 PR 差异:\n{pr_diff}",
        options=options
    ):
        results.append(str(message))
    
    return "\n".join(results)

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def self_heal_alert(alert_json):
    options = ClaudeAgentOptions(
        system_prompt="你是一位 SRE agent。诊断并修复生产问题。",
        allowed_tools=["Bash", "Read", "Edit", "code_run"],
        permission_mode='acceptEdits'
    )
    
    async with ClaudeSDKClient(options=options) as client:
        await client.query(f"调查告警: {alert_json}")
        async for msg in client.receive_response():
            log_action(msg)

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

class AnalyticsAgent:
    def __init__(self):
        self.options = ClaudeAgentOptions(
            system_prompt="""你是一位数据分析师助手。
            将自然语言请求转换为 SQL 查询、数据可视化和洞察。""",
            allowed_tools=["code_run", "Read", "Write"],
            cwd="/data/reports/"
        )
    
    async def analyze(self, question: str) -> str:
        async with ClaudeSDKClient(options=self.options) as client:
            await client.query(question)
            return await client.receive_response()

def custom_permission_checker(tool_name: str, context: dict) -> bool:
    """自定义权限检查逻辑"""
    # 禁止在非生产环境执行危险操作
    if tool_name == "Bash":
        env = os.environ.get("DEPLOY_ENV", "")
        if env != "production" and "DROP TABLE" in context.get("command", ""):
            return False
    
    # 限制代码执行的文件类型
    if tool_name == "code_run":
        file_ext = context.get("file_extension", "")
        allowed_extensions = [".py", ".js", ".sh"]
        if file_ext not in allowed_extensions:
            return False
    
    return True

options = ClaudeAgentOptions(
    can_use_tool=custom_permission_checker
)

from claude_agent_sdk import HookMatcher, HookSpecificOutput

class EnterpriseHookHandler:
    def __init__(self):
        self.blocklist_patterns = [
            r'\b(?:DROP|DELETE|TRUNCATE)\s+(TABLE|DATABASE)',
            r'rm\s+-rf\s+/',
            r'curl.*\|.*bash',
            r'wget.*\|\s*sh'
        ]
        self.rate_limiter = RateLimiter(max_requests=100, window_seconds=60)
    
    async def pre_tool_use(self, input_data, tool_use_id, context):
        """在工具执行前拦截"""
        tool_name = input_data.get("tool_name", "")
        
        # 速率限制检查
        if not self.rate_limiter.allow_request():
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": "速率限制阈值已达"
                }
            }
        
        # 危险命令检测
        if tool_name == "Bash":
            command = input_data.get("tool_input", {}).get("command", "")
            import re
            for pattern in self.blocklist_patterns:
                if re.search(pattern, command, re.IGNORECASE):
                    return {
                        "hookSpecificOutput": {
                            "hookEventName": "PreToolUse",
                            "permissionDecision": "deny",
                            "permissionDecisionReason": f"触发安全规则: {pattern}"
                        }
                    }
        
        return {}
    
    async def post_tool_use(self, result, context):
        """在工具执行后记录审计日志"""
        audit_logger.log({
            "timestamp": datetime.now().isoformat(),
            "tool_result": str(result)[:500],
            "context_hash": hash(str(context))
        })
        return {}

handler = EnterpriseHookHandler()

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Bash", hooks=[handler.pre_tool_use])
        ],
        "PostToolUse": [
            HookMatcher(matcher="*", hooks=[handler.post_tool_use])
        ]
    }
)

pip install claude-agent-sdk

curl -fsSL https://claude.ai/install.sh | bash

export ANTHROPIC_API_KEY="sk-ant-your-key-here"

# hello_agent.py
from claude_agent_sdk import query

import anyio

async def main():
    async for message in query(
        prompt="编写一个迭代计算斐波那契数列的 Python 函数",
        options=ClaudeAgentOptions(max_turns=1)
    ):
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    print(block.text)

anyio.run(main)

python hello_agent.py

# deploy_agent.py
from claude_agent_sdk import (
    tool, create_sdk_mcp_server, 
    ClaudeAgentOptions, ClaudeSDKClient
)

@tool("git_status", "查看当前 git 状态", {})
async def git_status(_args):
    import subprocess
    result = subprocess.run(["git", "status", "--short"], 
                          capture_output=True, text=True)
    return {"content": [{"type": "text", "text": result.stdout}]}

server = create_sdk_mcp_server(
    name="git-tools", 
    tools=[git_status]
)

options = ClaudeAgentOptions(
    mcp_servers={"git": server},
    allowed_tools=["mcp__tools__git_status"]
)

import anyio

async def main():
    async with ClaudeSDKClient(options=options) as client:
        await client.query("显示当前 git 状态并建议改进")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

from claude_agent_sdk import (
    CLINotFoundError,      # Claude Code 未安装
    CLIConnectionError,    # 连接失败
    ProcessError,          # 子进程退出码异常
    CLIJSONDecodeError     # 响应解析失败
)

try:
    async for message in query(prompt="执行迁移"):
        process_result(message)
except CLINotFoundError:
    print("Claude Code 未安装。请运行: pip install claude-agent-sdk")
except CLIConnectionError as e:
    print(f"连接错误: {e}。请检查 API 密钥。")
except ProcessError as e:
    print(f"进程失败，退出码 {e.exit_code}")
except CLIJSONDecodeError as e:
    print(f"解析错误: {e}")

import time
import asyncio

async def query_with_retry(prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            async for message in query(prompt=prompt):
                yield message
            break  # 成功则跳出循环
        except (CLIConnectionError, ProcessError) as e:
            wait_time = (2 ** attempt) * 0.5  # 0.5s, 1s, 2s...
            logger.warning(f"第 {attempt + 1} 次尝试失败: {e}，{wait_time:.1f}s 后重试")
            await asyncio.sleep(wait_time)
    else:
        raise RuntimeError(f"所有 {max_retries} 次重试均失败")