Persistent Subprocess Pools (Agent SDK)

Why a pool

Spinning up a Claude Agent SDK subprocess per request is expensive (startup cost, MCP attach, system prompt warm-up). For chat-style workloads, a pool of long-lived clients keyed by conversation_id is dramatically faster. cwkPippa's ClaudeSessionManager is exactly this pool.

What the pool owns

The pool owns: lifecycle (connect on first use, disconnect on TTL or shutdown), error handling (replace dead clients), capacity caps (max concurrent subprocesses), and metrics (active count, lifetime, last-use timestamp). Treat it like a database connection pool — the same disciplines apply.

Stale subprocess hygiene

Long-lived subprocesses leak resources if you forget about them. Sweep the pool periodically: drop clients with no activity in N minutes, free their MCP connections, log the reason. Bound the pool size; refuse new sessions when full and surface the failure rather than queuing forever.

Principle: A pool is its own component. Build it like infrastructure, not like glue code.

Code

Minimal session pool·python

import asyncio, time
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

class SessionManager:
    def __init__(self, max_size: int = 50, ttl_seconds: int = 1800):
        self._clients: dict[str, tuple[ClaudeSDKClient, float]] = {}
        self._lock = asyncio.Lock()
        self.max_size = max_size
        self.ttl = ttl_seconds

    async def get(self, conv_id: str, options: ClaudeAgentOptions) -> ClaudeSDKClient:
        async with self._lock:
            await self._sweep()
            if conv_id in self._clients:
                client, _ = self._clients[conv_id]
                self._clients[conv_id] = (client, time.time())
                return client
            if len(self._clients) >= self.max_size:
                raise RuntimeError("session pool full")
            client = ClaudeSDKClient(options=options)
            await client.connect()
            self._clients[conv_id] = (client, time.time())
            return client

    async def _sweep(self):
        now = time.time()
        stale = [cid for cid, (_, last) in self._clients.items() if now - last > self.ttl]
        for cid in stale:
            client, _ = self._clients.pop(cid)
            await client.disconnect()

Persistent Subprocess Pools (Agent SDK)

Why a pool

What the pool owns

Stale subprocess hygiene

Code

External links

Exercise

Progress

Comments 0