from claude_agent_sdk import list_sessionsfor session in list_sessions(directory="/path/to/project", limit=10): print(f"{session.summary} ({session.session_id})")
如果 session_id 不是有效的 UUID 或 tag 在清理后为空，则抛出 ValueError；如果找不到会话，则抛出 FileNotFoundError。
示例
标记会话，然后在稍后的读取中按该标签过滤。传递 None 以清除现有标签。
Copy
from claude_agent_sdk import list_sessions, tag_session# Tag a sessiontag_session("550e8400-e29b-41d4-a716-446655440000", "needs-review")# Later: find all sessions with that tagfor session in list_sessions(directory="/path/to/project"): if session.tag == "needs-review": print(session.summary)
import asynciofrom claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessageasync def main(): async with ClaudeSDKClient() as client: # First question await client.query("What's the capital of France?") # Process response async for message in client.receive_response(): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, TextBlock): print(f"Claude: {block.text}") # Follow-up question - the session retains the previous context await client.query("What's the population of that city?") async for message in client.receive_response(): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, TextBlock): print(f"Claude: {block.text}") # Another follow-up - still in the same conversation await client.query("What are some famous landmarks there?") async for message in client.receive_response(): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, TextBlock): print(f"Claude: {block.text}")asyncio.run(main())
示例 - 使用 ClaudeSDKClient 进行流式输入
Copy
import asynciofrom claude_agent_sdk import ClaudeSDKClientasync def message_stream(): """Generate messages dynamically.""" yield { "type": "user", "message": {"role": "user", "content": "Analyze the following data:"}, } await asyncio.sleep(0.5) yield { "type": "user", "message": {"role": "user", "content": "Temperature: 25°C, Humidity: 60%"}, } await asyncio.sleep(0.5) yield { "type": "user", "message": {"role": "user", "content": "What patterns do you see?"}, }async def main(): async with ClaudeSDKClient() as client: # Stream input to Claude await client.query(message_stream()) # Process response async for message in client.receive_response(): print(message) # Follow-up in same session await client.query("Should we be concerned about these readings?") async for message in client.receive_response(): print(message)asyncio.run(main())
示例 - 使用中断
Copy
import asynciofrom claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, ResultMessageasync def interruptible_task(): options = ClaudeAgentOptions(allowed_tools=["Bash"], permission_mode="acceptEdits") async with ClaudeSDKClient(options=options) as client: # Start a long-running task await client.query("Count from 1 to 100 slowly, using the bash sleep command") # Let it run for a bit await asyncio.sleep(2) # Interrupt the task await client.interrupt() print("Task interrupted!") # Drain the interrupted task's messages (including its ResultMessage) async for message in client.receive_response(): if isinstance(message, ResultMessage): print(f"Interrupted task finished with subtype={message.subtype!r}") # subtype is "error_during_execution" for interrupted tasks # Send a new command await client.query("Just say hello instead") # Now receive the new response async for message in client.receive_response(): if isinstance(message, ResultMessage) and message.subtype == "success": print(f"New result: {message.result}")asyncio.run(interruptible_task())
从文件加载自定义系统提示而不是作为字符串传递的配置。SDK 将其映射到 CLI --system-prompt-file 标志。当提示很大时使用文件形式：SDK 在 CLI 子进程 argv 上传递字符串 system_prompt，这受到 OS 命令行长度限制的限制，然后 SDK 才能发送任何 API 请求。在 Linux 上，单个参数长于大约 128 KB 会在进程生成时失败，出现 Argument list too long。在 Windows 上，整个命令行被限制为大约 32 KB，因此字符串形式在更低的阈值处失败。
Copy
class SystemPromptFile(TypedDict): type: Literal["file"] path: str
# Do not load user, project, or local settings from diskfrom claude_agent_sdk import query, ClaudeAgentOptionsasync for message in query( prompt="Analyze this code", options=ClaudeAgentOptions( setting_sources=[] ),): print(message)
from claude_agent_sdk import query, ClaudeAgentOptionsasync for message in query( prompt="Analyze this code", options=ClaudeAgentOptions( setting_sources=["user", "project", "local"] ),): print(message)
仅加载特定设置源：
Copy
# Load only project settings, ignore user and localasync for message in query( prompt="Run CI checks", options=ClaudeAgentOptions( setting_sources=["project"] # Only .claude/settings.json ),): print(message)
测试和 CI 环境：
Copy
# Ensure consistent behavior in CI by excluding local settingsasync for message in query( prompt="Run tests", options=ClaudeAgentOptions( setting_sources=["project"], # Only team-shared settings permission_mode="bypassPermissions", ),): print(message)
仅 SDK 应用程序：
Copy
# Define everything programmatically.# Pass [] to opt out of filesystem setting sources.async for message in query( prompt="Review this PR", options=ClaudeAgentOptions( setting_sources=[], agents={...}, mcp_servers={...}, allowed_tools=["Read", "Grep", "Glob"], ),): print(message)
加载 CLAUDE.md 项目说明：
Copy
# Load project settings to include CLAUDE.md filesasync for message in query( prompt="Add a new feature following project conventions", options=ClaudeAgentOptions( system_prompt={ "type": "preset", "preset": "claude_code", # Use Claude Code's system prompt }, setting_sources=["project"], # Loads CLAUDE.md from project allowed_tools=["Read", "Write", "Edit"], ),): print(message)
PermissionMode = Literal[ "default", # Standard permission behavior "acceptEdits", # Auto-accept file edits "plan", # Planning mode - explore without editing "dontAsk", # Deny anything not pre-approved instead of prompting "bypassPermissions", # Bypass permission checks; explicit ask rules still prompt (use with caution) "auto", # A model classifier approves or denies each tool call]
EffortLevel
用于指导思考深度的努力级别。
Copy
EffortLevel = Literal[ "low", # Minimal thinking, fastest responses "medium", # Moderate thinking "high", # Deep reasoning "xhigh", # Extended reasoning; falls back to "high" on models that don't support it "max", # Maximum effort]
class ClaudeSDKError(Exception): """Base error for Claude SDK."""
CLINotFoundError
当 Claude Code CLI 未安装或找不到时引发。
Copy
class CLINotFoundError(CLIConnectionError): def __init__( self, message: str = "Claude Code not found", cli_path: str | None = None ): """ Args: message: Error message (default: "Claude Code not found") cli_path: Optional path to the CLI that was not found """
CLIConnectionError
当连接到 Claude Code 失败时引发。
Copy
class CLIConnectionError(ClaudeSDKError): """Failed to connect to Claude Code."""
class CLIJSONDecodeError(ClaudeSDKError): def __init__(self, line: str, original_error: Exception): """ Args: line: The line that failed to parse original_error: The original JSON decode exception """ self.line = line self.original_error = original_error
HookEvent = Literal[ "PreToolUse", # Called before tool execution "PostToolUse", # Called after tool execution "PostToolUseFailure", # Called when a tool execution fails "UserPromptSubmit", # Called when user submits a prompt "Stop", # Called when stopping execution "SubagentStop", # Called when a subagent stops "PreCompact", # Called before message compaction "Notification", # Called for notification events "SubagentStart", # Called when a subagent starts "PermissionRequest", # Called when a permission decision is needed]
class HookContext(TypedDict): signal: Any | None # Future: abort signal support
HookMatcher
用于将 hooks 匹配到特定事件或工具的配置。
Copy
@dataclassclass HookMatcher: matcher: str | None = ( None # Tool name or pattern to match (e.g., "Bash", "Write|Edit") ) hooks: list[HookCallback] = field( default_factory=list ) # List of callbacks to execute timeout: float | None = ( None # Timeout in seconds for all hooks in this matcher (default: 60) )
from claude_agent_sdk import ( ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, TextBlock,)import asyncioclass ConversationSession: """Maintains a single conversation session with Claude.""" def __init__(self, options: ClaudeAgentOptions | None = None): self.client = ClaudeSDKClient(options) self.turn_count = 0 async def start(self): await self.client.connect() print("Starting conversation session. Claude will remember context.") print( "Commands: 'exit' to quit, 'interrupt' to stop current task, 'new' for new session" ) while True: user_input = input(f"\n[Turn {self.turn_count + 1}] You: ") if user_input.lower() == "exit": break elif user_input.lower() == "interrupt": await self.client.interrupt() print("Task interrupted!") continue elif user_input.lower() == "new": # Disconnect and reconnect for a fresh session await self.client.disconnect() await self.client.connect() self.turn_count = 0 print("Started new conversation session (previous context cleared)") continue # Send message - the session retains all previous messages await self.client.query(user_input) self.turn_count += 1 # Process response print(f"[Turn {self.turn_count}] Claude: ", end="") async for message in self.client.receive_response(): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, TextBlock): print(block.text, end="") print() # New line after response await self.client.disconnect() print(f"Conversation ended after {self.turn_count} turns.")async def main(): options = ClaudeAgentOptions( allowed_tools=["Read", "Write", "Bash"], permission_mode="acceptEdits" ) session = ConversationSession(options) await session.start()# Example conversation:# Turn 1 - You: "Create a file called hello.py"# Turn 1 - Claude: "I'll create a hello.py file for you..."# Turn 2 - You: "What's in that file?"# Turn 2 - Claude: "The hello.py file I just created contains..." (remembers!)# Turn 3 - You: "Add a main function to it"# Turn 3 - Claude: "I'll add a main function to hello.py..." (knows which file!)asyncio.run(main())
使用 Hooks 进行行为修改
Copy
from claude_agent_sdk import ( ClaudeSDKClient, ClaudeAgentOptions, HookMatcher, HookContext,)import asynciofrom typing import Anyasync def pre_tool_logger( input_data: dict[str, Any], tool_use_id: str | None, context: HookContext) -> dict[str, Any]: """Log all tool usage before execution.""" tool_name = input_data.get("tool_name", "unknown") print(f"[PRE-TOOL] About to use: {tool_name}") # You can modify or block the tool execution here if tool_name == "Bash" and "rm -rf" in str(input_data.get("tool_input", {})): return { "hookSpecificOutput": { "hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": "Dangerous command blocked", } } return {}async def post_tool_logger( input_data: dict[str, Any], tool_use_id: str | None, context: HookContext) -> dict[str, Any]: """Log results after tool execution.""" tool_name = input_data.get("tool_name", "unknown") print(f"[POST-TOOL] Completed: {tool_name}") return {}async def user_prompt_modifier( input_data: dict[str, Any], tool_use_id: str | None, context: HookContext) -> dict[str, Any]: """Add context to user prompts.""" original_prompt = input_data.get("prompt", "") # Add a timestamp as additional context for Claude to see from datetime import datetime timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") return { "hookSpecificOutput": { "hookEventName": "UserPromptSubmit", "additionalContext": f"[Submitted at {timestamp}] Original prompt: {original_prompt}", } }async def main(): options = ClaudeAgentOptions( hooks={ "PreToolUse": [ HookMatcher(hooks=[pre_tool_logger]), HookMatcher(matcher="Bash", hooks=[pre_tool_logger]), ], "PostToolUse": [HookMatcher(hooks=[post_tool_logger])], "UserPromptSubmit": [HookMatcher(hooks=[user_prompt_modifier])], }, allowed_tools=["Read", "Write", "Bash"], ) async with ClaudeSDKClient(options=options) as client: await client.query("List files in current directory") async for message in client.receive_response(): # Hooks will automatically log tool usage passasyncio.run(main())
实时进度监控
Copy
from claude_agent_sdk import ( ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, ToolUseBlock, ToolResultBlock, TextBlock,)import asyncioasync def monitor_progress(): options = ClaudeAgentOptions( allowed_tools=["Write", "Bash"], permission_mode="acceptEdits" ) async with ClaudeSDKClient(options=options) as client: await client.query("Create 5 Python files with different sorting algorithms") # Monitor progress in real-time async for message in client.receive_response(): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, ToolUseBlock): if block.name == "Write": file_path = block.input.get("file_path", "") print(f"Creating: {file_path}") elif isinstance(block, ToolResultBlock): print("Completed tool execution") elif isinstance(block, TextBlock): print(f"Claude says: {block.text[:100]}...") print("Task completed!")asyncio.run(monitor_progress())
示例用法
基本文件操作（使用 query）
Copy
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ToolUseBlockimport asyncioasync def create_project(): options = ClaudeAgentOptions( allowed_tools=["Read", "Write", "Bash"], permission_mode="acceptEdits", cwd="/home/user/project", ) async for message in query( prompt="Create a Python project structure with setup.py", options=options ): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, ToolUseBlock): print(f"Using tool: {block.name}")asyncio.run(create_project())
错误处理
Copy
from claude_agent_sdk import query, CLINotFoundError, ProcessError, CLIJSONDecodeErrortry: async for message in query(prompt="Hello"): print(message)except CLINotFoundError: print( "Claude Code CLI not found. Try reinstalling: pip install --force-reinstall claude-agent-sdk" )except ProcessError as e: print(f"Process failed with exit code: {e.exit_code}")except CLIJSONDecodeError as e: print(f"Failed to parse response: {e}")
使用客户端的流式模式
Copy
from claude_agent_sdk import ClaudeSDKClientimport asyncioasync def interactive_session(): async with ClaudeSDKClient() as client: # Send initial message await client.query("What's the weather like?") # Process responses async for msg in client.receive_response(): print(msg) # Send follow-up await client.query("Tell me more about that") # Process follow-up response async for msg in client.receive_response(): print(msg)asyncio.run(interactive_session())
使用 ClaudeSDKClient 的自定义工具
Copy
from claude_agent_sdk import ( ClaudeSDKClient, ClaudeAgentOptions, tool, create_sdk_mcp_server, AssistantMessage, TextBlock,)import asynciofrom typing import Any# Define custom tools with @tool decorator@tool("calculate", "Perform mathematical calculations", {"expression": str})async def calculate(args: dict[str, Any]) -> dict[str, Any]: try: result = eval(args["expression"], {"__builtins__": {}}) return {"content": [{"type": "text", "text": f"Result: {result}"}]} except Exception as e: return { "content": [{"type": "text", "text": f"Error: {str(e)}"}], "is_error": True, }@tool("get_time", "Get current time", {})async def get_time(args: dict[str, Any]) -> dict[str, Any]: from datetime import datetime current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") return {"content": [{"type": "text", "text": f"Current time: {current_time}"}]}async def main(): # Create SDK MCP server with custom tools my_server = create_sdk_mcp_server( name="utilities", version="1.0.0", tools=[calculate, get_time] ) # Configure options with the server options = ClaudeAgentOptions( mcp_servers={"utils": my_server}, allowed_tools=["mcp__utils__calculate", "mcp__utils__get_time"], ) # Use ClaudeSDKClient for interactive tool usage async with ClaudeSDKClient(options=options) as client: await client.query("What's 123 * 456?") # Process calculation response async for message in client.receive_response(): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, TextBlock): print(f"Calculation: {block.text}") # Follow up with time query await client.query("What time is it now?") async for message in client.receive_response(): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, TextBlock): print(f"Time: {block.text}")asyncio.run(main())
from claude_agent_sdk import query, ClaudeAgentOptions, SandboxSettingssandbox_settings: SandboxSettings = { "enabled": True, "autoAllowBashIfSandboxed": True, "network": {"allowLocalBinding": True},}async for message in query( prompt="Build and test my project", options=ClaudeAgentOptions(sandbox=sandbox_settings),): print(message)
from claude_agent_sdk import ( query, ClaudeAgentOptions, HookMatcher, PermissionResultAllow, PermissionResultDeny, ToolPermissionContext,)async def can_use_tool( tool: str, input: dict, context: ToolPermissionContext) -> PermissionResultAllow | PermissionResultDeny: # Check if the model is requesting to bypass the sandbox if tool == "Bash" and input.get("dangerouslyDisableSandbox"): # The model is requesting to run this command outside the sandbox print(f"Unsandboxed command requested: {input.get('command')}") if is_command_authorized(input.get("command")): return PermissionResultAllow() return PermissionResultDeny( message="Command not authorized for unsandboxed execution" ) return PermissionResultAllow()# Required: dummy hook keeps the stream open for can_use_toolasync def dummy_hook(input_data, tool_use_id, context): return {"continue_": True}async def prompt_stream(): yield { "type": "user", "message": {"role": "user", "content": "Deploy my application"}, }async def main(): async for message in query( prompt=prompt_stream(), options=ClaudeAgentOptions( sandbox={ "enabled": True, "allowUnsandboxedCommands": True, # Model can request unsandboxed execution }, permission_mode="default", can_use_tool=can_use_tool, hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[dummy_hook])]}, ), ): print(message)
