Providers

The provider abstraction and the three built-in backends. See the Providers guide and Auto-provider mode for usage.

Abstract interface

Provider

Bases: ABC

ABC that each coding agent backend implements.

name abstractmethod property

name: str

Provider identifier (e.g. 'claude_code', 'codex').

binary_name property

binary_name: str

Name of the backing CLI executable (as looked up on PATH).

Defaults to the provider name; override when the CLI binary is named differently (e.g. claude_code → claude). Used by the default find_binary.

resolve_model

resolve_model(tier: ModelTier) -> str | None

Translate a ModelTier into a concrete model identifier, if needed.

Each provider must override this to map abstract tiers (e.g. ModelTier.STRONGEST) to the actual model string it supports. Return None when the provider's own default/config should select the tier.

Source code in caw/provider.py

def resolve_model(self, tier: ModelTier) -> str | None:
    """Translate a `ModelTier` into a concrete model identifier, if needed.

    Each provider must override this to map abstract tiers (e.g.
    ``ModelTier.STRONGEST``) to the actual model string it supports. Return
    ``None`` when the provider's own default/config should select the tier.
    """
    raise NotImplementedError(
        f"{self.name} provider does not implement resolve_model(); "
        f"pass an explicit model string instead of ModelTier.{tier.name}"
    )

resolve_tool_restrictions

resolve_tool_restrictions(tools: ToolGroup) -> dict[str, Any]

Translate ToolGroup into provider-specific session kwargs.

Receives a concrete ToolGroup value (never None — the Agent layer applies the default before calling this).

Source code in caw/provider.py

def resolve_tool_restrictions(self, tools: ToolGroup) -> dict[str, Any]:
    """Translate ToolGroup into provider-specific session kwargs.

    Receives a concrete ToolGroup value (never None — the Agent layer
    applies the default before calling this).
    """
    return {}

check_limit

check_limit(model: str | None = None) -> int | None

Probe whether the provider's usage limit is currently active.

Sends a minimal test prompt and checks if the response indicates a usage-limit. Returns the estimated number of minutes to wait before the limit resets, or None if no limit is detected.

This incurs a small token cost for the probe request.

Source code in caw/provider.py

def check_limit(self, model: str | None = None) -> int | None:
    """Probe whether the provider's usage limit is currently active.

    Sends a minimal test prompt and checks if the response indicates a
    usage-limit.  Returns the estimated number of minutes to wait before
    the limit resets, or ``None`` if no limit is detected.

    This incurs a small token cost for the probe request.
    """
    from caw.display import get_global_display, set_global_display

    old_display = get_global_display()
    set_global_display(None)
    try:
        session = self.start_session(
            mcp_servers=[],
            model=model,
            system_prompt="Reply with the single word 'ok'.",
            **self._limit_probe_kwargs(),
        )
        try:
            turn = session.send("hi")
            return session.detect_usage_limit(turn)
        finally:
            session.end()
    finally:
        set_global_display(old_display)

find_binary

find_binary() -> str | None

Resolve the CLI executable's path, or None if not installed.

Default looks up binary_name on PATH. Override to add provider-specific fallback locations.

Source code in caw/provider.py

def find_binary(self) -> str | None:
    """Resolve the CLI executable's path, or ``None`` if not installed.

    Default looks up `binary_name` on ``PATH``.  Override to add
    provider-specific fallback locations.
    """
    import shutil

    return shutil.which(self.binary_name)

check_auth

check_auth() -> 'AuthSignal | None'

Best-effort, non-authoritative introspection of credentials.

Returns an AuthSignal, or None when the provider cannot introspect its credentials at all. Override in subclasses; the default returns None (unknown).

Source code in caw/provider.py

def check_auth(self) -> "AuthSignal | None":  # noqa: F821 (forward ref)
    """Best-effort, non-authoritative introspection of credentials.

    Returns an `AuthSignal`, or ``None`` when the
    provider cannot introspect its credentials at all.  Override in
    subclasses; the default returns ``None`` (unknown).
    """
    return None

check_health

check_health(*, live: bool = False, model: str | None = None) -> 'ProviderHealth'

Report raw health signals for this provider.

Fast by default: checks whether the CLI is installed and introspects credentials. When live is True, additionally runs the check_limit probe (one request) to confirm the provider responds and whether it is currently rate-limited.

Forms no verdict on "availability" — see caw.health.ProviderHealth.

Source code in caw/provider.py

def check_health(self, *, live: bool = False, model: str | None = None) -> "ProviderHealth":  # noqa: F821
    """Report raw health signals for this provider.

    Fast by default: checks whether the CLI is installed and introspects
    credentials.  When ``live`` is True, additionally runs the
    `check_limit` probe (one request) to confirm the provider
    responds and whether it is currently rate-limited.

    Forms no verdict on "availability" — see `caw.health.ProviderHealth`.
    """
    from caw.health import ProviderHealth

    binary = self.find_binary()
    health = ProviderHealth(
        provider=self.name,
        installed=binary is not None,
        binary_path=binary,
        auth=self.check_auth(),
    )
    if live and health.installed:
        health.probed = True
        try:
            wait = self.check_limit(model=model)
            health.rate_limited = wait is not None
            health.wait_minutes = wait
        except Exception as exc:  # noqa: BLE001 — surface as a signal, not a raise
            health.error = str(exc)
    return health

start_interactive

start_interactive(initial_prompt: str, mcp_servers: list[MCPServer], capture_bytes: int = 0, **kwargs: Any) -> InteractiveResult

Launch the provider binary interactively with an initial prompt.

Hands control to the user's terminal — stdin/stdout/stderr are inherited so the user interacts with the agent directly. A copy of stdout is captured via a pty.

Returns an InteractiveResult with the exit code and captured output.

Source code in caw/provider.py

def start_interactive(
    self, initial_prompt: str, mcp_servers: list[MCPServer], capture_bytes: int = 0, **kwargs: Any
) -> InteractiveResult:
    """Launch the provider binary interactively with an initial prompt.

    Hands control to the user's terminal — stdin/stdout/stderr are
    inherited so the user interacts with the agent directly.
    A copy of stdout is captured via a pty.

    Returns an `InteractiveResult` with the exit code and
    captured output.
    """
    raise NotImplementedError(f"{self.name} provider does not support interactive mode.")

start_session abstractmethod

start_session(mcp_servers: list[MCPServer], **kwargs: Any) -> ProviderSession

Create and return a new provider session.

Source code in caw/provider.py

@abstractmethod
def start_session(self, mcp_servers: list[MCPServer], **kwargs: Any) -> ProviderSession:
    """Create and return a new provider session."""
    ...

resume_key_from_trajectory

resume_key_from_trajectory(trajectory: Trajectory) -> str | None

Extract the resume key from a persisted trajectory.

Mirrors ProviderSession.resume_key but reads from a stored trajectory rather than a live session. Returns None if the trajectory predates resume support or was never sent to.

Source code in caw/provider.py

def resume_key_from_trajectory(self, trajectory: Trajectory) -> str | None:
    """Extract the resume key from a persisted *trajectory*.

    Mirrors `ProviderSession.resume_key` but reads from a stored
    trajectory rather than a live session.  Returns ``None`` if the
    trajectory predates resume support or was never sent to.
    """
    return None

resume_session

resume_session(mcp_servers: list[MCPServer], *, session_id: str, resume_key: str, trajectory: Trajectory | None = None, **kwargs: Any) -> ProviderSession

Rebuild a live session ready to continue an existing conversation.

resume_key is the provider-side key the backend CLI needs to resume (see ProviderSession.resume_key); session_id is caw's own bookkeeping id (the on-disk directory name). The next ProviderSession.send must resume rather than start fresh.

When trajectory is given, the prior history/usage is carried forward via ProviderSession._restore_from_trajectory. When it is None (e.g. resuming without a data_dir), the backend session is still resumed but caw's trajectory starts empty.

Source code in caw/provider.py

def resume_session(
    self,
    mcp_servers: list[MCPServer],
    *,
    session_id: str,
    resume_key: str,
    trajectory: Trajectory | None = None,
    **kwargs: Any,
) -> ProviderSession:
    """Rebuild a live session ready to continue an existing conversation.

    ``resume_key`` is the provider-side key the backend CLI needs to resume
    (see `ProviderSession.resume_key`); ``session_id`` is caw's own
    bookkeeping id (the on-disk directory name).  The next
    `ProviderSession.send` must resume rather than start fresh.

    When ``trajectory`` is given, the prior history/usage is carried forward
    via `ProviderSession._restore_from_trajectory`.  When it is
    ``None`` (e.g. resuming without a ``data_dir``), the backend session is
    still resumed but caw's trajectory starts empty.
    """
    raise NotImplementedError(f"{self.name} provider does not support resuming sessions.")

ProviderSession

Bases: ABC

ABC that each provider implements to manage a live session.

trajectory abstractmethod property

trajectory: Trajectory

The accumulated trajectory so far.

session_id property

session_id: str | None

Provider-assigned session ID (if any).

resume_key property

resume_key: str | None

The provider-side key needed to resume this conversation.

This is whatever the backend CLI accepts to continue an existing session (claude's session id, codex's thread id, opencode's session id). None until the backend has assigned one — typically after the first send.

last_raw_output property

last_raw_output: str | None

Raw CLI stdout from the most recent send() call (if available).

send abstractmethod

send(message: str) -> Turn

Send a message and return the agent's response turn.

Source code in caw/provider.py

@abstractmethod
def send(self, message: str) -> Turn:
    """Send a message and return the agent's response turn."""
    ...

end abstractmethod

end() -> Trajectory

Finalize the session and return the complete trajectory.

Source code in caw/provider.py

@abstractmethod
def end(self) -> Trajectory:
    """Finalize the session and return the complete trajectory."""
    ...

detect_usage_limit

detect_usage_limit(turn: Turn) -> int | None

Check whether turn indicates the provider's usage limit was hit.

Returns the number of minutes to wait before retrying, or None if no limit was detected. Override in provider subclasses to implement provider-specific detection logic.

Source code in caw/provider.py

def detect_usage_limit(self, turn: Turn) -> int | None:
    """Check whether *turn* indicates the provider's usage limit was hit.

    Returns the number of minutes to wait before retrying, or ``None``
    if no limit was detected.  Override in provider subclasses to
    implement provider-specific detection logic.
    """
    return None

set_step_callback

set_step_callback(callback: Callable[[list], None] | None) -> None

Set callback invoked after each step within send().

Source code in caw/provider.py

def set_step_callback(self, callback: Callable[[list], None] | None) -> None:
    """Set callback invoked after each step within send()."""
    pass  # default no-op; concrete providers override

set_logger

set_logger(logger: AgentLogger | None) -> None

Attach a generic logger that receives a one-line summary per event.

Source code in caw/provider.py

def set_logger(self, logger: AgentLogger | None) -> None:
    """Attach a generic logger that receives a one-line summary per event."""
    pass  # default no-op; concrete providers override

Built-in providers

ClaudeCodeProvider

Bases: Provider

Provider that delegates to the claude CLI.

CodexProvider

Bases: Provider

Provider that delegates to the codex CLI.

start_interactive

start_interactive(initial_prompt: str, mcp_servers: list[MCPServer], capture_bytes: int = 0, **kwargs: Any) -> InteractiveResult

Launch codex interactively (TUI) with an initial prompt.

Invoking the bare codex binary with a positional prompt starts its full-screen interactive session. We run it through a pty so the user drives the TUI directly while a copy of stdout is captured.

The sandbox mapping mirrors the headless exec path (see CodexSession.send): an explicit restrictive sandbox is passed through as --sandbox; otherwise codex runs with approvals and the sandbox bypassed (--dangerously-bypass-approvals-and-sandbox).

Source code in caw/providers/codex.py

def start_interactive(
    self, initial_prompt: str, mcp_servers: list[MCPServer], capture_bytes: int = 0, **kwargs: Any
) -> InteractiveResult:
    """Launch ``codex`` interactively (TUI) with an initial prompt.

    Invoking the bare ``codex`` binary with a positional prompt starts its
    full-screen interactive session.  We run it through a pty so the user
    drives the TUI directly while a copy of stdout is captured.

    The sandbox mapping mirrors the headless ``exec`` path (see
    ``CodexSession.send``): an explicit restrictive ``sandbox`` is passed
    through as ``--sandbox``; otherwise codex runs with approvals and the
    sandbox bypassed (``--dangerously-bypass-approvals-and-sandbox``).
    """
    from caw._pty import drive_interactive_pty

    cmd = ["codex"]

    model = kwargs.get("model")
    if model:
        cmd += ["-m", model]

    reasoning = kwargs.get("reasoning")
    if reasoning:
        cmd += ["-c", f'model_reasoning_effort="{reasoning}"']

    sandbox = kwargs.get("sandbox")
    if sandbox is None or sandbox == "danger-full-access":
        cmd += ["--dangerously-bypass-approvals-and-sandbox"]
    else:
        # The user is present in interactive mode, so codex's default
        # approval flow handles escalation; just set the sandbox policy.
        # (Unlike the headless `exec` path, the top-level TUI rejects
        # `--full-auto`.)
        cmd += ["--sandbox", sandbox]

    cmd += _build_mcp_config_args(mcp_servers)

    # codex has no --system-prompt flag; prepend it like the headless path.
    system_prompt = kwargs.get("system_prompt")
    prompt = f"{system_prompt}\n\n{initial_prompt}" if system_prompt else initial_prompt
    cmd.append(prompt)

    return drive_interactive_pty(cmd, capture_bytes=capture_bytes)

OpencodeProvider

Bases: Provider

Provider that delegates to the opencode CLI.

start_interactive

start_interactive(initial_prompt, mcp_servers, capture_bytes=0, **kwargs)

Launch opencode interactively (TUI) with an initial prompt.

opencode's interactive mode uses a full split-footer TUI. We launch it through a pty so stdin/stdout/stderr are inherited and a copy of stdout is captured.

Source code in caw/providers/opencode.py

def start_interactive(self, initial_prompt, mcp_servers, capture_bytes=0, **kwargs):  # type: ignore[override]
    """Launch ``opencode`` interactively (TUI) with an initial prompt.

    opencode's interactive mode uses a full split-footer TUI. We launch it
    through a pty so stdin/stdout/stderr are inherited and a copy of stdout
    is captured.
    """
    from caw._pty import drive_interactive_pty

    cmd = [_find_opencode_binary(), "run", "--interactive"]

    model = kwargs.get("model")
    if model:
        cmd += ["--model", model]
    reasoning = kwargs.get("reasoning")
    if reasoning:
        cmd += ["--variant", reasoning]
    if kwargs.get("dangerously_skip_permissions", True):
        cmd += ["--dangerously-skip-permissions"]

    config_path: str | None = None
    disabled_tools = kwargs.get("disabled_tools")
    if mcp_servers or disabled_tools:
        config: dict[str, Any] = {"$schema": "https://opencode.ai/config.json"}
        if mcp_servers:
            mcp: dict[str, Any] = {}
            for srv in mcp_servers:
                if srv.url:
                    entry: dict[str, Any] = {"type": "remote", "url": srv.url}
                else:
                    entry = {"type": "local", "command": [srv.command] + list(srv.args or [])}
                    if srv.env:
                        entry["environment"] = srv.env
                mcp[srv.name] = entry
            config["mcp"] = mcp
        if disabled_tools:
            config["tools"] = {name: False for name in disabled_tools}
        fd, config_path = tempfile.mkstemp(suffix=".json", prefix="caw_opencode_")
        with os.fdopen(fd, "w") as f:
            json.dump(config, f)

    cmd.append(initial_prompt)

    env = {"OPENCODE_CONFIG": config_path} if config_path else None

    def _cleanup() -> None:
        if config_path and os.path.exists(config_path):
            try:
                os.unlink(config_path)
            except OSError:
                pass

    return drive_interactive_pty(cmd, env=env, capture_bytes=capture_bytes, on_exit=_cleanup)

Registry & fallback order

register_provider

register_provider(name: str, cls: type[Provider]) -> None

Register a provider class under the given name.

Source code in caw/agent.py

def register_provider(name: str, cls: type[Provider]) -> None:
    """Register a provider class under the given name."""
    _PROVIDER_REGISTRY[name] = cls

set_provider_order

set_provider_order(order: list[str | tuple[str, str | ModelTier]] | None, *, models: dict[str, str | ModelTier] | None = None) -> None

Set the global provider fallback order, optionally with per-provider models.

Pass provider names/aliases in priority order, e.g. set_provider_order(["claude", "codex", "opencode"]). Agents created with provider="auto" (or with no provider and no CAW_PROVIDER) will select the first installed provider in this order and gracefully fall back to the next on a first-send failure. Pass None to clear it.

To also pin a model per provider, give (name, model) tuples and/or a models mapping; both accept a concrete model string or a ModelTier::

set_provider_order([("claude", ModelTier.STRONGEST), ("codex", "gpt-5.5")])
set_provider_order(["claude", "codex"], models={"claude": "opus"})

A provider's order-model is applied only when the Agent itself sets no model — an explicit model= (or CAW_MODEL) on the Agent wins. Because the model is attached to a specific provider, it is honored even when that provider is reached as a fallback (unlike a bare Agent-level model string, which is provider-specific and dropped on fallback).

An explicit provider= argument on an Agent always overrides this order.

Source code in caw/agent.py

def set_provider_order(
    order: list[str | tuple[str, str | ModelTier]] | None,
    *,
    models: dict[str, str | ModelTier] | None = None,
) -> None:
    """Set the global provider fallback order, optionally with per-provider models.

    Pass provider names/aliases in priority order, e.g.
    ``set_provider_order(["claude", "codex", "opencode"])``.  Agents created
    with ``provider="auto"`` (or with no ``provider`` and no ``CAW_PROVIDER``)
    will select the first *installed* provider in this order and gracefully
    fall back to the next on a first-send failure.  Pass ``None`` to clear it.

    To also pin a model per provider, give ``(name, model)`` tuples and/or a
    ``models`` mapping; both accept a concrete model string or a `ModelTier`::

        set_provider_order([("claude", ModelTier.STRONGEST), ("codex", "gpt-5.5")])
        set_provider_order(["claude", "codex"], models={"claude": "opus"})

    A provider's order-model is applied only when the Agent itself sets no
    ``model`` — an explicit ``model=`` (or ``CAW_MODEL``) on the Agent wins.
    Because the model is attached to a specific provider, it is honored even
    when that provider is reached as a fallback (unlike a bare Agent-level model
    string, which is provider-specific and dropped on fallback).

    An explicit ``provider=`` argument on an Agent always overrides this order.
    """
    global _PROVIDER_ORDER, _PROVIDER_ORDER_MODELS
    if not order:
        _PROVIDER_ORDER = None
        _PROVIDER_ORDER_MODELS = {}
        return
    names: list[str] = []
    order_models: dict[str, str | ModelTier] = {}
    for item in order:
        if isinstance(item, (tuple, list)):
            name = str(item[0])
            model = item[1] if len(item) > 1 else None
            if model is not None:
                order_models[name] = model
        else:
            name = str(item)
        names.append(name)
    if models:
        for name, model in models.items():
            if model is not None:
                order_models[str(name)] = model
    _PROVIDER_ORDER = names
    _PROVIDER_ORDER_MODELS = order_models

get_provider_order

get_provider_order() -> list[str] | None

Return the global provider fallback order, or None if unset.

Source code in caw/agent.py

def get_provider_order() -> list[str] | None:
    """Return the global provider fallback order, or ``None`` if unset."""
    return list(_PROVIDER_ORDER) if _PROVIDER_ORDER else None