ClaudeCode - DSPy Module for Claude Code SDK

A DSPy module that wraps the Claude Code Python SDK with a signature-driven interface. Each agent instance maintains a stateful conversation session, making it perfect for multi-turn agentic workflows.

Reference to the original Claude Agent Python SDK: Claude Agent Python SDK

Features

• Signature-driven - Use DSPy signatures for type safety and clarity
• Stateful sessions - Each agent instance = one conversation session
• Smart schema handling - Automatically handles str vs Pydantic outputs
• Rich outputs - Get typed results + execution trace + token usage
• Multi-turn conversations - Context preserved across calls
• Enhanced prompts - Automatically includes signature docstrings + InputField/OutputField descriptions for better context
• Async support - Both sync and async execution modes

Prerequisites:

• Python 3.10+
• Anthropic API key set in ANTHROPIC_API_KEY environment variable

Note: The Claude Code CLI is automatically bundled with the claude-agent-sdk package - no separate installation required! The SDK uses the bundled CLI by default. If you prefer to use a system-wide installation or a specific version:

• Install separately: curl -fsSL https://claude.ai/install.sh | bash
• Specify custom path: Pass cli_path="/path/to/claude" when creating the agent

Installation

# install with uv
uv add claude-agent-sdk dspy

# or with pip
pip install claude-agent-sdk dspy

Quick Start

Basic String Output

from claude_dspy import ClaudeCode

# create agent
agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory="."
)

# use it
result = agent(message="What files are in this directory?")
print(result.answer)  # String response
print(result.trace)   # Execution items
print(result.usage)   # Token counts

Structured Output with Pydantic

from claude_dspy import ClaudeCode
from pydantic import BaseModel, Field
import dspy

class BugReport(BaseModel):
    severity: str = Field(description="critical, high, medium, or low")
    description: str
    affected_files: list[str]

# Use class-based signature (recommended)
class BugReportSignature(dspy.Signature):
    """Analyze bugs and generate report."""
    message: str = dspy.InputField()
    report: BugReport = dspy.OutputField()

agent = ClaudeCode(
    signature=BugReportSignature,
    working_directory="."
)

result = agent(message="Analyze the bug in error.log")
print(result.report.severity)       # Typed access!
print(result.report.affected_files)

Note: String signatures like "message:str -> report:BugReport" only work with built-in types unless you use dspy.Signature(). For custom Pydantic models, either:

• Use dspy.Signature("...")
• Use class-based signatures (recommended)

API Reference

ClaudeCode

Main agent class that extends dspy.Module.

class ClaudeCode(dspy.Module):
    def __init__(
        self,
        **kwargs,
    )

Parameters:

Core Configuration:

• signature (required) - DSPy signature defining input/output fields (must have exactly 1 input and 1 output)
• model - Claude model to use (default: "claude-opus-4-5-20251101")
• working_directory - Directory where Claude will execute commands (default: ".")
• permission_mode - Permission mode: "default", "acceptEdits", "plan", "bypassPermissions"
• allowed_tools - List of allowed tool names (e.g., ["Read", "Write", "Bash", "Glob", "Grep"]). See Available Tools section for complete list.
• disallowed_tools - List of disallowed tool names
• sandbox - Sandbox configuration dict. See SDK docs for details.
• system_prompt - Custom system prompt or preset config
• api_key - Anthropic API key (falls back to ANTHROPIC_API_KEY env var)

MCP Servers:

• mcp_servers - MCP server configurations for custom tools. See MCP section below.

Session Management:

• continue_conversation - Continue the most recent conversation (default: False)
• resume - Session ID to resume from a previous session
• max_turns - Maximum number of conversation turns
• fork_session - Fork to a new session when resuming (default: False)

Advanced Options:

• permission_prompt_tool_name - MCP tool name for permission prompts
• settings - Path to custom settings file
• add_dirs - Additional directories Claude can access
• env - Environment variables to pass to Claude Code
• extra_args - Additional CLI arguments
• max_buffer_size - Maximum bytes when buffering CLI stdout
• cli_path - Custom path to Claude Code CLI executable

Callbacks and Hooks:

• stderr - Callback function for stderr output: Callable[[str], None]
• can_use_tool - Permission callback for tool usage control
• hooks - Hook configurations for intercepting events. See SDK docs for details.

User and Settings:

• user - User identifier
• include_partial_messages - Include partial message streaming events (default: False)
• setting_sources - Which settings to load: ["user", "project", "local"]

Subagents and Plugins:

• agents - Programmatically defined subagents
• plugins - Custom plugins to load

Methods

__call__(**kwargs) -> Prediction (or forward)

Execute the agent with an input message.

Arguments:

• **kwargs - Must contain the input field specified in signature

Returns:

• Prediction object with:
  • Typed output field - Named according to signature (e.g., result.answer)
  • trace - list[TraceItem] - Execution trace
  • usage - Usage - Token usage statistics

Example:

agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory="."
)

result = agent(message="Hello")
print(result.answer)     # Access typed output
print(result.trace)      # List of execution items
print(result.usage)      # Token usage stats

aforward(**kwargs) -> Prediction

Async version of __call__() for use in async contexts.

Example:

async def main():
    agent = ClaudeCode(
        signature="message:str -> answer:str",
        working_directory="."
    )
    result = await agent.aforward(message="Hello")
    print(result.answer)

Properties

session_id: Optional[str]

Get the session ID for this agent instance.

• Returns None until first call
• Persists across multiple calls
• Useful for debugging and logging

Example:

agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory="."
)

print(agent.session_id)  # None

result = agent(message="Hello")
print(agent.session_id)  # 'eb1b2f39-e04c-4506-9398-b50053b1fd83'

Key Implementation Details

Simple Initialization

All configuration is passed as keyword arguments to the ClaudeCode constructor:

from claude_dspy import ClaudeCode

# All parameters are kwargs
agent = ClaudeCode(
    signature="message:str -> answer:str",  # Required
    model="claude-opus-4-5-20251101",       # Optional, defaults shown
    working_directory=".",                   # Optional
    permission_mode="acceptEdits",           # Optional
    allowed_tools=["Read", "Write", "Bash"], # Optional
)

Available Tools

The allowed_tools parameter accepts any valid Claude Code tool name:

File Operations:

• "Read" - Read files and directories
• "Write" - Write and create files
• "Edit" - Edit existing files

Command Execution:

• "Bash" - Execute bash commands

Code Search:

• "Glob" - Search for files by pattern
• "Grep" - Search file contents

Web Tools:

• "WebSearch" - Search the web
• "WebFetch" - Fetch web content

Other Tools:

• "NotebookEdit" - Edit Jupyter notebooks
• And other Claude Code tools...

Advanced Features

Using MCP Servers

MCP (Model Context Protocol) servers allow you to add custom tools to Claude. The SDK supports creating in-process MCP servers with custom tools.

from claude_dspy import ClaudeCode
from claude_agent_sdk import tool, create_sdk_mcp_server
from 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
        }

# Create MCP server
calculator_server = create_sdk_mcp_server(
    name="calculator",
    version="1.0.0",
    tools=[calculate]
)

# Use with ClaudeCode
agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
    mcp_servers={"calc": calculator_server},
    allowed_tools=["mcp__calc__calculate"]  # MCP tools are prefixed with "mcp__<server>__<tool>"
)

result = agent(message="Calculate 123 * 456")
print(result.answer)

Session Management

Resume and continue conversations from previous sessions:

from claude_dspy import ClaudeCode

# First conversation
agent1 = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory="."
)
result1 = agent1(message="Create a file called notes.txt")
session_id = agent1.session_id
print(f"Session ID: {session_id}")

# Resume the same conversation later
agent2 = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
    resume=session_id  # Resume from session ID
)
result2 = agent2(message="What file did we just create?")
print(result2.answer)  # Claude remembers the previous context!

Using Hooks

Intercept and modify tool execution with hooks:

from claude_dspy import ClaudeCode
from typing import Any

async def pre_tool_logger(input_data: dict[str, Any], tool_use_id: str | None, context: Any) -> dict[str, Any]:
    """Log all tool usage before execution."""
    tool_name = input_data.get('tool_name', 'unknown')
    print(f"About to use tool: {tool_name}")

    # Block dangerous commands
    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 {}

agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
    hooks={
        'PreToolUse': [
            {'matcher': 'Bash', 'hooks': [pre_tool_logger]}
        ]
    },
    allowed_tools=["Read", "Write", "Bash"]
)

result = agent(message="List files in this directory")

Loading Project Settings

Control which filesystem settings to load:

from claude_dspy import ClaudeCode

# Load only project settings (e.g., CLAUDE.md files)
agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
    system_prompt={
        "type": "preset",
        "preset": "claude_code"
    },
    setting_sources=["project"],  # Load .claude/settings.json and CLAUDE.md
    allowed_tools=["Read", "Write"]
)

result = agent(message="Add a feature following project conventions")

Usage Patterns

Pattern 1: Multi-turn Conversation

Each agent instance maintains a stateful session:

from claude_dspy import ClaudeCode

agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
)

# Turn 1
result1 = agent(message="What's the main bug?")
print(result1.answer)

# Turn 2 - has context from Turn 1
result2 = agent(message="How do we fix it?")
print(result2.answer)

# Turn 3 - has context from Turn 1 + 2
result3 = agent(message="Write tests for the fix")
print(result3.answer)

# All use same session_id
print(agent.session_id)

Pattern 2: Fresh Context

Want a new conversation? Create a new agent:

from claude_dspy import ClaudeCode

# Agent 1 - Task A
agent1 = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
)
result1 = agent1(message="Analyze bug in module A")

# Agent 2 - Task B (no context from Agent 1)
agent2 = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
)
result2 = agent2(message="Analyze bug in module B")

Pattern 3: Field Descriptions for Enhanced Context

Enhance prompts with signature docstrings and field descriptions - all automatically included in the prompt:

import dspy
from claude_dspy import ClaudeCode

class MySignature(dspy.Signature):
    """Analyze code architecture."""  # Used as task description

    message: str = dspy.InputField(
        desc="Request to process"  # Provides input context
    )
    analysis: str = dspy.OutputField(
        desc="A detailed markdown report with sections: "
        "1) Architecture overview, 2) Key components, 3) Dependencies"  # Guides output format
    )

agent = ClaudeCode(
    signature=MySignature,
    working_directory=".",
)
result = agent(message="Analyze this codebase")

# The prompt sent to Claude will include:
# 1. Task: "Analyze code architecture." (from docstring)
# 2. Input context: "Request to process" (from InputField desc)
# 3. Your message: "Analyze this codebase"
# 4. Output guidance: "Please produce the following output: A detailed markdown report..." (from OutputField desc)

Pattern 4: Inspecting Execution Trace

Access detailed execution information:

from claude_dspy import ClaudeCode, ToolUseItem, ToolResultItem

agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
)

result = agent(message="Fix the bug")

# Filter trace by type
tool_uses = [item for item in result.trace if isinstance(item, ToolUseItem)]
for tool in tool_uses:
    print(f"Tool: {tool.tool_name}")
    print(f"Input: {tool.tool_input}")

tool_results = [item for item in result.trace if isinstance(item, ToolResultItem)]
for result_item in tool_results:
    print(f"Result: {result_item.content}")
    print(f"Error: {result_item.is_error}")

Pattern 5: Token Usage Tracking

Monitor API usage:

result = agent(message="...")

print(f"Input tokens: {result.usage.input_tokens}")
print(f"Cached tokens: {result.usage.cached_input_tokens}")
print(f"Output tokens: {result.usage.output_tokens}")
print(f"Total: {result.usage.total_tokens}")

Pattern 6: Safe Execution with Permissions

Control what the agent can do:

from claude_dspy import ClaudeCode

# Read-only (safest)
agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
    permission_mode="default",
    allowed_tools=["Read"],  # Only allow reading files
)

# Auto-accept file edits
agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
    permission_mode="acceptEdits",
    allowed_tools=["Read", "Write"],  # Allow reading and writing
)

# Full permissions with command execution
agent = ClaudeCode(
    signature="message:str -> answer:str",
    working_directory=".",
    permission_mode="acceptEdits",
    allowed_tools=["Read", "Write", "Bash"],  # All tools enabled
)

Advanced Examples

Example 1: Code Review Agent

from pydantic import BaseModel, Field
from claude_dspy import ClaudeCode

class CodeReview(BaseModel):
    summary: str = Field(description="High-level summary")
    issues: list[str] = Field(description="List of issues found")
    severity: str = Field(description="critical, high, medium, or low")
    recommendations: list[str] = Field(description="Actionable recommendations")

agent = ClaudeCode(
    signature="message:str -> review:CodeReview",
    model="claude-opus-4-5-20251101",
    working_directory="/path/to/project",
    permission_mode="default",
    allowed_tools=["Read"],  # Read-only for code review
)

result = agent(message="Review the changes in src/main.py")

print(f"Severity: {result.review.severity}")
for issue in result.review.issues:
    print(f"- {issue}")

Example 2: Iterative Debugging

from claude_dspy import ClaudeCode

agent = ClaudeCode(
    signature="message:str -> response:str",
    working_directory=".",
    permission_mode="acceptEdits",
    allowed_tools=["Read", "Write", "Bash"],
)

# Turn 1: Find the bug
result1 = agent(message="Find the bug in src/calculator.py")
print(result1.response)

# Turn 2: Propose a fix
result2 = agent(message="What's the best way to fix it?")
print(result2.response)

# Turn 3: Implement the fix
result3 = agent(message="Implement the fix")
print(result3.response)

# Turn 4: Write tests
result4 = agent(message="Write tests for the fix")
print(result4.response)

Example 3: Async Usage

import asyncio
from claude_dspy import ClaudeCode

async def main():
    agent = ClaudeCode(
        signature="message:str -> answer:str",
        working_directory=".",
    )

    # Use aforward in async context
    result = await agent.aforward(message="Analyze this code")
    print(result.answer)

    # Cleanup
    await agent.disconnect()

asyncio.run(main())

Trace Item Types

When accessing result.trace, you'll see various item types:

Type	Fields	Description
AgentMessageItem	text, model	Agent's text response
ThinkingItem	text, model	Agent's internal reasoning
ToolUseItem	tool_name, tool_input, tool_use_id	Tool invocation
ToolResultItem	tool_name, tool_use_id, content, is_error	Tool result
ErrorItem	message, error_type	Error that occurred

How It Works

Signature � Claude Flow

1. Define signature: 'message:str -> answer:str'

2. ClaudeCode validates (must have 1 input, 1 output)

3. __init__ creates ClaudeSDKClient with options

4. forward(message="...") extracts message

5. If output field has desc � append to message

6. If output type ` str � generate JSON schema

7. Call client.query(message) with optional output_format

8. Iterate through receive_response(), collect messages

9. Parse response (JSON if Pydantic, str otherwise)

10. Return Prediction(output=..., trace=..., usage=...)

Output Type Handling

String output:

sig = dspy.Signature('message:str -> answer:str')
# No schema passed to Claude Code
# Response used as-is

Pydantic output:

sig = dspy.Signature('message:str -> report:BugReport')
# JSON schema generated from BugReport
# Schema passed to Claude Code via output_format
# Response parsed with BugReport.model_validate_json()

Prompt Building

ClaudeCode automatically builds rich prompts from your signature to provide maximum context to Claude:

class MySignature(dspy.Signature):
    """Analyze code quality."""  # 1. Task description

    message: str = dspy.InputField(
        desc="Path to file or module"  # 2. Input context
    )
    report: str = dspy.OutputField(
        desc="Markdown report with issues and recommendations"  # 3. Output guidance
    )

agent = ClaudeCode(
    signature=MySignature,
    working_directory="."
)
result = agent(message="Analyze src/main.py")  # 4. Your actual input

The final prompt sent to Claude:

Task: Analyze code quality.

Input context: Path to file or module

Analyze src/main.py

Please produce the following output: Markdown report with issues and recommendations

This automatic context enhancement helps Claude better understand:

• What the overall task is (docstring)
• What the input represents (InputField desc)
• What format the output should have (OutputField desc)

Troubleshooting

Error: "ClaudeCode requires exactly 1 input field"

Your signature has too many or too few fields. ClaudeCode expects exactly one input and one output:

# L Wrong - multiple inputs
sig = dspy.Signature('context:str, question:str -> answer:str')

# Correct - single input
sig = dspy.Signature('message:str -> answer:str')

Error: "Failed to parse Claude response as MyModel"

The model returned JSON that doesn't match your Pydantic schema. Check:

1. Schema is valid and clear
2. Field descriptions are helpful
3. Model has enough context to generate correct structure

Error: "Claude Code CLI not found"

Install Claude Code CLI:

# Visit code.claude.com for installation instructions
# or use npm:
npm install -g @anthropic-ai/claude-code

Async event loop issues

Use aforward() when already in an async context:

# L Don't do this in async context
async def main():
    result = agent(message="...")  # Can cause issues

# � Do this instead
async def main():
    result = await agent.aforward(message="...")

Design Philosophy

Why 1 input, 1 output?

ClaudeCode is designed for conversational agentic workflows. The input is always a message/prompt, and the output is always a response. This keeps the interface simple and predictable.

For complex inputs, compose them into the message:

# Instead of: 'context:str, question:str -> answer:str'
message = f"Context: {context}\n\nQuestion: {question}"
result = agent(message=message)

Why stateful sessions?

Agents often need multi-turn context (e.g., "fix the bug" � "write tests for it"). Stateful sessions make this natural without manual history management.

Want fresh context? Create a new agent instance.

Why return trace + usage?

Observability is critical for agentic systems. You need to know:

• What tools were used
• What the agent was thinking
• How many tokens were consumed
• If any errors occurred

The trace provides full visibility into agent execution.

Comparison with CodexAgent

Feature	CodexAgent	ClaudeCode
SDK	OpenAI Codex SDK	Claude Code Python SDK
Thread management	Built-in thread ID	Session-based (ClaudeSDKClient)
Streaming	Yes	Yes (via receive_response)
Async support	No	Yes (aforward)
Tool types	Codex-specific	Claude Code tools (Bash, Read, Write, etc.)
Sandbox	Simple mode enum	Detailed config dict
Permission control	Sandbox modes	Permission modes + allowed_tools
Configuration	Direct parameters	Config object (ClaudeCodeConfig)

Contributing

Issues and PRs welcome! This is an implementation of Claude Code SDK integration with DSPy.

License

See LICENSE file.

Related Documentation

• Claude Agent SDK - Python Reference - Complete SDK API reference
• Claude Agent SDK - Overview - SDK concepts and guides
• DSPy Documentation - DSPy framework documentation
• Claude Code CLI - Claude Code command-line interface

---

Note: This is a community implementation of Claude Code SDK integration with DSPy, inspired by the CodexAgent design pattern.