ccproxy.models.claude_sdk

ccproxy.models.claude_sdk

Strongly-typed Pydantic models for Claude SDK types.

This module provides Pydantic models that mirror the Claude SDK types from the official claude-code-sdk-python repository. These models enable strong typing throughout the proxy system and provide runtime validation.

Based on: https://github.com/anthropics/claude-code-sdk-python/blob/main/src/claude_code_sdk/types.py

TextBlock

Bases: BaseModel

Text content block from Claude SDK.

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/models/claude_sdk.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_code_sdk",
    }

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/models/claude_sdk.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_code_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.

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/models/claude_sdk.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

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/models/claude_sdk.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

SystemMessage

Bases: BaseModel

System message from Claude SDK.

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 for backward compatibility.

SDKMessageMode

Bases: SystemMessage

Custom content block for system messages with source attribution.

ToolUseSDKBlock

Bases: BaseModel

Custom content block for tool use with SDK metadata.

ToolResultSDKBlock

Bases: BaseModel

Custom content block for tool results with SDK metadata.

ResultMessageBlock

Bases: ResultMessage

Custom content block for result messages with session data.

SDKMessageContent

Bases: BaseModel

Content structure for SDK query messages.

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.

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/models/claude_sdk.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())

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/models/claude_sdk.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,
    )

convert_sdk_text_block

convert_sdk_text_block(text_content)

Convert raw text content to TextBlock model.

Source code in ccproxy/models/claude_sdk.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_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/models/claude_sdk.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)

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/models/claude_sdk.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_system_message

convert_sdk_system_message(subtype, data)

Convert raw system message data to SystemMessage model.

Source code in ccproxy/models/claude_sdk.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_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/models/claude_sdk.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,
    )