ccproxy.api.routes.claude

ccproxy.api.routes.claude

Claude SDK endpoints for CCProxy API Server.

create_openai_chat_completion async

create_openai_chat_completion(
    request,
    openai_request,
    claude_service=Depends(get_claude_service),
)

Create a chat completion using Claude SDK with OpenAI-compatible format.

This endpoint handles OpenAI API format requests and converts them to Anthropic format before using the Claude SDK directly.

Source code in ccproxy/api/routes/claude.py

@router.post("/v1/chat/completions", response_model=None)
async def create_openai_chat_completion(
    request: Request,
    openai_request: OpenAIChatCompletionRequest,
    claude_service: ClaudeSDKService = Depends(get_claude_service),
) -> StreamingResponse | OpenAIChatCompletionResponse:
    """Create a chat completion using Claude SDK with OpenAI-compatible format.

    This endpoint handles OpenAI API format requests and converts them
    to Anthropic format before using the Claude SDK directly.
    """
    try:
        # Create adapter instance
        adapter = OpenAIAdapter()

        # Convert entire OpenAI request to Anthropic format using adapter
        anthropic_request = adapter.adapt_request(openai_request.model_dump())

        # Extract stream parameter
        stream = openai_request.stream or False

        # Call Claude SDK service with adapted request
        if request and hasattr(request, "state") and hasattr(request.state, "context"):
            # Use existing context from middleware
            ctx = request.state.context
            # Add service-specific metadata
            ctx.add_metadata(streaming=stream)

        response = await claude_service.create_completion(
            messages=anthropic_request["messages"],
            model=anthropic_request["model"],
            temperature=anthropic_request.get("temperature"),
            max_tokens=anthropic_request.get("max_tokens"),
            stream=stream,
            user_id=getattr(openai_request, "user", None),
        )

        if stream:
            # Handle streaming response
            async def openai_stream_generator() -> AsyncIterator[bytes]:
                # Use adapt_stream for streaming responses
                async for openai_chunk in adapter.adapt_stream(response):  # type: ignore[arg-type]
                    yield f"data: {json.dumps(openai_chunk)}\n\n".encode()
                # Send final chunk
                yield b"data: [DONE]\n\n"

            return StreamingResponse(
                openai_stream_generator(),
                media_type="text/event-stream",
                headers={
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive",
                },
            )
        else:
            # Convert non-streaming response to OpenAI format using adapter
            openai_response = adapter.adapt_response(response)  # type: ignore[arg-type]
            return OpenAIChatCompletionResponse.model_validate(openai_response)

    except Exception as e:
        # Re-raise specific proxy errors to be handled by the error handler
        from ccproxy.core.errors import ClaudeProxyError

        if isinstance(e, ClaudeProxyError):
            raise
        raise HTTPException(
            status_code=500, detail=f"Internal server error: {str(e)}"
        ) from e

create_anthropic_message async

create_anthropic_message(
    request, claude_service=Depends(get_claude_service)
)

Create a message using Claude SDK with Anthropic format.

This endpoint handles Anthropic API format requests directly using the Claude SDK without any format conversion.

Source code in ccproxy/api/routes/claude.py

@router.post("/v1/messages", response_model=None)
async def create_anthropic_message(
    request: MessageCreateParams,
    claude_service: ClaudeSDKService = Depends(get_claude_service),
) -> StreamingResponse | MessageResponse:
    """Create a message using Claude SDK with Anthropic format.

    This endpoint handles Anthropic API format requests directly
    using the Claude SDK without any format conversion.
    """
    try:
        # Extract parameters from Anthropic request
        messages = [msg.model_dump() for msg in request.messages]
        model = request.model
        temperature = request.temperature
        max_tokens = request.max_tokens
        stream = request.stream or False

        # Call Claude SDK service directly with Anthropic format
        response = await claude_service.create_completion(
            messages=messages,
            model=model,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=stream,
            user_id=getattr(request, "user_id", None),
        )

        if stream:
            # Handle streaming response
            async def anthropic_stream_generator() -> AsyncIterator[bytes]:
                async for chunk in response:  # type: ignore[union-attr]
                    if chunk:
                        yield f"data: {json.dumps(chunk)}\n\n".encode()
                # Send final chunk
                yield b"data: [DONE]\n\n"

            return StreamingResponse(
                anthropic_stream_generator(),
                media_type="text/event-stream",
                headers={
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive",
                },
            )
        else:
            # Return Anthropic format response directly
            return MessageResponse.model_validate(response)

    except Exception as e:
        # Re-raise specific proxy errors to be handled by the error handler
        from ccproxy.core.errors import ClaudeProxyError

        if isinstance(e, ClaudeProxyError):
            raise
        raise HTTPException(
            status_code=500, detail=f"Internal server error: {str(e)}"
        ) from e

list_models async

list_models(claude_service=Depends(get_claude_service))

List available models using Claude SDK service.

Returns a combined list of Anthropic models and recent OpenAI models.

Source code in ccproxy/api/routes/claude.py

@router.get("/v1/models", response_model=None)
async def list_models(
    claude_service: ClaudeSDKService = Depends(get_claude_service),
) -> dict[str, Any]:
    """List available models using Claude SDK service.

    Returns a combined list of Anthropic models and recent OpenAI models.
    """
    try:
        return await claude_service.list_models()
    except Exception as e:
        # Re-raise specific proxy errors to be handled by the error handler
        from ccproxy.core.errors import ClaudeProxyError

        if isinstance(e, ClaudeProxyError):
            raise
        raise HTTPException(
            status_code=500, detail=f"Internal server error: {str(e)}"
        ) from e