orchestrata_ultimate.providers.adapters.claude_cli_adapter

Reference › orchestrata_ultimate › providers › adapters › claude_cli_adapter

claude_cli_adapter is the one provider adapter that does not use HTTP: ClaudeCLIAdapter runs the Claude Code command-line tool as a subprocess and still satisfies the provider protocol, so the router treats it like any other provider. client registers it without needing an API key, since the CLI authenticates through its own login.

_find_claude locates the binary. complete flattens the messages into one prompt (_build_prompt), maps the model name to the CLI's alias (_map_model turns the dated Sonnet and Opus names and the short names sonnet and opus into CLI model values), and runs claude --print --model <alias> --tools "" with the prompt on standard input. It runs in a temporary scratch directory unless a working directory was given, with a timeout set in the constructor; a missing binary raises RuntimeError. _subprocess_env passes the inherited environment on without ANTHROPIC_API_KEY, because the presence of that variable makes the CLI ignore its login. _build_response turns the return code, standard output and standard error into a raw provider response. Pricing is empty (_build_pricing), get_models and supports_capability describe the configured models, and close has nothing to release.

At a glance

Class diagram

Dependency diagram

API

Classes

ClaudeCLIAdapter

Line 76

Provider adapter that calls the Claude Code CLI.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/providers/adapters/claude_cli_adapter.py

"""Claude CLI adapter — wraps the Claude Code CLI as a ProviderProtocol.

Uses `claude --print` subprocess to send prompts and collect responses.
This adapter enables the production ProviderManager to use Claude Code
without an API key (OAuth-based authentication via the CLI).
"""

import asyncio
import logging
import os
import shutil
import tempfile
from collections.abc import Mapping
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    ModelCapability,
    ModelInfo,
    ProviderConfig,
)

_logger = logging.getLogger(__name__)

TIMEOUT_SECONDS = 600


def _find_claude(env: Mapping[str, str] | None = None) -> str:
    """Find the claude CLI binary.

    Args:
        env: Environment mapping used for ``PATH`` and ``HOME`` lookups.
            Defaults to ``os.environ`` so production callers are unaffected.
    """
    env = env if env is not None else os.environ
    found = shutil.which("claude", path=env.get("PATH"))
    if found:
        return found
    home = env.get("HOME") or os.path.expanduser("~")
    for candidate in [
        os.path.join(home, ".local", "bin", "claude"),
        "/usr/bin/claude",
        "/usr/local/bin/claude",
    ]:
        if os.path.isfile(candidate):
            return candidate
    return ""


def _build_prompt(messages: list[Any]) -> str:
    """Flatten system/user messages into a single CLI prompt string."""
    system_msg = ""
    user_msg = ""
    for msg in messages:
        role = getattr(msg, "role", "")
        content = getattr(msg, "content", "")
        if role == "system":
            system_msg = content
        elif role == "user":
            user_msg = content

    if system_msg:
        return f"{system_msg}\n\n{user_msg}"
    return user_msg


def _map_model(model: str) -> str:
    """Map an API/alias model name to the Claude CLI ``--model`` value."""
    if model in ("claude-sonnet-4-6", "sonnet"):
        return "sonnet"
    if model in ("claude-opus-4-6", "opus"):
        return "opus"
    return model


class ClaudeCLIAdapter:
    """Provider adapter that calls the Claude Code CLI.

    Implements the same interface as HTTP-based adapters but uses
    subprocess execution instead of API calls. Supports OAuth
    authentication (no API key needed).

    Sandbox contract: this adapter is a TEXT-COMPLETION seam — its return
    value is read only as text, never acted on as an agentic session. Every
    invocation therefore (1) runs the spawned CLI in a fresh, per-call
    scratch directory instead of the caller's cwd, and (2) passes
    ``--tools ""`` to disable the CLI's built-in tool set entirely. Without
    these, the child inherits the caller's cwd and full tool permissions
    (Write/Edit/Bash), which lets it perform unsupervised host-level
    mutations a text-completion caller never authorized (observed: files
    written into the repo root during a cassette-recording run).
    """

    def __init__(
        self,
        timeout_seconds: float = TIMEOUT_SECONDS,
        config: ProviderConfig | None = None,
        env: Mapping[str, str] | None = None,
        workdir: str | None = None,
    ) -> None:
        """Initialize the Claude CLI adapter.

        Args:
            timeout_seconds: Per-call subprocess timeout. Defaults to the
                module-level ``TIMEOUT_SECONDS``.
            config: Provider configuration (models/pricing), used by
                ``get_models()``/``supports_capability()`` to satisfy
                ``ProviderProtocol`` like the HTTP-based sibling adapters.
                Defaults to an empty ``ProviderConfig`` for callers that
                construct this adapter without one.
            env: Environment mapping used to locate the ``claude`` binary.
                Defaults to ``os.environ``.
            workdir: Fixed working directory for the spawned CLI subprocess.
                Defaults to ``None``, meaning each ``complete()`` call gets
                its own fresh ``tempfile.mkdtemp`` scratch directory (removed
                afterwards). Only set this if the caller genuinely needs a
                specific working directory -- most callers should not.
        """
        self._claude_bin = _find_claude(env)
        self._pricing: dict[str, Any] = {}
        self._timeout_seconds = timeout_seconds
        self._config = config or ProviderConfig()
        self._workdir = workdir

    async def complete(
        self,
        messages: list[Any],
        model: str,
        parameters: dict[str, Any],
    ) -> Any:
        """Send a prompt to Claude CLI and return the response.

        Args:
            messages: List of Message objects with role and content.
            model: Model name (e.g., "sonnet", "opus").
            parameters: Generation parameters (temperature, etc.).

        Returns:
            RawProviderResponse with content and token counts.
        """
        if not self._claude_bin:
            raise RuntimeError("Claude CLI not found")

        prompt = _build_prompt(messages)
        cli_model = _map_model(model)

        cmd = [
            self._claude_bin,
            "--print",
            "--model",
            cli_model,
            # Text-completion sandbox contract (see class docstring): the
            # spawned CLI gets no tool access at all -- its output is read
            # only as text, never trusted to act.
            "--tools",
            "",
            "-",
        ]

        scratch_dir: str | None = None
        cwd = self._workdir
        if cwd is None:
            scratch_dir = tempfile.mkdtemp(prefix="claude-cli-adapter-")
            cwd = scratch_dir

        # The CLI authenticates by claude.ai LOGIN. It refuses to use that
        # login whenever ANTHROPIC_API_KEY is set — "connectors are disabled
        # because ANTHROPIC_API_KEY or another auth source is set and takes
        # precedence over your claude.ai login" — so a stray key in the
        # environment does not merely go unused, it DISABLES the working auth
        # path. Measured 2026-08-29: a `CHANGEME_ANTHROPIC_API_KEY`
        # placeholder exported into the session made every one of the 11
        # volks fail, while `env -u ANTHROPIC_API_KEY claude -p` answered
        # normally.
        #
        # Blanking it in .env is not enough: a process that inherited the
        # export before the file changed still carries it, and this adapter
        # is invoked from long-lived sessions. Strip it at the subprocess
        # boundary instead, which is the only place that can be sure. An
        # explicit key belongs to the API adapters, not to the CLI one.
        env = self._subprocess_env(os.environ)

        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                cwd=cwd,
                env=env,
                stdin=asyncio.subprocess.PIPE,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )

            try:
                stdout, stderr = await asyncio.wait_for(
                    proc.communicate(input=prompt.encode("utf-8")),
                    timeout=self._timeout_seconds,
                )
                return self._build_response(
                    prompt, cli_model, proc.returncode, stdout, stderr
                )

            except TimeoutError:
                # Kill the orphaned subprocess and reap it so we don't leak
                # the process and its stdio file descriptors.
                try:
                    proc.kill()
                except ProcessLookupError:
                    pass  # already exited
                await proc.wait()
                raise RuntimeError(
                    f"Claude CLI timed out after {self._timeout_seconds}s"
                )
        finally:
            if scratch_dir is not None:
                shutil.rmtree(scratch_dir, ignore_errors=True)

    @staticmethod
    def _subprocess_env(source: Mapping[str, str]) -> dict[str, str]:
        """The environment the CLI subprocess gets: everything but the key.

        ``ANTHROPIC_API_KEY`` is not merely unused by this adapter — its
        presence DISABLES the claude.ai login the CLI authenticates with
        ("connectors are disabled because ANTHROPIC_API_KEY ... takes
        precedence over your claude.ai login"). A stray or placeholder value
        anywhere in the inherited environment therefore breaks every call,
        and blanking it in .env cannot help a process that inherited the
        export earlier. Removing it here is the only place with the whole
        picture.
        """
        return {k: v for k, v in source.items() if k != "ANTHROPIC_API_KEY"}

    def _build_response(
        self,
        prompt: str,
        cli_model: str,
        returncode: int | None,
        stdout: bytes,
        stderr: bytes,
    ) -> Any:
        """Turn the CLI subprocess output into a RawProviderResponse.

        A non-zero exit is a FAILURE, whatever landed on stdout. The guard
        here used to be ``if not content``, which made the failure conditional
        on stdout being EMPTY — and ``claude -p`` writes its errors to stdout,
        not stderr. Measured 2026-08-29 with a placeholder ANTHROPIC_API_KEY:
        the CLI printed ``Invalid API key · Fix external API key`` on stdout
        and exited 1, so that string was returned as a SUCCESSFUL completion.
        All 11 volks "answered" with it and ``pipeline_quick_test.py``
        reported ``Status: resolved`` with one artifact whose content was the
        auth error. A provider that turns rc=1 into a green result cannot be
        distinguished from one that works, which is the worst thing an
        adapter can do.

        Raises:
            RuntimeError: If the CLI exited non-zero.
        """
        from orchestrata_ultimate.core.protocols import RawProviderResponse

        content = stdout.decode("utf-8", errors="replace").strip()

        if returncode != 0:
            err = stderr.decode("utf-8", errors="replace")
            _logger.warning(
                "Claude CLI exited %d: %s",
                returncode,
                err[:200] or content[:200],
            )
            raise RuntimeError(
                f"Claude CLI failed (rc={returncode}): "
                f"stderr={err[:200]!r} stdout={content[:200]!r}"
            )

        return RawProviderResponse(
            content=content,
            input_tokens=len(prompt) // 4,
            output_tokens=len(content) // 4,
            model=cli_model,
        )

    @traces(spec="011", story="US-001", reqs=["REQ-011-024"])
    async def close(self) -> None:
        """No resources to close for CLI adapter."""

    def _build_pricing(self) -> dict[str, Any]:
        """Return empty pricing (CLI usage is free via OAuth)."""
        return {}

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-028"])
    def get_models(self) -> list[ModelInfo]:
        """Return available models with metadata (ProviderProtocol).

        Sourced from the ``ProviderConfig`` passed to ``__init__``, mirroring
        the HTTP-based sibling adapters (AnthropicAdapter, DeepSeekAdapter, etc).
        """
        models = [
            ModelInfo(
                name=model_name,
                provider="claude_cli",
                capabilities=pricing.capabilities,
                context_window=pricing.context_window,
                pricing=pricing,
            )
            for model_name, pricing in self._config.pricing.items()
        ]
        return models

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-007"])
    def supports_capability(self, capability: ModelCapability) -> bool:
        """Check if any configured model supports the capability (ProviderProtocol)."""
        for pricing in self._config.pricing.values():
            if capability in pricing.capabilities:
                return True
        return False