Claude Code 开源项目/工具分享

一个用于与 Claude CLI 工具交互的 Python SDK。

Codesys 提供同步和异步类：

Agent——增强的同步 SDK，具备专业级功能（支持 MCP、高级工具管理、重试、超时、结构化响应等）。（codesys/agent.py）

AsyncAgent——功能完全相同的全异步版本，通过 async/await 支持非阻塞工作流。（codesys/async_agent.py）

本 README 统一提供这两个版本的文档。如果你只想看看它们的实际用法，请直接跳到快速入门部分。

快速入门：同步与异步

pip install codesys

模型名称：claude-opus-4-20250514 claude-sonnnet-4-20250514 claude-3-5-haiku-20241022

运行 export ANTHROPIC_MODEL=claude-3-5-haiku-20241022 进行设置。

必须安装 Claude CLI 工具，确保它在你的 PATH 中可用，并已配置 API 密钥。

同步快速入门

from codesys import Agent
import os

# Initialize with a working directory set to the current working directory
agent = Agent(working_dir=os.getcwd())

# This can be a prompt string or claude code command (treat it as your claude code input)
lines = agent.run("""/init""", stream=True)

异步快速入门

import asyncio
from codesys import AsyncAgent

async def main():
    agent = AsyncAgent()
    result = await agent.run("Hello, Claude!")
    print(result)

asyncio.run(main())

我发现，使用这个 SDK 最有效的方法是模拟我使用 Claude Code 时的实际工作流，而这套工作流的效果非常出色。

工作流很简单：先探索代码库并规划任务，然后执行计划。

#!/usr/bin/env python3

import os
from codesys import Agent

# Configuration - modify these values as needed
WORKING_DIR = os.getcwd()  # Use the current working directory

USER_MESSAGE = """Your super long, complex task here."""

def generate_plan_and_execute():
    """Generate a plan and then execute it using the same conversation session."""
    agent = Agent(working_dir=WORKING_DIR)

# Step 1: Generate the plan
    print("Generating plan...")
    prompt = f'''
generate a plan into plan.md file given the following task:
<task>
{USER_MESSAGE}
</task>
Given this task, explore the codebase and create a plan for the implementation into plan.md for our developer to accomplish this task step by step. ultrathink
'''
    agent.run(prompt, stream=True)

# Step 2: Execute the plan continuing the same conversation
    print("\nExecuting plan from plan.md...")
    prompt = '''
Implement the task laid out in plan.md: ultrathink
'''
    agent.run_convo(prompt, stream=True)

if __name__ == "__main__":
    print(f"Working directory: {WORKING_DIR}")
    print(f"Task: {USER_MESSAGE}")
    generate_plan_and_execute()

Claude CLI 工具的简洁接口

支持 Claude CLI 的所有选项

支持自动或手动流式输出

可自定义工具访问权限

支持具有会话连续性的对话管理

支持按 ID 恢复指定对话

Agent(working_dir=None, allowed_tools=None)

working_dir（str，可选）：Claude 使用的工作目录。默认为当前目录。

allowed_tools（list，可选）：允许 Claude 使用的工具列表。默认为 ["Edit", "Bash", "Write"]。

run(prompt, stream=False, output_format=None, additional_args=None, auto_print=True, continue_session=False, session_id=None)

使用指定的提示词运行 Claude。

prompt（str）：发送给 Claude 的提示词。

stream（bool）：如果为 True，则处理流式输出。如果为 False，则返回完整输出。

output_format（str，可选）：可选的输出格式（例如 "stream-json"）。

additional_args（dict，可选）：传递给 Claude CLI 的额外参数。

auto_print（bool）：如果为 True 且 stream=True，则自动打印输出。如果为 False，则需要手动处理流式输出。

continue_session（bool）：如果为 True，则继续最近一次 Claude 会话。

session_id（str，可选）：如果提供，则恢复具有此 ID 的指定 Claude 会话。

如果 stream=False：以字符串形式返回完整输出。

如果 stream=True 且 auto_print=False：返回一个 subprocess.Popen 对象，以便手动处理流式输出。

如果 stream=True 且 auto_print=True：自动打印输出，并以列表形式返回收集到的各行内容。

run_with_tools(prompt, tools, stream=False, auto_print=True, continue_session=False, session_id=None)

使用指定的允许工具运行 Claude。

prompt（str）：发送给 Claude 的提示词。

tools（list）：允许 Claude 使用的工具列表。

stream（bool）：如果为 True，则处理流式输出。

auto_print（bool）：如果为 True 且 stream=True，则自动打印输出。

continue_session（bool）：如果为 True，则继续最近一次 Claude 会话。

session_id（str，可选）：如果提供，则恢复具有此 ID 的指定 Claude 会话。

如果 stream=False：以字符串形式返回完整输出。

如果 stream=True 且 auto_print=False：返回一个 subprocess.Popen 对象。

如果 stream=True 且 auto_print=True：自动打印输出，并返回收集到的各行内容。

run_convo(prompt, **kwargs)

继续最近一次 Claude 对话。此方法会保持与上一次交互相同的会话状态，从而支持能够感知上下文的后续提示词。

prompt（str）：发送给 Claude 的提示词。

**kwargs：传递给 run 方法的额外参数（stream、output_format 等）。

返回类型与 run 方法相同，具体取决于所使用的参数。

resume_convo(session_id, prompt, **kwargs)

按 ID 恢复指定的 Claude 对话。这样，即使已经启动了其他会话，你仍然可以返回之前的对话。

session_id（str）：要恢复的会话 ID。

prompt（str）：发送给 Claude 的提示词。

**kwargs：传递给 run 方法的额外参数（stream、output_format 等）。

返回类型与 run 方法相同，具体取决于所使用的参数。

get_last_session_id()

获取最近一次 Claude 运行的会话 ID。适合用于保存会话 ID，以便稍后恢复对话。

如果会话 ID 可用，则返回该 ID；否则返回 None。

示例：自动流式输出

from codesys import Agent

agent = Agent()
# This will automatically print the output line by line
lines = agent.run("Generate a short story", stream=True)

示例：手动处理流式输出并解析 JSON

from codesys import Agent
import json

agent = Agent()
process = agent.run("Generate a short story", stream=True, output_format="stream-json", auto_print=False)

for line in process.stdout:
    if line.strip():
        try:
            data = json.loads(line)
            print(data.get("content", ""))
        except json.JSONDecodeError:
            print(f"Error parsing JSON: {line}")

from codesys import Agent

# Initialize with a working directory
agent = Agent(working_dir="/Users/seansullivan/codesys/")

# Run Claude with a prompt and automatically print streaming output
lines = agent.run("create another example of example1_custom_tools.py which shows how to use read only tools. note the source code of the sdk in codesys/agent.py", stream=True)

"""
Example 1: Customizing tools during initialization

This example demonstrates how to initialize an Agent with only specific tools.
"""

from codesys import Agent

# Initialize with only specific tools
restricted_agent = Agent(
    working_dir="./",
    allowed_tools=["Edit", "Write", "View"]  # Only allow editing, writing files and viewing
)  # Implementation in agent.py lines 19-39

print(f"Agent initialized with tools: {restricted_agent.allowed_tools}")

from codesys import Agent

# Initialize with default tools
agent = Agent(working_dir="./")  # Implementation in agent.py lines 19-39
print(f"Default tools: {agent.allowed_tools}")

# Run with only specific tools for one operation
bash_only_response = agent.run_with_tools(
    prompt="List files in the current directory",
    tools=["Bash"],  # Only allow Bash for this specific run
    stream=False
)  # Implementation in agent.py lines 132-155

print(f"Tools after run_with_tools: {agent.allowed_tools}  # Original tools are restored")

"""
Example 3: Manual handling of streaming output

This example demonstrates how to manually handle streaming output from the agent.
"""

from codesys import Agent
import json
import time

# Initialize an agent
agent = Agent(working_dir="./")

# Get a process for streaming manually
process = agent.run(
    prompt="Explain what an LLM Agent is in 3 sentences",
    stream=True,
    auto_print=False  # Don't auto-print, we'll handle the output manually
)  # Implementation in agent.py lines 41-96 (stream=True, auto_print=False path)

print("手动处理流式输出，逐行处理：")
for i, line in enumerate(process.stdout):
    # 解析 JSON 行
    try:
        data = json.loads(line)
        # 对每一段输出执行某些操作
        print(f"第 {i+1} 行：{data.get('content', '')}")
    except json.JSONDecodeError:
        print(f"原始行：{line}")

# 模拟处理耗时
    time.sleep(0.1)
    # 与 agent.py 第 98-116 行对比（自动处理流式输出）

"""
示例 4：使用输出格式和附加参数

此示例演示如何使用不同的输出格式并传递附加参数。
"""

from codesys import Agent

# 初始化智能体
agent = Agent(working_dir="./")

# 使用自定义输出格式和附加参数运行
response = agent.run(
    prompt="关于这个代码库，你能告诉我些什么？",
    output_format="json",  # 请求 JSON 输出
    additional_args={
        "temperature": 0.7,     # 设置 temperature
        "max-tokens": 500,      # 限制输出 token 数量
        "silent": True          # 隐藏进度输出
    }
)  # 实现位于 agent.py 第 41-70 行（output_format 处理）和第 74-80 行（additional_args）

print(f"响应类型：{type(response)}")
print("响应的前 100 个字符：", response[:100] if isinstance(response, str) else "不是字符串")

"""
示例 5：使用 run_convo 和 resume_convo 进行多轮对话

此示例演示如何继续与 Claude 对话并保持上下文。
"""

from codesys import Agent
import time

# 初始化智能体
agent = Agent(working_dir="./")

# 开始新对话
print("正在开始新对话...")
response1 = agent.run(
    prompt="分析这个项目的结构。它的主要组件有哪些？",
    stream=True
)

# 通过追问继续同一个对话
print("\n通过追问继续对话...")
response2 = agent.run_convo(
    prompt="你对这个代码库有哪些改进建议？",
    stream=True
)  # 实现位于 agent.py 第 184-197 行

# 获取会话 ID，以便稍后使用
session_id = agent.get_last_session_id()
print(f"\n会话 ID：{session_id}")

# 开始另一个对话
print("\n正在开始一个新的、无关的对话...")
agent.run(
    prompt="介绍一下 Python 的类型提示系统。",
    stream=True
)

# 稍后通过 ID 恢复原对话
print("\n正在恢复最初关于代码库改进的对话...")
agent.resume_convo(
    session_id=session_id,
    prompt="能否详细说明你提出的第一项改进建议？",
    stream=True
)  # 实现位于 agent.py 第 199-211 行

这个增强版 Claude Code SDK 实现了分析指南中的所有改进，同时与现有代码保持完全向后兼容。

MCP 服务器支持：本地和远程 Model Context Protocol 服务器

高级工具管理：工具组、策略和细粒度控制

结构化响应解析：用于 JSON 响应的丰富数据类

增强的错误处理：提供更便于调试的特定异常类型

专业功能：速率限制、重试逻辑、超时

系统提示词支持：自定义和追加系统提示词

流式传输增强：针对不同消息类型的自定义处理器

✅ 向后兼容

所有现有示例均可保持不变地运行：

plan_and_execute.py ✓

example8_share_state.py ✓

📦 安装与使用

使用当前 SDK（推荐）

当前的 agent.py 非常适合大多数使用场景：

from codesys import Agent

# 仍然和以前完全一样正常工作
agent = Agent(working_dir="./")
response = agent.run("你好", stream=True)

使用增强版

如需高级功能，请使用增强版智能体：

from enhanced_agent import Agent

# 使用新功能进行增强版初始化
agent = Agent(
    working_dir="./",
    allowed_tools=["View", "Edit", "Bash"],
    disallowed_tools=["Write"],  # 新增：显式拒绝列表
    max_turns=5,                 # 新增：对话轮次限制
    rate_limit_delay=0.2,        # 新增：速率限制
    max_retries=3,               # 新增：重试逻辑
    timeout=30                   # 新增：默认超时时间
)

🔧 增强功能指南

1. MCP 服务器支持

添加 Model Context Protocol 服务器以扩展能力：

# 本地 MCP 服务器
agent.add_local_mcp_server(
    "filesystem",
    command=["python", "-m", "mcp_filesystem"],
    args=["--root", "./"],
    env={"MCP_LOG_LEVEL": "INFO"}
)

# 远程 MCP 服务器
agent.add_remote_mcp_server(
    "database",
    url="http://localhost:8080/mcp",
    auth={"token": "your-token"}
)

# 使用 MCP 工具
response = agent.run_with_mcp(
    "使用 MCP 文件系统工具列出文件",
    mcp_tools=["filesystem__list_files", "filesystem__read_file"]
)

2. 系统提示词控制

覆盖默认系统提示词，或在其后追加内容：

# 覆盖系统提示词
response = agent.run(
    "审查这段代码中的安全问题",
    system_prompt="你是一名安全专家。请重点关注漏洞。",
    verbose=True,
    timeout=60
)

# 追加系统提示词
response = agent.run(
    "分析这个函数",
    append_system_prompt="以 JSON 格式提供指标。",
    output_format="json"
)

3. 结构化响应解析

获取丰富的已解析响应对象，而非原始字符串：

# 获取结构化响应
structured = agent.run_with_structured_response(
    "分析这个代码库并返回指标"
)

print(f"会话 ID：{structured.session_id}")
print(f"消息数量：{len(structured.messages)}")
print(f"工具调用次数：{len(structured.tool_calls)}")
print(f"最终文本：{structured.final_text}")

# 访问每条消息
for message in structured.messages:
    print(f"角色：{message.role}")
    print(f"内容：{message.content}")

4. 高级工具管理

使用工具组和策略实现更好的控制：

# 工具组（预定义集合）
response = agent.run_with_tool_groups(
    "检查项目结构",
    tool_groups=["file_ops", "system"]  # Edit、View、Write、Bash、LSTool
)

# 可用工具组：
# - file_ops：Edit、View、Write
# - system：Bash、LSTool
# - search：GrepTool、GlobTool
# - batch：BatchTool、MultiEdit
# - notebook：NotebookEdit、NotebookRead
# - web：WebFetchTool
# - agent：AgentTool

# 自定义工具策略
tool_manager = ToolManager()
tool_manager.add_tool_policy("Bash", "allow")
tool_manager.add_tool_policy("Write", "deny")

agent = Agent(tool_manager=tool_manager)
response = agent.run_with_tool_policies("安全地分析文件")

5. 增强的错误处理

通过特定异常类型实现更好的错误管理：

try:
    response = agent.run_with_retry(
        "复杂的分析任务",
        timeout=30,
        max_turns_override=10
    )
except ClaudeTimeoutError as e:
    print(f"Request timed out: {e}")
except ClaudeAuthenticationError as e:
    print(f"Authentication failed: {e}")
except ClaudeToolError as e:
    print(f"Tool usage error: {e}")
except ClaudeMCPError as e:
    print(f"MCP server error: {e}")
except ClaudeSDKError as e:
    print(f"General SDK error: {e}")

6. 使用自定义处理器进行流式传输

处理不同类型的流式消息：

def text_handler(text):
    print(f"[TEXT] {text}", end="")

def tool_handler(tool_call):
    print(f"\n[TOOL] {tool_call.get('name', 'unknown')} called")

def error_handler(error):
    print(f"\n[ERROR] {error}")

# 使用自定义处理器进行流式传输
result = agent.run_streaming_with_handlers(
    "编写一个带有说明的 Python 脚本",
    text_handler=text_handler,
    tool_handler=tool_handler,
    error_handler=error_handler
)

7. 速率限制和重试逻辑

为生产环境内置弹性机制：

# 配置速率限制和重试
agent = Agent(
    rate_limit_delay=0.5,  # 请求之间至少间隔 500ms
    max_retries=5          # 失败时最多重试 5 次
)

# 使用指数退避自动重试
response = agent.run_with_retry(
    "分析这个复杂的代码库",
    timeout=120  # 超时时间为 2 分钟
)

阶段 1：继续使用当前 SDK

现有代码可以正常工作，无需任何更改：

# 继续和以前完全一样正常工作
from codesys import Agent
agent = Agent(working_dir="./")
response = agent.run("你好", stream=True)

阶段 2：逐步添加增强功能

导入增强版智能体并开始使用新功能：

from enhanced_agent import Agent

# 从基础增强功能开始
agent = Agent(
    working_dir="./",
    max_turns=5,        # 添加轮次限制
    timeout=30          # 添加默认超时时间
)

# 所有现有方法仍然可用
response = agent.run_convo("继续我们的讨论")

阶段 3：使用高级功能

逐步采用 MCP、工具策略和结构化响应：

# 添加 MCP 服务器
agent.add_local_mcp_server("db", ["python", "db_server.py"])

## 使用结构化响应
```python
structured = agent.run_with_structured_response("analyze project")

# 使用工具组
response = agent.run_with_tool_groups("check files", ["file_ops"])

📊 当前 vs 增强版本对比

当前 SDK 适合：

快速原型和测试

简单的自动化脚本

基础的 Claude 交互

学习和实验

增强版 SDK 适合：

生产应用

具有多个工具的复杂工作流

与外部系统（MCP）的集成

企业级部署

需要可靠性和监控的应用

查看示例文件：

simple_enhanced_demo.py - 展示概念和向后兼容性

enhanced_agent.py - 完整的增强实现

你的现有示例 - 继续保持不变

🔧 配置选项

增强型 Agent 参数

Agent(
    working_dir="./",                    # 工作目录
    allowed_tools=["View", "Edit"],      # 允许的工具列表
    disallowed_tools=["Write"],          # 显式禁用列表
    max_turns=5,                         # 最大会话轮数
    mcp_config_path="mcp.json",          # MCP 配置文件
    permission_prompt_tool="auth_tool",  # MCP 权限工具
    prompt_for_key=False,                # 交互式 API 密钥提示
    default_api_key=None,                # 默认 API 密钥
    rate_limit_delay=0.1,                # 速率限制（秒）
    max_retries=3,                       # 重试次数
    tool_manager=custom_manager          # 自定义工具管理器
)

agent.run(
    prompt="Your prompt",
    stream=False,                        # 启用流式传输
    output_format="json",                # 输出格式
    system_prompt="Custom prompt",       # 覆盖系统提示
    append_system_prompt="Addition",     # 追加到系统提示
    timeout=30,                          # 请求超时
    verbose=True,                        # 详细日志
    max_turns_override=10,               # 覆盖最大轮数
    allowed_tools_override=["View"],     # 覆盖允许的工具
    mcp_config_path="custom.json",       # 覆盖 MCP 配置
    permission_prompt_tool="tool"        # 覆盖权限工具
)

📈 性能和可靠性

增强 SDK 包括：

指数退避的重试机制

速率限制以防止 API 节流

超时保护以防止请求挂起

支持自定义处理程序的内存高效流式传输

自动清理临时 MCP 配置文件

全面的日志记录以便调试

增强型 agent 保持了你原始 SDK 的相同设计原则：

默认简单 - 基础用法保持简单

需要时强大 - 高级功能可用

向后兼容 - 现有代码始终有效

安全 - API 密钥从所有输出中过滤

文档齐全 - 清晰的示例和文档字符串

与原始 SDK 相同的许可证。

这是 Agent 类的异步版本，提供相同的全面功能，但对非阻塞操作提供完整的 async/await 支持。

✅ 完整的 Async/Await 支持 - 所有方法都是异步且非阻塞的 ✅ 并行执行 - 并发运行多个 Claude 请求 ✅ 异步流式传输 - 使用异步迭代进行流式响应 ✅ 相同的 API - 与同步版本相同的接口 ✅ 增强的错误处理 - 异步感知的异常处理 ✅ 速率限制 - 使用 asyncio.sleep 的异步速率限制 ✅ MCP 支持 - 完整的模型上下文协议支持 ✅ 工具管理 - 高级工具过滤和策略

import asyncio
from codesys import AsyncAgent

async def main():
    agent = AsyncAgent()
    result = await agent.run("Hello, Claude!")
    print(result)

asyncio.run(main())

与同步版本的关键差异

async def basic_example():
    agent = AsyncAgent()
    result = await agent.run("What is the capital of France?")
    print(result)

async def parallel_example():
    agent = AsyncAgent()

    prompts = [
        "What is 2 + 2?",
        "What is the capital of Japan?",
        "Explain photosynthesis briefly"
    ]

    # 并行运行所有请求
    tasks = [agent.run(prompt) for prompt in prompts]
    results = await asyncio.gather(*tasks)

    for result in results:
        print(result[:50] + "...")

async def streaming_example():
    agent = AsyncAgent()

    lines = await agent.run(
        "Write a poem about programming",
        stream=True,
        auto_print=True
    )
    print(f"Collected {len(lines)} lines")

自定义异步处理程序

async def handlers_example():
    agent = AsyncAgent()

    def text_handler(text):
        print(f"[TEXT]: {text}")

    def tool_handler(tool_call):
        print(f"[TOOL]: {tool_call['name']}")

    result = await agent.run_streaming_with_handlers(
        "Calculate 2 + 2",
        text_handler=text_handler,
        tool_handler=tool_handler
    )

对话连续性

async def conversation_example():
    agent = AsyncAgent()

    # 第一条消息
    await agent.run("My name is Alice")

    # 继续对话
    response = await agent.run_convo("What's my name?")
    print(response)

async def structured_example():
    agent = AsyncAgent()

    response = await agent.run_with_structured_response(
        "Explain quantum computing"
    )

    print(f"Session: {response.session_id}")
    print(f"Messages: {len(response.messages)}")
    print(f"Text: {response.final_text}")

async def tools_example():
    agent = AsyncAgent()

    # 用特定工具运行
    result = await agent.run_with_tools(
        "List files in current directory",
        tools=["Bash", "LSTool"]
    )

    # 用工具组运行
    result2 = await agent.run_with_tool_groups(
        "Search for Python files",
        tool_groups=["file_ops", "search"]
    )

MCP（模型上下文协议）

async def mcp_example():
    agent = AsyncAgent()

    # 添加 MCP 服务器
    agent.add_local_mcp_server(
        "my_server",
        command=["python", "-m", "my_mcp_server"]
    )

    # 使用 MCP 工具
    result = await agent.run_with_mcp(
        "Use my_server to get data",
        mcp_tools=["my_server__get_data"]
    )

错误处理和重试

async def retry_example():
    agent = AsyncAgent(max_retries=3)

    try:
        result = await agent.run_with_retry("Hello Claude!")
        print(result)
    except ClaudeSDKError as e:
        print(f"Error: {e}")