ccproxy.claude_sdk.manager

ccproxy.claude_sdk.manager

Claude SDK Session Manager - Pure dependency injection architecture.

This module provides a SessionManager class that encapsulates session pool lifecycle management using dependency injection patterns without any global state.

SessionManager

SessionManager(settings, metrics_factory=None)

Manages the lifecycle of session-based Claude SDK clients with dependency injection.

Parameters:

Name	Type	Description	Default
settings	Settings	Optional settings containing session pool configuration	required
metrics_factory	MetricsFactory | None	Optional callable that returns a metrics instance. If None, no metrics will be used.	None

Source code in ccproxy/claude_sdk/manager.py

def __init__(
    self,
    settings: Settings,
    metrics_factory: MetricsFactory | None = None,
) -> None:
    """Initialize SessionManager with optional settings and metrics factory.

    Args:
        settings: Optional settings containing session pool configuration
        metrics_factory: Optional callable that returns a metrics instance.
                       If None, no metrics will be used.
    """
    import structlog

    logger = structlog.get_logger(__name__)

    self._settings = settings
    self._session_pool: SessionPool | None = None
    self._lock = asyncio.Lock()
    self._metrics_factory = metrics_factory

    # Initialize session pool if enabled
    session_pool_enabled = self._should_enable_session_pool()
    logger.debug(
        "session_manager_init",
        has_settings=bool(settings),
        has_metrics_factory=bool(metrics_factory),
        session_pool_enabled=session_pool_enabled,
    )

    if session_pool_enabled:
        self._session_pool = SessionPool(settings.claude.sdk_session_pool)
        logger.info(
            "session_manager_session_pool_initialized",
            session_ttl=self._session_pool.config.session_ttl,
            max_sessions=self._session_pool.config.max_sessions,
            cleanup_interval=self._session_pool.config.cleanup_interval,
        )
    else:
        logger.debug(
            "session_manager_session_pool_skipped",
            reason="session_pool_disabled_in_settings",
        )

is_active property

is_active

Check if the session manager has an active session pool.

start async

start()

Start the session manager and session pool.

Source code in ccproxy/claude_sdk/manager.py

async def start(self) -> None:
    """Start the session manager and session pool."""
    if self._session_pool:
        await self._session_pool.start()

shutdown async

shutdown()

Gracefully shuts down the session pool.

This method is idempotent - calling it multiple times is safe.

Source code in ccproxy/claude_sdk/manager.py

async def shutdown(self) -> None:
    """Gracefully shuts down the session pool.

    This method is idempotent - calling it multiple times is safe.
    """
    async with self._lock:
        # Close session pool
        if self._session_pool:
            await self._session_pool.stop()
            self._session_pool = None

get_session_client async

get_session_client(session_id, options)

Get session-aware client.

Source code in ccproxy/claude_sdk/manager.py

async def get_session_client(
    self,
    session_id: str,
    options: ClaudeCodeOptions,
) -> SessionClient:
    """Get session-aware client."""

    logger = structlog.get_logger(__name__)
    logger.debug(
        "session_manager_get_session_client",
        session_id=session_id,
        has_session_pool=bool(self._session_pool),
    )

    if not self._session_pool:
        logger.error(
            "session_manager_session_pool_unavailable",
            session_id=session_id,
        )
        raise ClaudeProxyError(
            message="Session pool not available",
            error_type="configuration_error",
            status_code=500,
        )

    return await self._session_pool.get_session_client(session_id, options)

interrupt_session async

interrupt_session(session_id)

Interrupt a specific session due to client disconnection.

Parameters:

Name	Type	Description	Default
session_id	str	The session ID to interrupt	required

Returns:

Type	Description
bool	True if session was found and interrupted, False otherwise

Source code in ccproxy/claude_sdk/manager.py

async def interrupt_session(self, session_id: str) -> bool:
    """Interrupt a specific session due to client disconnection.

    Args:
        session_id: The session ID to interrupt

    Returns:
        True if session was found and interrupted, False otherwise
    """
    if not self._session_pool:
        logger.warning(
            "session_manager_interrupt_session_no_pool",
            session_id=session_id,
        )
        return False

    logger.info(
        "session_manager_interrupt_session",
        session_id=session_id,
    )

    return await self._session_pool.interrupt_session(session_id)

interrupt_all_sessions async

interrupt_all_sessions()

Interrupt all active sessions (for shutdown or emergency cleanup).

Returns:

Type	Description
int	Number of sessions that were interrupted

Source code in ccproxy/claude_sdk/manager.py

async def interrupt_all_sessions(self) -> int:
    """Interrupt all active sessions (for shutdown or emergency cleanup).

    Returns:
        Number of sessions that were interrupted
    """
    if not self._session_pool:
        logger.warning("session_manager_interrupt_all_no_pool")
        return 0

    logger.info("session_manager_interrupt_all_sessions")
    return await self._session_pool.interrupt_all_sessions()

get_session_pool_stats async

get_session_pool_stats()

Get session pool statistics.

Source code in ccproxy/claude_sdk/manager.py

async def get_session_pool_stats(self) -> dict[str, Any]:
    """Get session pool statistics."""
    if not self._session_pool:
        return {"enabled": False}
    return await self._session_pool.get_stats()

reset_for_testing

reset_for_testing()

Synchronous reset for test environments.

Warning

This method should only be used in tests. It does not properly shut down the session pool - use shutdown() for production code.

Source code in ccproxy/claude_sdk/manager.py

def reset_for_testing(self) -> None:
    """Synchronous reset for test environments.

    Warning:
        This method should only be used in tests. It does not properly
        shut down the session pool - use shutdown() for production code.
    """
    self._session_pool = None

has_session_pool async

has_session_pool()

Check if session pool is available and enabled.

Source code in ccproxy/claude_sdk/manager.py

async def has_session_pool(self) -> bool:
    """Check if session pool is available and enabled."""
    return self._session_pool is not None and self._session_pool.config.enabled

has_session async

has_session(session_id)

Check if a session exists in the session pool.

Parameters:

Name	Type	Description	Default
session_id	str	The session ID to check	required

Returns:

Type	Description
bool	True if session exists, False otherwise

Source code in ccproxy/claude_sdk/manager.py

async def has_session(self, session_id: str) -> bool:
    """Check if a session exists in the session pool.

    Args:
        session_id: The session ID to check

    Returns:
        True if session exists, False otherwise
    """
    if not self._session_pool:
        return False
    return await self._session_pool.has_session(session_id)