How Claude Agent SDK Integration Works in the Ouroboros Orchestrator

The Ouroboros orchestrator integrates with the Claude Agent SDK through a thin wrapper called ClaudeAgentAdapter, which converts SDK message objects into internal AgentMessage instances while adding retry logic, resume capabilities, and implementing the AgentRuntime protocol required by the runner.

The Q00/ouroboros repository implements this Claude Agent SDK integration as a well-encapsulated abstraction layer. This design keeps the SDK dependency isolated while exposing its full power—including tool use, streaming responses, and session resumption—to the rest of the orchestrator.

Core Architecture of Claude Agent SDK Integration

The ClaudeAgentAdapter Wrapper

The primary integration point resides in src/ouroboros/orchestrator/adapter.py. The ClaudeAgentAdapter class wraps the underlying claude_agent_sdk.query function and implements the AgentRuntime protocol required by the orchestrator's runner.

Key responsibilities of the adapter include:

• Lazy SDK importing: The adapter imports claude_agent_sdk only when execute_task is first called, preventing tight coupling at module load time.
• Options construction: It builds a ClaudeAgentOptions object (lines 58‑64) that configures permission modes, working directories, and resume handles.
• Protocol compliance: By implementing AgentRuntime, the adapter allows OrchestratorRunner to treat Claude as a generic agent backend.

Message Translation and Normalization

The adapter translates SDK-specific message types (AssistantMessage, ResultMessage, etc.) into Ouroboros-native AgentMessage instances via the _convert_message method (lines 506‑600). This normalization layer extracts:

• Text content and thinking blocks
• Tool calls and their arguments
• Session identifiers for resume capability

The _format_tool_detail helper generates human-readable representations of tool invocations for logging and event emission.

Execution Flow: From Seed to Streamed Response

Prompt Construction and Tool Merging

The OrchestratorRunner (in src/ouroboros/orchestrator/runner.py) orchestrates the execution lifecycle. Before invoking the adapter, it performs several preparation steps:

1. System prompt building: Constructs a system prompt via build_system_prompt that defines the agent's role and constraints.
2. Task prompt assembly: Creates the specific user task via build_task_prompt.
3. Tool aggregation: Merges default tools (DEFAULT_TOOLS) with MCP-provided tools through _get_merged_tools.

Streaming and Message Conversion

Once prepared, the runner invokes the adapter's streaming API:

async for message in self._adapter.execute_task(
    prompt=task_prompt,
    tools=merged_tools,
    system_prompt=system_prompt,
):
    # Process each AgentMessage

The adapter internally calls claude_agent_sdk.query(), which yields SDK-specific message objects. Each object passes through _convert_message to become an AgentMessage containing normalized content and metadata.

Session Persistence and Resume Handles

When the SDK emits a session_id (during initialization or result messages), the adapter constructs a RuntimeHandle via _build_runtime_handle. This handle stores:

• Backend identifier: "claude"
• Native session ID: The SDK's session identifier
• Working directory: Execution context
• Permission mode: Security settings (e.g., "acceptEdits")
• Timestamp: For expiration tracking

The runner persists this handle with each message, enabling session resumption. When resuming, OrchestratorRunner.resume_session reconstructs the RuntimeHandle and passes it to the adapter, which supplies it as the resume field in ClaudeAgentOptions.

Error Handling and Resilience

Transient Error Retry Logic

The adapter implements robust retry logic for SDK failures. When execute_task encounters an exception, it checks _is_transient_error against known patterns:

• Rate limit errors (429)
• Server errors (5xx)
• Timeout exceptions
• Connection resets

If transient, the adapter retries up to MAX_RETRIES with exponential backoff starting at RETRY_WAIT_INITIAL and capped at RETRY_WAIT_MAX. If retries exhaust, the adapter emits a final AgentMessage with type="result" and subtype="error" containing the failure details.

Practical Implementation Examples

Direct Adapter Usage

For low-level integration, instantiate ClaudeAgentAdapter directly:

from ouroboros.orchestrator import ClaudeAgentAdapter

adapter = ClaudeAgentAdapter(permission_mode="acceptEdits")

async def run_simple_task():
    async for msg in adapter.execute_task(
        prompt="List the top‑3 Python standard‑library modules for file I/O.",
        tools=["Read", "Write"],          # optional tool whitelist

        system_prompt="You are a concise assistant."
    ):
        print(f"[{msg.type}] {msg.content}")

# In an async context:

# await run_simple_task()

This example streams Claude's replies and prints each AgentMessage with its type and content.

High-Level Runner Integration

For complete orchestration including prompt construction and persistence:

from ouroboros.orchestrator.runner import OrchestratorRunner
from ouroboros.orchestrator.adapter import ClaudeAgentAdapter
from ouroboros.core.seed import Seed
from ouroboros.persistence.event_store import InMemoryEventStore

# Minimal seed definition

seed = Seed(
    goal="Implement a function that returns the factorial of a number.",
    acceptance_criteria=[
        "The function is named `factorial` and accepts a non‑negative int.",
        "It raises ValueError for negative inputs.",
        "It returns correct results for at least 0,1,5,10.",
    ],
    constraints=[],
    metadata=...    # omitted for brevity

)

event_store = InMemoryEventStore()
runner = OrchestratorRunner(
    adapter=ClaudeAgentAdapter(),
    event_store=event_store,
    debug=True,
)

async def run_seed():
    result = await runner.execute_seed(seed)
    if result.is_ok:
        print("✅ Completed:", result.value.final_message)
    else:
        print("❌ Failed:", result.error.message)

# await run_seed()

The runner builds system and task prompts, merges tools, streams messages, persists progress, and returns a rich OrchestratorResult.

Resuming a Paused Session

To resume an interrupted session:

# Assume a previous session ID was stored somewhere

session_id = "session-abc123"

# Re‑create the same runner (same adapter) used before

runner = OrchestratorRunner(adapter=ClaudeAgentAdapter(), event_store=event_store)

async def resume():
    result = await runner.resume_session(session_id, seed)
    if result.is_ok:
        print("Resumed and finished:", result.value.final_message)
    else:
        print("Resume error:", result.error.message)

# await resume()

The runner fetches the last RuntimeHandle from persisted progress, passes it to ClaudeAgentAdapter, and the SDK continues the same Claude session via the resume field in ClaudeAgentOptions.

Key Source Files

File	Primary Concern
src/ouroboros/orchestrator/adapter.py	Implements ClaudeAgentAdapter, message conversion, retry logic, resume handle.
src/ouroboros/orchestrator/runner.py	Orchestrates seed execution, merges tools, handles cancellation, persists progress.
src/ouroboros/providers/claude_code_adapter.py	Alternative provider that also uses the Claude Agent SDK (used by CLI).
tests/unit/orchestrator/test_runner.py	Unit tests demonstrating adapter usage inside the runner.
tests/unit/providers/test_claude_code_adapter.py	Tests for the provider wrapper, ensuring correct option passing.
src/ouroboros/orchestrator/events.py	Event definitions that the runner emits for each streamed message.
src/ouroboros/orchestrator/workflow_state.py	Tracks acceptance-criteria progress; consumes AgentMessage data.

Summary

• ClaudeAgentAdapter in src/ouroboros/orchestrator/adapter.py provides the primary integration layer, wrapping the Claude Agent SDK's query() function and implementing the AgentRuntime protocol.
• Message normalization converts SDK-specific objects (AssistantMessage, ResultMessage) into internal AgentMessage instances via _convert_message, enabling the runner to process responses uniformly.
• Resilience features include transient-error detection with exponential backoff retry logic (MAX_RETRIES, RETRY_WAIT_INITIAL) and session resumption via RuntimeHandle objects that persist native SDK session IDs.
• OrchestratorRunner in src/ouroboros/orchestrator/runner.py handles high-level coordination: prompt construction, tool merging, streaming execution, progress persistence, and cancellation handling.
• Resume capability allows interrupted sessions to continue by reconstructing RuntimeHandle from stored progress and passing it to the adapter, which supplies the resume field in ClaudeAgentOptions.

Frequently Asked Questions

How does the Ouroboros orchestrator handle Claude Agent SDK authentication?

The ClaudeAgentAdapter delegates authentication to the underlying Claude Agent SDK itself. When the adapter lazily imports claude_agent_sdk.query during execute_task, it relies on the SDK's standard credential resolution—typically via environment variables like ANTHROPIC_API_KEY or the SDK's configuration file. The adapter itself does not manage API keys; it simply passes the permission_mode and other runtime options to the SDK's ClaudeAgentOptions object.

What is the difference between ClaudeAgentAdapter and ClaudeCodeAdapter?

ClaudeAgentAdapter (in src/ouroboros/orchestrator/adapter.py) is the core runtime implementation used by the OrchestratorRunner to execute seeds with full streaming, retry, and resume capabilities. ClaudeCodeAdapter (in src/ouroboros/providers/claude_code_adapter.py) is a higher-level provider wrapper used primarily by the CLI and plugin system. While both use the Claude Agent SDK internally, the provider adapter focuses on CLI-specific configuration and option passing, whereas the orchestrator adapter implements the AgentRuntime protocol required for seed-based workflow execution.

How does session resumption work after a crash or restart?

When a session runs, the ClaudeAgentAdapter extracts the native session_id from SDK messages and builds a RuntimeHandle containing the backend type ("claude"), session ID, working directory, and permission mode. The OrchestratorRunner persists this handle alongside each AgentMessage in the SessionRepository. If the process restarts, calling runner.resume_session(session_id) retrieves the last stored RuntimeHandle and passes it to the adapter's execute_task method. The adapter then includes this handle in the resume field of ClaudeAgentOptions, instructing the SDK to reconnect to the existing Claude session rather than starting fresh.

Can I use custom tools with the Claude Agent SDK integration?

Yes. The OrchestratorRunner merges default tools with custom tools before passing them to the adapter. When calling adapter.execute_task, you provide a tools parameter containing tool definitions (as shown in the direct usage example). The adapter passes these through to the SDK's ClaudeAgentOptions. For seed-based execution, the runner's _get_merged_tools method combines DEFAULT_TOOLS with any MCP-provided tools, ensuring the Claude Agent SDK receives the complete toolset required for the task.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}