""" Claude Code Plugin for Agentsd ============================== Bridges locally installed Claude Code CLI into agentsd as a standard plugin. Enables agentsd to use Claude Code for code generation, editing, and reasoning. Architecture: agentsd (Rust) ←gRPC→ claudecode_plugin.py → claude CLI (subprocess) Capabilities: - chat: Send a prompt to Claude Code, get response - code: Generate/edit code with full Claude Code capabilities - session: Resume previous Claude Code sessions - tools: Let Claude Code use its built-in tools (Bash, Edit, Read, etc.) Usage: # Start agentsd first, then: python claudecode_plugin.py """ import json import logging import os import subprocess import sys import time from pathlib import Path from concurrent import futures import grpc # Add SDK path SDK_DIR = Path(__file__).resolve().parent.parent / "_sdk" sys.path.insert(0, str(SDK_DIR)) import agentsd_pb2 as pb import agentsd_pb2_grpc as rpc logging.basicConfig(level=logging.INFO, format="[claudecode] %(levelname)s %(message)s") logger = logging.getLogger(__name__) # Find claude CLI CLAUDE_BIN = os.environ.get("CLAUDE_BIN", None) def find_claude(): """Verify claude CLI is available and return the path.""" global CLAUDE_BIN if CLAUDE_BIN is None: # Auto-detect candidates = [ "claude", "claude.cmd", os.path.expandvars(r"%APPDATA%\npm\claude.cmd"), os.path.expanduser("~/.local/bin/claude"), "/usr/local/bin/claude", ] for c in candidates: try: result = subprocess.run( [c, "--version"], capture_output=True, text=True, timeout=10 ) if result.returncode == 0: CLAUDE_BIN = c break except (FileNotFoundError, subprocess.TimeoutExpired, OSError): continue if CLAUDE_BIN is None: logger.error("Claude Code CLI not found") logger.error("Install: npm install -g @anthropic-ai/claude-code") return None try: result = subprocess.run( [CLAUDE_BIN, "--version"], capture_output=True, text=True, timeout=10 ) version = result.stdout.strip() logger.info("Found Claude Code: %s at %s", version, CLAUDE_BIN) return version except Exception as e: logger.error("Claude Code CLI error: %s", e) return None class ClaudeCodePlugin: """Bridges Claude Code CLI into agentsd.""" def __init__(self, agentsd_addr: str = "localhost:50051"): self.addr = agentsd_addr self.channel = None self.bridge = None self.version = None self.default_cwd = os.getcwd() def connect(self): """Connect to agentsd.""" self.channel = grpc.insecure_channel(self.addr) self.bridge = rpc.PluginBridgeStub(self.channel) logger.info("Connected to agentsd at %s", self.addr) def register(self): """Register with agentsd.""" resp = self.bridge.Register(pb.PluginInfo( id="claudecode", name="Claude Code", version=self.version or "unknown", plugin_type="standard", syscalls=["process.spawn", "display.text", "fs.read", "fs.write", "network.request"], depends=["hermes"], exports=[ pb.ExportDef(name="chat", description="Send prompt to Claude Code, get text response"), pb.ExportDef(name="code", description="Generate or edit code with Claude Code"), pb.ExportDef(name="review", description="Review code with Claude Code"), pb.ExportDef(name="ask", description="Ask Claude Code a question about the codebase"), pb.ExportDef(name="run", description="Run Claude Code with full tool access"), ], )) if resp.ok: logger.info("Registered with agentsd: session=%s", resp.session_id) else: logger.error("Registration failed: %s", resp.error) return resp.ok def call_claude(self, prompt: str, cwd: str = None, max_turns: int = None, allowed_tools: list = None, system_prompt: str = None, output_format: str = "text") -> dict: """ Call Claude Code CLI in print mode. Args: prompt: The prompt to send cwd: Working directory for Claude Code max_turns: Max agentic turns (default: unlimited) allowed_tools: List of tools to allow (e.g. ["Bash", "Read", "Edit"]) system_prompt: Override system prompt output_format: "text" or "json" or "stream-json" Returns: dict with "result" or "error" """ cmd = [CLAUDE_BIN, "-p", "--output-format", output_format] if max_turns is not None: cmd.extend(["--max-turns", str(max_turns)]) if allowed_tools: cmd.extend(["--allowedTools"] + allowed_tools) if system_prompt: cmd.extend(["--system-prompt", system_prompt]) work_dir = cwd or self.default_cwd logger.info("Calling claude: cwd=%s prompt=%s...", work_dir, prompt[:60]) try: result = subprocess.run( cmd, input=prompt, capture_output=True, text=True, cwd=work_dir, timeout=300, # 5 min max env={**os.environ, "CLAUDE_CODE_SIMPLE": "1"}, ) if result.returncode == 0: output = result.stdout.strip() if output_format == "json": try: return json.loads(output) except json.JSONDecodeError: return {"result": output} return {"result": output} else: error = result.stderr.strip() or f"exit code {result.returncode}" return {"error": error} except subprocess.TimeoutExpired: return {"error": "timeout (300s)"} except Exception as e: return {"error": str(e)} def handle_call(self, fn_name: str, args: dict) -> dict: """Route a call to the appropriate handler.""" prompt = args.get("prompt", args.get("text", "")) cwd = args.get("cwd", self.default_cwd) if fn_name == "chat": return self.call_claude( prompt=prompt, cwd=cwd, max_turns=1, # Single turn, no tools allowed_tools=[], output_format="text", ) elif fn_name == "code": return self.call_claude( prompt=prompt, cwd=cwd, allowed_tools=["Bash", "Read", "Edit", "Write", "Glob", "Grep"], output_format="text", ) elif fn_name == "review": review_prompt = f"Review the following code or changes. Be concise and actionable:\n\n{prompt}" return self.call_claude( prompt=review_prompt, cwd=cwd, max_turns=1, allowed_tools=["Read", "Glob", "Grep"], output_format="text", ) elif fn_name == "ask": return self.call_claude( prompt=prompt, cwd=cwd, max_turns=3, allowed_tools=["Read", "Glob", "Grep"], output_format="text", ) elif fn_name == "run": # Full access mode return self.call_claude( prompt=prompt, cwd=cwd, output_format="text", ) else: return {"error": f"unknown function: {fn_name}"} def run_server(self, port: int = 50053): """Run callback gRPC server for agentsd.""" server = grpc.server(futures.ThreadPoolExecutor(max_workers=4)) rpc.add_PluginBridgeServicer_to_server(ClaudeCodeServicer(self), server) server.add_insecure_port(f"[::]:{port}") server.start() logger.info("Claude Code plugin server listening on port %d", port) return server class ClaudeCodeServicer(rpc.PluginBridgeServicer): """gRPC servicer for Claude Code plugin.""" def __init__(self, plugin: ClaudeCodePlugin): self.plugin = plugin def Call(self, request, context): fn_name = request.fn try: args = json.loads(request.args) if request.args else {} except json.JSONDecodeError: args = {} logger.info("Call: %s(%s)", fn_name, list(args.keys())) result = self.plugin.handle_call(fn_name, args) result_bytes = json.dumps(result, ensure_ascii=False, default=str).encode("utf-8") if "error" in result and not result.get("result"): return pb.CallResponse(result=result_bytes, ok=False, error=result["error"]) return pb.CallResponse(result=result_bytes, ok=True, error="") def Register(self, request, context): context.set_code(grpc.StatusCode.UNIMPLEMENTED) return pb.RegisterResponse(ok=False, error="not a bridge") def Heartbeat(self, request, context): return pb.HeartbeatResponse(ok=True) def StreamCall(self, request, context): context.set_code(grpc.StatusCode.UNIMPLEMENTED) return pb.CallResponse(ok=False, error="not implemented") def main(): # Verify claude CLI version = find_claude() if not version: sys.exit(1) plugin = ClaudeCodePlugin() plugin.version = version # Connect and register plugin.connect() if not plugin.register(): sys.exit(1) # Start callback server server = plugin.run_server(port=50053) logger.info("Claude Code plugin running (v%s). Exports: chat, code, review, ask, run", version) try: while True: time.sleep(1) except KeyboardInterrupt: logger.info("Shutting down...") server.stop(grace=5) plugin.channel.close() if __name__ == "__main__": main()