orchestrata_ultimate.drivers.claude_api_driver

Reference › orchestrata_ultimate › drivers › claude_api_driver

claude_api_driver drives the Claude Code command-line tool as an agentic coder that writes files straight into the workspace, passing the task through standard input (SPEC-025, REQ-025-027). Despite the module name it uses the CLI and not the HTTP API. ClaudeAPIDriver is the driver that driver_factory returns for the claude and anthropic providers and also as its fallback for unknown providers. execute finds the binary with _find_claude, initialises the project if no CLAUDE.md exists yet (_init_project), runs the task via stdin (_run_claude) with a timeout of 1200 seconds, and then lists the source and configuration files that were created with _scan_source_files. The scan skips directories such as the virtual environment, caches, .git and the design folder, and picks up Python, TOML, INI, CFG and YAML files. A missing binary or a timeout produces a failure result and not an exception. It satisfies the AsyncCodeDriver protocol; tests are in tests/drivers/test_claude_api_driver.py.

At a glance

Class diagram

Dependency diagram

API

Classes

ClaudeAPIDriver

Decorators: traces(spec='025', reqs=['REQ-025-027']) · Line 98

Async code gen driver using Claude Code CLI with stdin pipe.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/drivers/claude_api_driver.py

# mypy: ignore-errors
"""Async Claude Code CLI driver with stdin pipe and workspace scanning.

Uses Claude Code as an agentic coding tool that writes files directly
to disk. Scans workspace for created files after completion.
"""

import asyncio
import logging
import os
import shutil
from collections.abc import Mapping

from orchestrata_ultimate.core.traceability import traces

_logger = logging.getLogger(__name__)

TIMEOUT_SECONDS = 1200

_SKIP_DIRS = frozenset(
    {
        ".venv",
        "__pycache__",
        ".git",
        "node_modules",
        ".claude",
        "design",
    }
)
_SKIP_PREFIXES = ("Module:",)
_SOURCE_EXTS = frozenset(
    {
        ".py",
        ".toml",
        ".cfg",
        ".ini",
        ".yaml",
        ".yml",
    }
)


@traces(spec="025", story="US-025-001", reqs=["REQ-025-019", "REQ-025-020"])
def _find_claude(env: Mapping[str, str] | None = None) -> str:
    """Find the claude CLI binary.

    Args:
        env: Optional environment mapping. Defaults to ``os.environ``.

    Returns:
        Path to claude binary, or empty string if not found.
    """
    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 _scan_source_files(workspace: str) -> list[str]:
    """Scan workspace for Python/config source files.

    Args:
        workspace: Root directory to scan.

    Returns:
        Sorted list of relative file paths.
    """
    files: list[str] = []
    for root, dirs, filenames in os.walk(workspace):
        dirs[:] = [
            d
            for d in dirs
            if d not in _SKIP_DIRS and not any(d.startswith(p) for p in _SKIP_PREFIXES)
        ]
        for fname in filenames:
            if fname.startswith("."):
                continue
            ext = os.path.splitext(fname)[1]
            if ext in _SOURCE_EXTS:
                rel = os.path.relpath(os.path.join(root, fname), workspace)
                files.append(rel)
    return sorted(files)


@traces(
    spec="025",
    reqs=["REQ-025-027"],
)
class ClaudeAPIDriver:
    """Async code gen driver using Claude Code CLI with stdin pipe.

    Passes prompt via stdin to avoid E7 "Argument list too long"
    on large prompts (40+ files). Scans workspace afterward.

    Attributes:
        workspace: Directory where code is generated.
    """

    def __init__(
        self,
        workspace: str | None = None,
        env: Mapping[str, str] | None = None,
    ) -> None:
        """Initialize with target workspace directory.

        Args:
            workspace: Working directory. Created if needed.
            env: Optional environment mapping. Defaults to ``os.environ``.
        """
        self._env = env if env is not None else os.environ
        self.workspace = workspace or self._env.get(
            "ORCHESTRATA_WORKSPACE", "/home/mohamed/workspace"
        )
        os.makedirs(self.workspace, exist_ok=True)
        # Overridable so callers (and tests) can bound the wait without
        # touching the module-level default; falls back to TIMEOUT_SECONDS.
        self._timeout_seconds: float = float(
            self._env.get("ORCHESTRATA_CLAUDE_TIMEOUT_SECONDS", TIMEOUT_SECONDS)
        )

    @traces(
        spec="025",
        story="US-025-001",
        reqs=["REQ-025-030", "REQ-025-033", "REQ-025-034"],
    )
    async def execute(
        self,
        task: str,
        files: list[str] | None = None,
        is_fix: bool = False,
    ) -> tuple[bool, str]:
        """Generate code by running Claude Code in the project dir.

        Args:
            task: The coding task/prompt.
            files: Optional file paths (unused — Claude reads workspace).
            is_fix: True if this is a repair prompt.

        Returns:
            Tuple of (success, output_or_file_count).
        """
        claude_bin = _find_claude(self._env)
        if not claude_bin:
            msg = "claude CLI not found"
            _logger.error(msg)
            return False, msg

        _logger.info("Generating code in %s", self.workspace)
        await self._init_project(claude_bin)

        try:
            output = await asyncio.wait_for(
                self._run_claude(claude_bin, task),
                timeout=self._timeout_seconds,
            )
        except TimeoutError:
            msg = f"Timed out after {self._timeout_seconds}s"
            _logger.warning(msg)
            return False, msg
        except Exception as exc:
            msg = f"Error: {exc}"
            _logger.error(msg)
            return False, msg

        created = await asyncio.to_thread(_scan_source_files, self.workspace)
        if not created:
            _logger.warning("No source files created")
            return False, output or "No files generated"

        for f in created:
            _logger.info("Wrote %s", f)

        _logger.info("Generated %d files", len(created))
        return True, output or f"Created {len(created)} files"

    @traces(
        spec="025",
        story="US-025-001",
        reqs=["REQ-025-028", "REQ-025-029"],
    )
    async def _init_project(self, claude_bin: str) -> None:
        """Run claude init if CLAUDE.md doesn't exist yet.

        Args:
            claude_bin: Path to claude binary.
        """
        claude_md = os.path.join(self.workspace, "CLAUDE.md")
        if os.path.exists(claude_md):
            return
        try:
            proc = await asyncio.create_subprocess_exec(
                claude_bin,
                "init",
                "--yes",
                cwd=self.workspace,
                stdout=asyncio.subprocess.DEVNULL,
                stderr=asyncio.subprocess.DEVNULL,
            )
            await asyncio.wait_for(proc.wait(), timeout=30)
            _logger.info("Initialized project")
        except (TimeoutError, FileNotFoundError) as exc:
            _logger.debug("project init skipped: %s", exc)

    @traces(spec="025", story="US-025-001", reqs=["REQ-025-027"])
    async def _run_claude(self, claude_bin: str, task: str) -> str:
        """Run claude with task via stdin pipe.

        Args:
            claude_bin: Path to claude binary.
            task: The coding task prompt.

        Returns:
            Claude Code stdout output.
        """
        proc = await asyncio.create_subprocess_exec(
            claude_bin,
            "--dangerously-skip-permissions",
            "--print",
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.STDOUT,
            cwd=self.workspace,
        )
        # stdin/stdout are typed Optional by asyncio's stubs (None only when
        # the corresponding pipe isn't requested); both are requested above.
        assert proc.stdin is not None
        assert proc.stdout is not None

        proc.stdin.write(task.encode("utf-8"))
        await proc.stdin.drain()
        proc.stdin.close()

        output_lines: list[str] = []
        while True:
            line = await proc.stdout.readline()
            if not line:
                break
            decoded = line.decode("utf-8", errors="replace")
            stripped = decoded.strip()
            if stripped:
                _logger.debug("[Claude-API] %s", stripped)
            output_lines.append(decoded)

        await proc.wait()
        return "".join(output_lines)