ccproxy.api

ccproxy.api

API layer for CCProxy API Server.

create_app

create_app(settings=None)

Create and configure the FastAPI application.

Parameters:

Name	Type	Description	Default
settings	Settings | None	Optional settings override. If None, uses get_settings().	None

Returns:

Type	Description
FastAPI	Configured FastAPI application instance.

Source code in ccproxy/api/app.py

def create_app(settings: Settings | None = None) -> FastAPI:
    """Create and configure the FastAPI application.

    Args:
        settings: Optional settings override. If None, uses get_settings().

    Returns:
        Configured FastAPI application instance.
    """
    if settings is None:
        settings = get_settings()
    # Configure logging based on settings BEFORE any module uses logger
    # This is needed for reload mode where the app is re-imported

    import structlog

    # Only configure if not already configured or if no file handler exists
    # okay we have the first debug line but after uvicorn start they are not show root_logger = logging.getLogger()
    # for h in root_logger.handlers:
    #     print(h)
    # has_file_handler = any(
    #     isinstance(h, logging.FileHandler) for h in root_logger.handlers
    # )

    if not structlog.is_configured():
        # Only setup logging if structlog is not configured at all
        # Always use console output, but respect file logging from settings
        json_logs = False
        setup_logging(
            json_logs=json_logs,
            log_level_name=settings.server.log_level,
            log_file=settings.server.log_file,
        )

    app = FastAPI(
        title="CCProxy API Server",
        description="High-performance API server providing Anthropic and OpenAI-compatible interfaces for Claude AI models",
        version=__version__,
        lifespan=lifespan,
    )

    # Setup middleware
    setup_cors_middleware(app, settings)
    setup_error_handlers(app)

    # Add request content logging middleware first (will run fourth due to middleware order)
    app.add_middleware(RequestContentLoggingMiddleware)

    # Add custom access log middleware second (will run third due to middleware order)
    app.add_middleware(AccessLogMiddleware)

    # Add request ID middleware fourth (will run first to initialize context)
    app.add_middleware(RequestIDMiddleware)

    # Add server header middleware (for non-proxy routes)
    # You can customize the server name here
    app.add_middleware(ServerHeaderMiddleware, server_name="uvicorn")

    # Include health router (always enabled)
    app.include_router(health_router, tags=["health"])

    # Include observability routers with granular controls
    if settings.observability.metrics_endpoint_enabled:
        app.include_router(prometheus_router, tags=["metrics"])

    if settings.observability.logs_endpoints_enabled:
        app.include_router(logs_router, prefix="/logs", tags=["logs"])

    if settings.observability.dashboard_enabled:
        app.include_router(dashboard_router, tags=["dashboard"])

    app.include_router(oauth_router, prefix="/oauth", tags=["oauth"])

    # Codex routes for OpenAI integration
    app.include_router(codex_router, tags=["codex"])

    # New /sdk/ routes for Claude SDK endpoints
    app.include_router(claude_router, prefix="/sdk", tags=["claude-sdk"])

    # New /api/ routes for proxy endpoints (includes OpenAI-compatible /v1/chat/completions)
    app.include_router(proxy_router, prefix="/api", tags=["proxy-api"])

    # Shared models endpoints for both SDK and proxy APIs
    app.include_router(models_router, prefix="/sdk", tags=["claude-sdk", "models"])
    app.include_router(models_router, prefix="/api", tags=["proxy-api", "models"])

    # Confirmation endpoints for SSE streaming and responses (conditional on builtin_permissions)
    if settings.claude.builtin_permissions:
        app.include_router(
            permissions_router, prefix="/permissions", tags=["permissions"]
        )
        setup_mcp(app)

    # Mount static files for dashboard SPA
    from pathlib import Path

    # Get the path to the dashboard static files
    current_file = Path(__file__)
    project_root = (
        current_file.parent.parent.parent
    )  # ccproxy/api/app.py -> project root
    dashboard_static_path = project_root / "ccproxy" / "static" / "dashboard"

    # Mount dashboard static files if they exist
    if dashboard_static_path.exists():
        # Mount the _app directory for SvelteKit assets at the correct base path
        app_path = dashboard_static_path / "_app"
        if app_path.exists():
            app.mount(
                "/dashboard/_app",
                StaticFiles(directory=str(app_path)),
                name="dashboard-assets",
            )

        # Mount favicon.svg at root level
        favicon_path = dashboard_static_path / "favicon.svg"
        if favicon_path.exists():
            # For single files, we'll handle this in the dashboard route or add a specific route
            pass

    return app

get_app

get_app()

Get the FastAPI application instance.

Returns:

Type	Description
FastAPI	FastAPI application instance.

Source code in ccproxy/api/app.py

def get_app() -> FastAPI:
    """Get the FastAPI application instance.

    Returns:
        FastAPI application instance.
    """
    return create_app()

get_claude_service

get_claude_service(settings)

Get Claude SDK service instance.

Parameters:

Name	Type	Description	Default
settings	SettingsDep	Application settings dependency	required

Returns:

Type	Description
ClaudeSDKService	Claude SDK service instance

Source code in ccproxy/api/dependencies.py

def get_claude_service(
    settings: SettingsDep,
) -> ClaudeSDKService:
    """Get Claude SDK service instance.

    Args:
        settings: Application settings dependency

    Returns:
        Claude SDK service instance
    """
    logger.debug("Creating Claude SDK service instance")
    # Get global metrics instance
    metrics = get_metrics()

    # Check if pooling should be enabled from configuration
    use_pool = settings.claude.sdk_session_pool.enabled
    session_manager = None

    if use_pool:
        logger.info(
            "claude_sdk_pool_enabled",
            message="Using Claude SDK client pooling for improved performance",
            pool_size=settings.claude.sdk_session_pool.max_sessions,
            max_pool_size=settings.claude.sdk_session_pool.max_sessions,
        )
        # Note: Session manager should be created in the lifespan function, not here
        # This dependency function should not create stateful resources

    return ClaudeSDKService(
        metrics=metrics,
        settings=settings,
        session_manager=session_manager,
    )

get_observability_metrics

get_observability_metrics()

Get observability metrics instance.

Returns:

Type	Description
PrometheusMetrics	PrometheusMetrics instance

Source code in ccproxy/api/dependencies.py

def get_observability_metrics() -> PrometheusMetrics:
    """Get observability metrics instance.

    Returns:
        PrometheusMetrics instance
    """
    logger.debug("get_observability_metrics")
    return get_metrics()

get_proxy_service

get_proxy_service(request, settings, credentials_manager)

Get proxy service instance.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object (for app state access)	required
settings	SettingsDep	Application settings dependency	required
credentials_manager	Annotated[CredentialsManager, Depends(get_credentials_manager)]	Credentials manager dependency	required

Returns:

Type	Description
ProxyService	Proxy service instance

Source code in ccproxy/api/dependencies.py

def get_proxy_service(
    request: Request,
    settings: SettingsDep,
    credentials_manager: Annotated[
        CredentialsManager, Depends(get_credentials_manager)
    ],
) -> ProxyService:
    """Get proxy service instance.

    Args:
        request: FastAPI request object (for app state access)
        settings: Application settings dependency
        credentials_manager: Credentials manager dependency

    Returns:
        Proxy service instance
    """
    logger.debug("get_proxy_service")
    # Create HTTP client for proxy
    from ccproxy.core.http import HTTPXClient

    http_client = HTTPXClient()
    proxy_client = BaseProxyClient(http_client)

    # Get global metrics instance
    metrics = get_metrics()

    return ProxyService(
        proxy_client=proxy_client,
        credentials_manager=credentials_manager,
        settings=settings,
        proxy_mode="full",
        target_base_url=settings.reverse_proxy.target_url,
        metrics=metrics,
        app_state=request.app.state,  # Pass app state for detection data access
    )