ccproxy.plugins.claude_sdk

ccproxy.plugins.claude_sdk

Claude SDK integration module.

ClaudeSDKClient

ClaudeSDKClient(config, session_manager=None)

Minimal Claude SDK client wrapper that handles core SDK interactions.

This class provides a clean interface to the Claude Code SDK while handling error translation and basic query execution. Supports both stateless query() calls and pooled connection reuse for improved performance.

Parameters:

Name	Type	Description	Default
config	ClaudeSDKSettings	Plugin-specific configuration for Claude SDK	required
session_manager	SessionManager | None	Optional SessionManager instance for dependency injection	None

Source code in ccproxy/plugins/claude_sdk/client.py

def __init__(
    self,
    config: ClaudeSDKSettings,
    session_manager: SessionManager | None = None,
) -> None:
    """Initialize the Claude SDK client.

    Args:
        config: Plugin-specific configuration for Claude SDK
        session_manager: Optional SessionManager instance for dependency injection
    """
    self._last_api_call_time_ms: float = 0.0
    self.config = config
    self._session_manager = session_manager

query_completion async

query_completion(
    message, options, request_id=None, session_id=None
)

Execute a query using the Claude Code SDK and return a StreamHandle.

Parameters:

Name	Type	Description	Default
message	SDKMessage	SDKMessage to send to Claude SDK	required
options	ClaudeAgentOptions	Claude Code options configuration	required
request_id	str | None	Optional request ID for correlation	None
session_id	str | None	Optional session ID for conversation continuity	None

Returns:

Type	Description
StreamHandle	StreamHandle that can create listeners for the stream

Raises:

Type	Description
ClaudeSDKError	If the query fails

Source code in ccproxy/plugins/claude_sdk/client.py

async def query_completion(
    self,
    message: SDKMessage,
    options: ClaudeAgentOptions,
    request_id: str | None = None,
    session_id: str | None = None,
) -> StreamHandle:
    """
    Execute a query using the Claude Code SDK and return a StreamHandle.

    Args:
        message: SDKMessage to send to Claude SDK
        options: Claude Code options configuration
        request_id: Optional request ID for correlation
        session_id: Optional session ID for conversation continuity

    Returns:
        StreamHandle that can create listeners for the stream

    Raises:
        ClaudeSDKError: If the query fails
    """
    # Determine routing strategy
    if self._should_use_session_pool(session_id):
        return await self._create_session_pool_stream_handle(
            message, options, request_id, session_id
        )
    else:
        return await self._create_direct_stream_handle(
            message, options, request_id, session_id
        )

validate_health async

validate_health()

Validate that the Claude SDK is healthy.

Returns:

Type	Description
bool	True if healthy, False otherwise

Source code in ccproxy/plugins/claude_sdk/client.py

async def validate_health(self) -> bool:
    """
    Validate that the Claude SDK is healthy.

    Returns:
        True if healthy, False otherwise
    """
    try:
        logger.debug("health_check_start", component="claude_sdk")

        # Simple health check - the SDK is available if we can import it
        # More sophisticated checks could be added here
        is_healthy = True

        logger.debug(
            "health_check_completed", component="claude_sdk", healthy=is_healthy
        )
        return is_healthy
    except Exception as e:
        logger.error(
            "health_check_failed",
            component="claude_sdk",
            error=str(e),
            error_type=type(e).__name__,
            exc_info=e,
        )
        return False

interrupt_session async

interrupt_session(session_id)

Interrupt a specific session due to client disconnection.

Parameters:

Name	Type	Description	Default
session_id	str	The session ID to interrupt	required

Returns:

Type	Description
bool	True if session was found and interrupted, False otherwise

Source code in ccproxy/plugins/claude_sdk/client.py

async def interrupt_session(self, session_id: str) -> bool:
    """Interrupt a specific session due to client disconnection.

    Args:
        session_id: The session ID to interrupt

    Returns:
        True if session was found and interrupted, False otherwise
    """
    logger.debug("sdk_client_interrupt_session_started", session_id=session_id)
    if self._session_manager:
        logger.debug(
            "client_interrupt_session_requested",
            session_id=session_id,
            has_session_manager=True,
        )
        return await self._session_manager.interrupt_session(session_id)
    else:
        logger.warning(
            "client_interrupt_session_no_session_manager",
            session_id=session_id,
        )
        return False

close async

close()

Close the client and cleanup resources.

Source code in ccproxy/plugins/claude_sdk/client.py

async def close(self) -> None:
    """Close the client and cleanup resources."""
    # Claude Code SDK doesn't require explicit cleanup
    pass

ClaudeSDKError

Bases: Exception

Base Claude SDK error.

StreamTimeoutError

StreamTimeoutError(message, session_id, timeout_seconds)

Bases: ClaudeSDKError

Stream timeout error when no SDK message is received within timeout.

Source code in ccproxy/plugins/claude_sdk/exceptions.py

def __init__(self, message: str, session_id: str, timeout_seconds: float):
    super().__init__(message)
    self.session_id = session_id
    self.timeout_seconds = timeout_seconds

AssistantMessage

Bases: BaseModel

Assistant message from Claude SDK.

convert_content_blocks classmethod

convert_content_blocks(v)

Convert Claude SDK dataclass blocks to Pydantic models.

Source code in ccproxy/plugins/claude_sdk/models.py

@field_validator("content", mode="before")
@classmethod
def convert_content_blocks(cls, v: Any) -> list[Any]:
    """Convert Claude SDK dataclass blocks to Pydantic models."""
    if not isinstance(v, list):
        return []

    converted_blocks = []
    for block in v:
        if isinstance(block, SDKTextBlock | SDKToolUseBlock | SDKToolResultBlock):
            # Convert Claude SDK dataclass to dict and add type field
            if isinstance(block, SDKTextBlock):
                converted_blocks.append({"type": "text", "text": block.text})
            elif isinstance(block, SDKToolUseBlock):
                converted_blocks.append(
                    cast(
                        Any,
                        {
                            "type": "tool_use",
                            "id": str(block.id),
                            "name": str(block.name),
                            "input": dict(block.input),
                        },
                    )
                )
            elif isinstance(block, SDKToolResultBlock):
                converted_blocks.append(
                    cast(
                        Any,
                        {
                            "type": "tool_result",
                            "tool_use_id": str(block.tool_use_id),
                            "content": block.content,
                            "is_error": block.is_error,
                        },
                    )
                )
        else:
            converted_blocks.append(block)

    return converted_blocks

ResultMessage

Bases: BaseModel

Result message from Claude SDK.

stop_reason property

stop_reason

Get stop reason from result or default to end_turn.

usage_model property

usage_model

Get usage information as a Usage model.

ResultMessageBlock

Bases: ResultMessage

Custom content block for result messages with session data.

SDKMessage

Bases: BaseModel

Message format used to send queries over the Claude SDK.

This represents the internal message structure expected by the Claude Code SDK client for query operations.

SDKMessageContent

Bases: BaseModel

Content structure for SDK query messages.

SDKMessageMode

Bases: SystemMessage

Custom content block for system messages with source attribution.

SystemMessage

Bases: BaseModel

System message from Claude SDK.

TextBlock

Bases: BaseModel

Text content block from Claude SDK.

ThinkingBlock

Bases: BaseModel

Thinking content block from Claude SDK.

Note: Thinking blocks are not normally sent by Claude Code SDK, but this model is included for defensive programming to handle any future SDK changes or edge cases where thinking content might be included in SDK responses.

ToolResultBlock

Bases: BaseModel

Tool result content block from Claude SDK.

to_sdk_block

to_sdk_block()

Convert to ToolResultSDKBlock format for streaming.

Source code in ccproxy/plugins/claude_sdk/models.py

def to_sdk_block(self) -> dict[str, Any]:
    """Convert to ToolResultSDKBlock format for streaming."""
    return {
        "type": "tool_result_sdk",
        "tool_use_id": self.tool_use_id,
        "content": self.content,
        "is_error": self.is_error,
        "source": "claude_agent_sdk",
    }

ToolResultSDKBlock

Bases: BaseModel

Custom content block for tool results with SDK metadata.

ToolUseBlock

Bases: BaseModel

Tool use content block from Claude SDK.

to_sdk_block

to_sdk_block()

Convert to ToolUseSDKBlock format for streaming.

Source code in ccproxy/plugins/claude_sdk/models.py

def to_sdk_block(self) -> dict[str, Any]:
    """Convert to ToolUseSDKBlock format for streaming."""
    return {
        "type": "tool_use_sdk",
        "id": self.id,
        "name": self.name,
        "input": self.input,
        "source": "claude_agent_sdk",
    }

ToolUseSDKBlock

Bases: BaseModel

Custom content block for tool use with SDK metadata.

UserMessage

Bases: BaseModel

User message from Claude SDK.

convert_content_blocks classmethod

convert_content_blocks(v)

Convert Claude SDK dataclass blocks to Pydantic models.

Source code in ccproxy/plugins/claude_sdk/models.py

@field_validator("content", mode="before")
@classmethod
def convert_content_blocks(cls, v: Any) -> list[Any]:
    """Convert Claude SDK dataclass blocks to Pydantic models."""
    if not isinstance(v, list):
        return []

    converted_blocks = []
    for block in v:
        if isinstance(block, SDKTextBlock | SDKToolUseBlock | SDKToolResultBlock):
            # Convert Claude SDK dataclass to dict and add type field
            if isinstance(block, SDKTextBlock):
                converted_blocks.append({"type": "text", "text": block.text})
            elif isinstance(block, SDKToolUseBlock):
                converted_blocks.append(
                    cast(
                        Any,
                        {
                            "type": "tool_use",
                            "id": str(block.id),
                            "name": str(block.name),
                            "input": dict(block.input),
                        },
                    )
                )
            elif isinstance(block, SDKToolResultBlock):
                converted_blocks.append(
                    cast(
                        Any,
                        {
                            "type": "tool_result",
                            "tool_use_id": str(block.tool_use_id),
                            "content": block.content,
                            "is_error": block.is_error,
                        },
                    )
                )
        else:
            converted_blocks.append(block)

    return converted_blocks

OptionsHandler

OptionsHandler(config)

Handles creation and management of Claude SDK options.

Parameters:

Name	Type	Description	Default
config	ClaudeSDKSettings	Plugin-specific configuration for Claude SDK	required

Source code in ccproxy/plugins/claude_sdk/options.py

def __init__(self, config: ClaudeSDKSettings) -> None:
    """
    Initialize options handler.

    Args:
        config: Plugin-specific configuration for Claude SDK
    """
    self.config = config

create_options

create_options(
    model,
    temperature=None,
    max_tokens=None,
    system_message=None,
    **additional_options,
)

Create Claude SDK options from API parameters.

Parameters:

Name	Type	Description	Default
model	str	The model name	required
temperature	float | None	Temperature for response generation	None
max_tokens	int | None	Maximum tokens in response	None
system_message	str | None	System message to include	None
**additional_options	Any	Additional options to set on the ClaudeAgentOptions instance	{}

Returns:

Type	Description
ClaudeAgentOptions	Configured ClaudeAgentOptions instance

Source code in ccproxy/plugins/claude_sdk/options.py

def create_options(
    self,
    model: str,
    temperature: float | None = None,
    max_tokens: int | None = None,
    system_message: str | None = None,
    **additional_options: Any,
) -> ClaudeAgentOptions:
    """
    Create Claude SDK options from API parameters.

    Args:
        model: The model name
        temperature: Temperature for response generation
        max_tokens: Maximum tokens in response
        system_message: System message to include
        **additional_options: Additional options to set on the ClaudeAgentOptions instance

    Returns:
        Configured ClaudeAgentOptions instance
    """
    # Start with configured defaults if available, otherwise create fresh instance
    if self.config and self.config.code_options:
        configured_opts = self.config.code_options
        options = ClaudeAgentOptions()

        # Copy all attributes from configured defaults
        for attr in dir(configured_opts):
            if not attr.startswith("_"):
                configured_value = getattr(configured_opts, attr)
                if configured_value is not None and hasattr(options, attr):
                    # Special handling for mcp_servers to ensure we copy the dict
                    if attr == "mcp_servers" and isinstance(configured_value, dict):
                        setattr(options, attr, configured_value.copy())
                    else:
                        setattr(options, attr, configured_value)
    else:
        options = ClaudeAgentOptions()

    # Override the model (API parameter takes precedence)
    options.model = model

    # Apply system message if provided (this is supported by ClaudeAgentOptions)
    if system_message is not None:
        options.system_prompt = system_message

    # If session_id is provided via additional_options, enable continue_conversation
    if additional_options.get("session_id"):
        options.continue_conversation = True

    # Automatically map additional_options to ClaudeAgentOptions attributes
    for key, value in additional_options.items():
        if hasattr(options, key):
            try:
                # Attempt type conversion if the attribute already exists
                attr_type = type(getattr(options, key))
                # Only convert if the attribute is not None
                if getattr(options, key) is not None:
                    setattr(options, key, attr_type(value))
                else:
                    setattr(options, key, value)
            except Exception:
                # Fallback to direct assignment if conversion fails
                setattr(options, key, value)

    return options

extract_system_message staticmethod

extract_system_message(messages)

Extract system message from Anthropic messages format.

Parameters:

Name	Type	Description	Default
messages	list[dict[str, Any]]	List of messages in Anthropic format	required

Returns:

Type	Description
str | None	System message content if found, None otherwise

Source code in ccproxy/plugins/claude_sdk/options.py

@staticmethod
def extract_system_message(messages: list[dict[str, Any]]) -> str | None:
    """
    Extract system message from Anthropic messages format.

    Args:
        messages: List of messages in Anthropic format

    Returns:
        System message content if found, None otherwise
    """
    for message in messages:
        if message.get("role") == "system":
            content = message.get("content", "")
            if isinstance(content, list):
                # Handle content blocks
                text_parts = []
                for block in content:
                    if block.get("type") == "text":
                        text_parts.append(block.get("text", ""))
                return " ".join(text_parts)
            return str(content)
    return None

get_supported_models staticmethod

get_supported_models()

Get list of supported Claude models.

Returns:

Type	Description
list[str]	List of supported model names

Source code in ccproxy/plugins/claude_sdk/options.py

@staticmethod
def get_supported_models() -> list[str]:
    """
    Get list of supported Claude models.

    Returns:
        List of supported model names
    """
    from ccproxy.plugins.claude_shared.model_defaults import (
        DEFAULT_CLAUDE_MODEL_CARDS,
    )

    return [card.id for card in DEFAULT_CLAUDE_MODEL_CARDS]

validate_model staticmethod

validate_model(model)

Validate if a model is supported.

Parameters:

Name	Type	Description	Default
model	str	The model name to validate	required

Returns:

Type	Description
bool	True if supported, False otherwise

Source code in ccproxy/plugins/claude_sdk/options.py

@staticmethod
def validate_model(model: str) -> bool:
    """
    Validate if a model is supported.

    Args:
        model: The model name to validate

    Returns:
        True if supported, False otherwise
    """
    return model in OptionsHandler.get_supported_models()

get_default_options staticmethod

get_default_options()

Get default options for API parameters.

Returns:

Type	Description
dict[str, Any]	Dictionary of default API parameter values

Source code in ccproxy/plugins/claude_sdk/options.py

@staticmethod
def get_default_options() -> dict[str, Any]:
    """
    Get default options for API parameters.

    Returns:
        Dictionary of default API parameter values
    """
    return {
        "model": "claude-3-5-sonnet-20241022",
        "temperature": 0.7,
        "max_tokens": 4000,
    }

convert_sdk_result_message

convert_sdk_result_message(
    session_id,
    subtype="",
    duration_ms=0,
    duration_api_ms=0,
    is_error=False,
    num_turns=0,
    usage=None,
    total_cost_usd=None,
    result=None,
)

Convert raw result message data to ResultMessage model.

Source code in ccproxy/plugins/claude_sdk/models.py

def convert_sdk_result_message(
    session_id: str,
    subtype: str = "",
    duration_ms: int = 0,
    duration_api_ms: int = 0,
    is_error: bool = False,
    num_turns: int = 0,
    usage: dict[str, Any] | None = None,
    total_cost_usd: float | None = None,
    result: str | None = None,
) -> ResultMessage:
    """Convert raw result message data to ResultMessage model."""
    return ResultMessage(
        session_id=session_id,
        subtype=subtype,
        duration_ms=duration_ms,
        duration_api_ms=duration_api_ms,
        is_error=is_error,
        num_turns=num_turns,
        usage=usage,
        total_cost_usd=total_cost_usd,
        result=result,
    )

convert_sdk_system_message

convert_sdk_system_message(subtype, data)

Convert raw system message data to SystemMessage model.

Source code in ccproxy/plugins/claude_sdk/models.py

def convert_sdk_system_message(subtype: str, data: dict[str, Any]) -> SystemMessage:
    """Convert raw system message data to SystemMessage model."""
    return SystemMessage(subtype=subtype, data=data)

convert_sdk_text_block

convert_sdk_text_block(text_content)

Convert raw text content to TextBlock model.

Source code in ccproxy/plugins/claude_sdk/models.py

def convert_sdk_text_block(text_content: str) -> TextBlock:
    """Convert raw text content to TextBlock model."""
    return TextBlock(text=text_content)

convert_sdk_tool_result_block

convert_sdk_tool_result_block(
    tool_use_id, content=None, is_error=None
)

Convert raw tool result data to ToolResultBlock model.

Source code in ccproxy/plugins/claude_sdk/models.py

def convert_sdk_tool_result_block(
    tool_use_id: str,
    content: str | list[dict[str, Any]] | None = None,
    is_error: bool | None = None,
) -> ToolResultBlock:
    """Convert raw tool result data to ToolResultBlock model."""
    return ToolResultBlock(tool_use_id=tool_use_id, content=content, is_error=is_error)

convert_sdk_tool_use_block

convert_sdk_tool_use_block(tool_id, tool_name, tool_input)

Convert raw tool use data to ToolUseBlock model.

Source code in ccproxy/plugins/claude_sdk/models.py

def convert_sdk_tool_use_block(
    tool_id: str, tool_name: str, tool_input: dict[str, Any]
) -> ToolUseBlock:
    """Convert raw tool use data to ToolUseBlock model."""
    return ToolUseBlock(id=tool_id, name=tool_name, input=tool_input)

create_sdk_message

create_sdk_message(
    content, session_id=None, parent_tool_use_id=None
)

Create an SDKMessage instance for sending queries to Claude SDK.

Parameters:

Name	Type	Description	Default
content	str	The text content to send to Claude	required
session_id	str | None	Optional session ID for conversation continuity	None
parent_tool_use_id	str | None	Optional parent tool use ID	None

Returns:

Type	Description
SDKMessage	SDKMessage instance ready to send to Claude SDK

Source code in ccproxy/plugins/claude_sdk/models.py

def create_sdk_message(
    content: str,
    session_id: str | None = None,
    parent_tool_use_id: str | None = None,
) -> SDKMessage:
    """Create an SDKMessage instance for sending queries to Claude SDK.

    Args:
        content: The text content to send to Claude
        session_id: Optional session ID for conversation continuity
        parent_tool_use_id: Optional parent tool use ID

    Returns:
        SDKMessage instance ready to send to Claude SDK
    """
    return SDKMessage(
        message=SDKMessageContent(content=content),
        session_id=session_id,
        parent_tool_use_id=parent_tool_use_id,
    )

to_sdk_variant

to_sdk_variant(base_model, sdk_class)

Convert a base model to its SDK variant using model_validate().

Parameters:

Name	Type	Description	Default
base_model	BaseModel	The base model instance to convert	required
sdk_class	type[T]	The target SDK class to convert to	required

Returns:

Type	Description
T	Instance of the SDK class with data from the base model

Example

text_block = TextBlock(text="message") text_block_sdk = to_sdk_variant(text_block, TextBlockSDK)

Source code in ccproxy/plugins/claude_sdk/models.py

def to_sdk_variant(base_model: BaseModel, sdk_class: type[T]) -> T:
    """Convert a base model to its SDK variant using model_validate().

    Args:
        base_model: The base model instance to convert
        sdk_class: The target SDK class to convert to

    Returns:
        Instance of the SDK class with data from the base model

    Example:
        >>> text_block = TextBlock(text="message")
        >>> text_block_sdk = to_sdk_variant(text_block, TextBlockSDK)
    """
    return sdk_class.model_validate(base_model.model_dump())