claude.py•2.25 kB
"""Claude Code CLI adapter."""
import asyncio
import subprocess
from typing import Any
from .base import CLIAdapter
class ClaudeAdapter(CLIAdapter):
"""Adapter for Claude Code CLI."""
async def execute(self, task: str, progress_callback: Any = None, timeout: int | None = None, **kwargs: Any) -> tuple[str, str, int]:
"""Execute task using Claude Code with optional streaming."""
if progress_callback:
return await self.execute_streaming(task, progress_callback, timeout, **kwargs)
cmd = self.format_task(task, **kwargs)
resolved_cmd = self.resolve_command(cmd)
process = await asyncio.create_subprocess_exec(
*resolved_cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env={**subprocess.os.environ, **self.get_env()},
)
try:
effective_timeout = timeout or self.get_timeout()
stdout, stderr = await asyncio.wait_for(
process.communicate(), timeout=effective_timeout
)
return (
stdout.decode("utf-8", errors="replace"),
stderr.decode("utf-8", errors="replace"),
process.returncode or 0,
)
except asyncio.TimeoutError:
process.kill()
await process.wait()
raise TimeoutError(f"Claude CLI timed out after {effective_timeout}s")
def validate(self) -> bool:
"""Validate Claude CLI is available."""
try:
subprocess.run(
["which", "claude"] if subprocess.os.name != "nt" else ["where", "claude"],
capture_output=True,
check=True,
)
return True
except subprocess.CalledProcessError:
return False
def format_task(self, task: str, **kwargs: Any) -> list[str]:
"""Format task for Claude CLI."""
cmd = ["claude"]
# Add any custom args from config
cmd.extend(self.get_args())
# Add mode if specified
if mode := kwargs.get("mode"):
cmd.extend(["--mode", mode])
# Add task as final argument
cmd.append(task)
return cmd