ccproxy.services.adapters.base

ccproxy.services.adapters.base

Base adapter for provider plugins.

BaseAdapter

BaseAdapter(config, **kwargs)

Bases: ABC

Base adapter for provider-specific request handling.

Parameters:

Name	Type	Description	Default
config	Any	Plugin configuration	required
**kwargs	Any	Additional keyword arguments for subclasses	{}

Source code in ccproxy/services/adapters/base.py

def __init__(self, config: Any, **kwargs: Any) -> None:
    """Initialize the base adapter.

    Args:
        config: Plugin configuration
        **kwargs: Additional keyword arguments for subclasses
    """
    self.config = config
    self.tool_accumulator_class = kwargs.pop("tool_accumulator_class", None)

handle_request abstractmethod async

handle_request(request)

Handle a provider-specific request.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object with endpoint and method in request.state.context	required

Returns:

Type	Description
Response | StreamingResponse | DeferredStreaming	Response, StreamingResponse, or DeferredStreaming object

Source code in ccproxy/services/adapters/base.py

@abstractmethod
async def handle_request(
    self, request: Request
) -> Response | StreamingResponse | DeferredStreaming:
    """Handle a provider-specific request.

    Args:
        request: FastAPI request object with endpoint and method in request.state.context

    Returns:
        Response, StreamingResponse, or DeferredStreaming object
    """
    ...

handle_streaming abstractmethod async

handle_streaming(request, endpoint, **kwargs)

Handle a streaming request.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object	required
endpoint	str	Target endpoint path	required
**kwargs	Any	Additional provider-specific arguments	{}

Returns:

Type	Description
StreamingResponse | DeferredStreaming	StreamingResponse or DeferredStreaming object

Source code in ccproxy/services/adapters/base.py

@abstractmethod
async def handle_streaming(
    self, request: Request, endpoint: str, **kwargs: Any
) -> StreamingResponse | DeferredStreaming:
    """Handle a streaming request.

    Args:
        request: FastAPI request object
        endpoint: Target endpoint path
        **kwargs: Additional provider-specific arguments

    Returns:
        StreamingResponse or DeferredStreaming object
    """
    ...

validate_request async

validate_request(request, endpoint)

Validate request before processing.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object	required
endpoint	str	Target endpoint path	required

Returns:

Type	Description
dict[str, Any] | None	Validation result or None if valid

Source code in ccproxy/services/adapters/base.py

async def validate_request(
    self, request: Request, endpoint: str
) -> dict[str, Any] | None:
    """Validate request before processing.

    Args:
        request: FastAPI request object
        endpoint: Target endpoint path

    Returns:
        Validation result or None if valid
    """
    return None

transform_request async

transform_request(request_data)

Transform request data if needed.

Parameters:

Name	Type	Description	Default
request_data	dict[str, Any]	Original request data	required

Returns:

Type	Description
dict[str, Any]	Transformed request data

Source code in ccproxy/services/adapters/base.py

async def transform_request(self, request_data: dict[str, Any]) -> dict[str, Any]:
    """Transform request data if needed.

    Args:
        request_data: Original request data

    Returns:
        Transformed request data
    """
    return request_data

transform_response async

transform_response(response_data)

Transform response data if needed.

Parameters:

Name	Type	Description	Default
response_data	dict[str, Any]	Original response data	required

Returns:

Type	Description
dict[str, Any]	Transformed response data

Source code in ccproxy/services/adapters/base.py

async def transform_response(self, response_data: dict[str, Any]) -> dict[str, Any]:
    """Transform response data if needed.

    Args:
        response_data: Original response data

    Returns:
        Transformed response data
    """
    return response_data

cleanup abstractmethod async

cleanup()

Cleanup adapter resources.

This method should be overridden by concrete adapters to clean up any resources like HTTP clients, sessions, or background tasks. Called during application shutdown.

Source code in ccproxy/services/adapters/base.py

@abstractmethod
async def cleanup(self) -> None:
    """Cleanup adapter resources.

    This method should be overridden by concrete adapters to clean up
    any resources like HTTP clients, sessions, or background tasks.
    Called during application shutdown.
    """
    ...