Building Your Own Adapter

The Adapter pattern abstracts the LLM behind a stream() interface yielding typed chunks. This lets you swap between API-key auth, Codex OAuth, or even different providers.

Narrow boundary, rich downstream

The temptation when designing an Adapter ABC is to expose every provider feature: chat(), stream(), embed(), transcribe(), generate_image(). Resist it. A narrow ABC with one method (stream(messages, tools) -> AsyncIterator[Event]) keeps every provider on equal footing and forces provider-specific bits to specialize per-implementation.

cwkPippa's backend/adapters/base.py is exactly this shape: one ABC, one streaming method, an Event protocol the consumer parses. The Codex / Gemini / Ollama vessels live under backend/variants/ and specialize that contract; they do not push provider differences upward into routes, store, or frontend. Rule 2 of the architecture: cost is absorbed downstream, not pushed upstream.

Code

Adapter ABC with a single stream() method·python

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import AsyncIterator, Literal, Optional

@dataclass
class TextDelta:
    type: Literal["text_delta"] = "text_delta"
    content: str = ""

@dataclass
class ToolCallDelta:
    type: Literal["tool_call_delta"] = "tool_call_delta"
    call_id: str = ""
    name: str = ""
    arguments_delta: str = ""

@dataclass
class ToolCallComplete:
    type: Literal["tool_call_complete"] = "tool_call_complete"
    call_id: str = ""
    name: str = ""
    arguments: str = ""  # complete JSON string

@dataclass
class StreamDone:
    type: Literal["done"] = "done"
    finish_reason: str = "stop"
    usage: Optional[dict] = None

Chunk = TextDelta | ToolCallDelta | ToolCallComplete | StreamDone

OpenAIAdapter implementation·python

class LLMAdapter(ABC):
    @abstractmethod
    async def stream(self, messages, tools=None, model=None) -> AsyncIterator[Chunk]:
        ...

class OpenAIAdapter(LLMAdapter):
    """Raw HTTP adapter — uses httpx, no SDK dependency."""
    BASE_URL = "https://api.openai.com/v1/chat/completions"

    def __init__(self, api_key=None, model="gpt-4.1"):
        self.api_key = api_key or os.environ["OPENAI_API_KEY"]
        self.default_model = model
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(connect=10, read=120, write=10, pool=5),
            headers={"Authorization": f"Bearer {self.api_key}",
                     "Content-Type": "application/json"},
        )

    async def stream(self, messages, tools=None, model=None) -> AsyncIterator[Chunk]:
        body = {"model": model or self.default_model, "messages": messages, "stream": True}
        if tools: body["tools"] = tools
        # ... parse SSE, yield TextDelta / ToolCallDelta / StreamDone

class CodexAdapter(LLMAdapter):
    """Uses ~/.codex/auth.json OAuth tokens."""
    def __init__(self, model="gpt-4o"):
        self.default_model = model
    async def stream(self, messages, tools=None, model=None) -> AsyncIterator[Chunk]:
        token = get_bearer_token()  # from auth.json
        adapter = OpenAIAdapter(api_key=token, model=model or self.default_model)
        async for chunk in adapter.stream(messages, tools):
            yield chunk

Narrow boundary, rich downstream

Code

External links

Exercise

Progress

Comments 0