ccproxy.services.claude_sdk_service

ccproxy.services.claude_sdk_service

Claude SDK service orchestration for business logic.

ClaudeSDKService

ClaudeSDKService(
    sdk_client=None,
    auth_manager=None,
    metrics=None,
    settings=None,
)

Service layer for Claude SDK operations orchestration.

This class handles business logic coordination between the pure SDK client, authentication, metrics, and format conversion while maintaining clean separation of concerns.

Parameters:

Name	Type	Description	Default
sdk_client	ClaudeSDKClient | None	Claude SDK client instance	None
auth_manager	AuthManager | None	Authentication manager (optional)	None
metrics	PrometheusMetrics | None	Prometheus metrics instance (optional)	None
settings	Settings | None	Application settings (optional)	None

Source code in ccproxy/services/claude_sdk_service.py

def __init__(
    self,
    sdk_client: ClaudeSDKClient | None = None,
    auth_manager: AuthManager | None = None,
    metrics: PrometheusMetrics | None = None,
    settings: Settings | None = None,
) -> None:
    """
    Initialize Claude SDK service.

    Args:
        sdk_client: Claude SDK client instance
        auth_manager: Authentication manager (optional)
        metrics: Prometheus metrics instance (optional)
        settings: Application settings (optional)
    """
    self.sdk_client = sdk_client or ClaudeSDKClient()
    self.auth_manager = auth_manager
    self.metrics = metrics
    self.settings = settings
    self.message_converter = MessageConverter()
    self.options_handler = OptionsHandler(settings=settings)

create_completion async

create_completion(
    messages,
    model,
    temperature=None,
    max_tokens=None,
    stream=False,
    user_id=None,
    **kwargs,
)

Create a completion using Claude SDK with business logic orchestration.

Parameters:

Name	Type	Description	Default
messages	list[dict[str, Any]]	List of messages in Anthropic format	required
model	str	The model to use	required
temperature	float | None	Temperature for response generation	None
max_tokens	int | None	Maximum tokens in response	None
stream	bool	Whether to stream responses	False
user_id	str | None	User identifier for auth/metrics	None
**kwargs	Any	Additional arguments	{}

Returns:

Type	Description
dict[str, Any] | AsyncIterator[dict[str, Any]]	Response dict or async iterator of response chunks if streaming

Raises:

Type	Description
ClaudeProxyError	If request fails
ServiceUnavailableError	If service is unavailable

Source code in ccproxy/services/claude_sdk_service.py

async def create_completion(
    self,
    messages: list[dict[str, Any]],
    model: str,
    temperature: float | None = None,
    max_tokens: int | None = None,
    stream: bool = False,
    user_id: str | None = None,
    **kwargs: Any,
) -> dict[str, Any] | AsyncIterator[dict[str, Any]]:
    """
    Create a completion using Claude SDK with business logic orchestration.

    Args:
        messages: List of messages in Anthropic format
        model: The model to use
        temperature: Temperature for response generation
        max_tokens: Maximum tokens in response
        stream: Whether to stream responses
        user_id: User identifier for auth/metrics
        **kwargs: Additional arguments

    Returns:
        Response dict or async iterator of response chunks if streaming

    Raises:
        ClaudeProxyError: If request fails
        ServiceUnavailableError: If service is unavailable
    """
    # Validate authentication if auth manager is configured
    if self.auth_manager and user_id:
        try:
            await self._validate_user_auth(user_id)
        except Exception as e:
            logger.error(
                "authentication_failed",
                user_id=user_id,
                error=str(e),
                error_type=type(e).__name__,
                exc_info=True,
            )
            raise

    # Extract system message and create options
    system_message = self.options_handler.extract_system_message(messages)

    # Map model to Claude model
    model = adapter.map_openai_model_to_claude(model)

    options = self.options_handler.create_options(
        model=model,
        temperature=temperature,
        max_tokens=max_tokens,
        system_message=system_message,
        **kwargs,
    )

    # Convert messages to prompt format
    prompt = self.message_converter.format_messages_to_prompt(messages)

    # Generate request ID for correlation
    from uuid import uuid4

    request_id = str(uuid4())

    # Use request context for observability
    endpoint = "messages"  # Claude SDK uses messages endpoint
    async with request_context(
        method="POST",
        path=f"/sdk/v1/{endpoint}",
        endpoint=endpoint,
        model=model,
        streaming=stream,
        service_type="claude_sdk_service",
        metrics=self.metrics,  # Pass metrics for active request tracking
    ) as ctx:
        try:
            if stream:
                # For streaming, return the async iterator directly
                # Pass context to streaming method
                return self._stream_completion(
                    prompt, options, model, request_id, ctx
                )
            else:
                result = await self._complete_non_streaming(
                    prompt, options, model, request_id, ctx
                )
                return result

        except Exception as e:
            # Log error via access logger (includes metrics)
            await log_request_access(
                context=ctx,
                method="POST",
                error_message=str(e),
                metrics=self.metrics,
                error_type=type(e).__name__,
            )
            raise

list_models async

list_models()

List available Claude models and recent OpenAI models.

Returns:

Type	Description
dict[str, Any]	Dictionary with combined list of models in mixed format

Source code in ccproxy/services/claude_sdk_service.py

async def list_models(self) -> dict[str, Any]:
    """
    List available Claude models and recent OpenAI models.

    Returns:
        Dictionary with combined list of models in mixed format
    """
    # Get Claude models
    supported_models = self.options_handler.get_supported_models()

    # Create Anthropic-style model entries
    anthropic_models = []
    for model_id in supported_models:
        anthropic_models.append(
            {
                "type": "model",
                "id": model_id,
                "display_name": self._get_display_name(model_id),
                "created_at": self._get_created_timestamp(model_id),
            }
        )

    # Add recent OpenAI models (GPT-4 variants and O1 models)
    openai_models = [
        {
            "id": "gpt-4o",
            "object": "model",
            "created": 1715367049,
            "owned_by": "openai",
        },
        {
            "id": "gpt-4o-mini",
            "object": "model",
            "created": 1721172741,
            "owned_by": "openai",
        },
        {
            "id": "gpt-4-turbo",
            "object": "model",
            "created": 1712361441,
            "owned_by": "openai",
        },
        {
            "id": "gpt-4-turbo-preview",
            "object": "model",
            "created": 1706037777,
            "owned_by": "openai",
        },
        {
            "id": "o1",
            "object": "model",
            "created": 1734375816,
            "owned_by": "openai",
        },
        {
            "id": "o1-mini",
            "object": "model",
            "created": 1725649008,
            "owned_by": "openai",
        },
        {
            "id": "o1-preview",
            "object": "model",
            "created": 1725648897,
            "owned_by": "openai",
        },
        {
            "id": "o3",
            "object": "model",
            "created": 1744225308,
            "owned_by": "openai",
        },
        {
            "id": "o3-mini",
            "object": "model",
            "created": 1737146383,
            "owned_by": "openai",
        },
    ]

    # Return combined response in mixed format
    return {
        "data": anthropic_models + openai_models,
        "has_more": False,
        "object": "list",
    }

validate_health async

validate_health()

Validate that the service is healthy.

Returns:

Type	Description
bool	True if healthy, False otherwise

Source code in ccproxy/services/claude_sdk_service.py

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

    Returns:
        True if healthy, False otherwise
    """
    try:
        return await self.sdk_client.validate_health()
    except Exception as e:
        logger.error(
            "health_check_failed",
            error=str(e),
            error_type=type(e).__name__,
            exc_info=True,
        )
        return False

close async

close()

Close the service and cleanup resources.

Source code in ccproxy/services/claude_sdk_service.py

async def close(self) -> None:
    """Close the service and cleanup resources."""
    await self.sdk_client.close()