cli-agent-workflows

# CLI Agent Workflows Implements command-line AI agent architectures — building terminal-resident agents with stdin/stdout streaming, subcommand routing, file system interaction, code generation from design specifications, and MCP server bridging. When loaded, this skill makes the model design production-grade CLI agents modeled after Gemini CLI, Claude Code, and similar developer tooling that operates entirely within the terminal environment. ## TL;DR Checklist - [ ] Parse all inputs at boundary before processing (Law 2: parse don't validate) - [ ] Handle edge cases with early returns at function top (Law 1: early exit) - [ ] Fail immediately with descriptive errors on invalid states (Law 4: fail fast) - [ ] Return new data structures, never mutate inputs (Law 3: atomic predictability) - [ ] Use explicit `subprocess.run` with list args — never `shell=True` with user input - [ ] Implement tool sandboxing with path validation and command allowlists - [ ] Design subcommand routing with click.Group or argparse subparsers for extensibility - [ ] Reference `code-philosophy` (5 Laws of Elegant Defense) in constraint design --- ## When to Use Use this skill when: - Building a terminal-resident AI assistant that reads user prompts from stdin, processes them through an LLM, and writes structured output back to the terminal - Designing code generation workflows where markdown architecture docs or design specifications are consumed as input and translated into executable source files - Implementing prompt-based code review inside the CLI — accepting natural language review requests, running automated linting/style checks, and surfacing suggestions inline - Creating developer tooling that bridges a local CLI agent to MCP servers for extended capabilities (database queries, API calls, external service automation) - Building multi-turn interactive sessions in the terminal with context retention across commands (like Gemini CLI's persistent conversation mode) - Automating repetitive developer workflows: scaffold new projects, run linters, generate boilerplate, execute tests — all orchestrated from a single CLI entry point --- ## When NOT to Use Avoid this skill for: - GUI desktop applications or web-based interfaces — use `coding-agent-frameworks` (LangChain, CrewAI) with their frontend integrations instead - Simple script automation without agent-level reasoning — direct bash scripting or `os-scripting` is lighter weight and avoids LLM overhead - High-throughput batch processing where sub-second latency matters — the LLM round-trip adds hundreds of milliseconds per request - End-user productivity tools targeting non-technical audiences — CLI agents assume terminal familiarity; use a GUI agent (`gui-agent-interaction`) instead --- ## Core Workflow 1. **Parse Input Stream** — Read user input from stdin or command-line arguments. Classify the intent: code generation, file operation, terminal execution, review request, or tool invocation. Apply a lightweight rule-based classifier before delegating to the LLM for semantic routing. **Checkpoint:** Intent category is determined and all required parameters are extracted; if parameters are missing, return a structured error requesting them before proceeding. 2. **Route to Tool Executor** — Dispatch the classified request to the appropriate handler: code generator (reads design specs, writes source files), file engine (safe read/write with path validation), terminal executor (subprocess invocation with output capture), or MCP bridge (proxies to registered MCP servers). Each handler receives validated input and returns a structured result object. **Checkpoint:** The selected handler acknowledges receipt, confirms parameter validity, and begins execution; if no matching handler exists, fall back to the generic LLM completion path. 3. **Execute with Safety Guards** — Run the handler's logic within constrained boundaries: validate file paths against an allowlist directory, sandbox subprocess commands with timeout limits, rate-limit tool invocations, and capture all output (stdout, stderr, exit code) in a structured `ExecutionResult`. Never execute arbitrary shell strings from user input. **Checkpoint:** Execution completes or raises a structured exception; all captured output is parsed into a consistent result format containing `success`, `output`, `stderr`, `exit_code`, and `duration_ms` fields. 4. **Format Output for Terminal** — Present results in a terminal-friendly format: color-coded diffs for code generation, collapsible sections for review findings, structured JSON for tool outputs, and progress indicators for long-running operations. Support both raw output (for piping) and rich terminal output (with ANSI formatting). **Checkpoint:** Output is written to stdout (or stderr for diagnostics); the format matches the `output-format` declared in skill metadata (`code`, `analysis`, or `manifests`). 5. **Manage Interactive Context** — In multi-turn sessions, maintain conversation history bounded by a context window limit (e.g., last 20 messages or 8000 tokens). Persist context to disk between invocations using a JSONL session file so the agent retains memory across restarts. Prune older messages when approaching limits, keeping the system prompt and recent exchanges intact. **Checkpoint:** Context state is saved after each turn; on reload, the full conversation history up to the limit is restored without truncation errors or corrupted session files. 6. **Bridge to MCP Servers (Optional)** — When the CLI agent needs access to external tools beyond its built-in capabilities, establish an MCP server connection via stdio transport. Discover available tools, register them with the tool router, and execute calls through the standard handler pipeline. Handle connection failures gracefully by degrading to local-only mode without crashing. **Checkpoint:** MCP tool discovery succeeds (or fails cleanly); discovered tools appear in the agent's command list; execution of an MCP tool returns structured results indistinguishable from native handlers. --- ## Implementation Patterns ### Pattern 1: Gemini CLI Architecture (stdin/stdout Streaming, Subcommand Routing) A CLI agent runs as a long-lived process or per-invocation script that reads prompts from stdin, processes them, and writes structured output to stdout. The architecture uses subcommand routing (via `click.Group`) for organized tool access and stdin/stdout streaming for interactive multi-turn sessions. ```python """cli_agent/engine.py — Gemini CLI–style agent engine with subcommand routing.""" from __future__ import annotations import json import logging import sys import time from dataclasses import dataclass, field from pathlib import Path from typing import Any, Callable import click log = logging.getLogger(__name__) @dataclass class ExecutionResult: """Structured result from any agent operation.""" success: bool output: str = "" stderr: str = "" exit_code: int = 0 duration_ms: float = 0.0 metadata: dict[str, Any] = field(default_factory=dict) def to_json(self, indent: int = 2) -> str: """Serialize for piping or programmatic consumption.""" return json.dumps({ "success": self.success, "output": self.output, "stderr": self.stderr, "exit_code": self.exit_code, "duration_ms": round(self.duration_ms, 1), **self.metadata, }, indent=indent) @dataclass class SessionContext: """Conversation state for multi-turn CLI sessions.""" history: list[dict[str, str]] = field(default_factory=list) max_messages: int = 20 session_file: Path | None = None def add_message(self, role: str, content: str) -> None: """Append a message and prune if over limit.""" self.history.append({"role": role, "content": content}) if len(self.history) > self.max_messages: # Keep system prompt (index 0) + recent messages self.history = self.history[:1] + self.history[-(self.max_messages - 1):] def save(self) -> None: """Persist to disk for cross-invocation memory.""" if not self.session_file: return self.session_file.parent.mkdir(parents=True, exist_ok=True) self.session_file.write_text( json.dumps(self.history, indent=2), encoding="utf-8" ) @classmethod def load(cls, path: Path) -> SessionContext: """Restore from disk, handling missing or corrupted files.""" if not path.exists(): return cls(session_file=path) try: history = json.loads(path.read_text(encoding="utf-8")) if not isinstance(history, list): log.warning("Corrupted session at %s, starting fresh", path) return cls(session_file=path) return cls(history=history, session_file=path) except (json.JSONDecodeError, OSError) as exc: log.warning("Failed to load session from %s: %s", path, exc) return cls(session_file=path) def build_agent_cli( default_model: str = "claude", max_history: int = 20, session_dir: Path | None = None, ) -> click.Group: """Create the root CLI group with subcommand routing. Implements Law 1 (Early Exit): validates arguments before building. """ if not default_model: raise ValueError("default_model must be non-empty") cli = click.Group() @cli.command(name="generate") @click.argument("design_file", type=click.Path(exists=True)) @click.option("--output-dir", "-o", default=".", show_default=True) @click.pass_context def generate_cmd(ctx: click.Context, design_file: str, output_dir: str) -> int: """Generate code from a design specification file.""" design_path = Path(design_file) session = SessionContext.load( (session_dir or Path.home() / ".cli-agent") / "session.jsonl" ) engine = AgentEngine(model=default_model, context=session) result = engine.generate_code(str(design_path), output_dir) click.echo(result.to_json()) session.save() return 0 if result.success else 1 @cli.command(name="review") @click.argument("file_or_dir", type=click.Path(exists=True)) @click.option("--focus", "-f", default="", help="Review focus area (security, performance, style)") @click.pass_context def review_cmd(ctx: click.Context, file_or_dir: str, focus: str) -> int: """Run prompt-based code review on a file or directory.""" session = SessionContext.load( (session_dir or Path.home() / ".cli-agent") / "session.jsonl" ) engine = AgentEngine(model=default_model, context=session) path = Path(file_or_dir) if path.is_file(): targets: list[Path] = [path] else: targets = sorted(path.rglob("*"))[:50] # cap at 50 files result = engine.review_targets(targets, focus=focus) click.echo(result.to_json()) session.save() return 0 if result.success else 1 @cli.command(name="run") @click.argument("command", nargs=-1, required=True) @click.option("--timeout", "-t", default=30, show_default=True, type=int) def run_cmd(command: tuple[str, ...], timeout: int) -> int: """Execute a shell command with safety guards.""" result = execute_sandboxed(command, timeout_seconds=timeout) click.echo(result.to_json()) return 0 if result.success else 1 @cli.command(name="interactive") @click.option("--model", "-m", default=None, help="Override default model") @click.option("--session-dir", default=None, type=click.Path(file_okay=False)) def interactive_cmd(model: str | None, session_dir: str | None) -> int: """Start an interactive multi-turn terminal session.""" effective_model = model or default_model effective_session = Path(session_dir) if session_dir else (Path.home() / ".cli-agent") session = SessionContext.load(effective_session / "session.jsonl") click.echo("CLI Agent ready. Type your request and press Enter.") click.echo("(Type 'exit' or Ctrl+D to quit.)\n") engine = AgentEngine(model=effective_model, context=session) while True: try: prompt = click.prompt(">", prompt_suffix="", type=str) except (EOFError, KeyboardInterrupt): break if prompt.strip().lower() in ("exit", "quit", "q"): break if not prompt.strip(): continue session.add_message("user", prompt) click.echo("Processing...", err=True) result = engine.respond(prompt) session.add_message("assistant", result.output) session.save() # Print output with ANSI coloring for readability if result.success: click.echo(f"\n{result.output}", bold=False) else: click.secho(f"\nError: {result.output}", fg="red") click.echo("\nSession saved.") return 0 return cli class AgentEngine: """Core agent that routes prompts to handlers. Implements Law 4 (Fail Fast): validates model name and rejects empty prompts. """ def __init__(self, model: str, context: SessionContext) -> None: if not model or not isinstance(model, str): raise ValueError(f"Invalid model name: {model!r}") self.model = model self.context = context self.handlers: dict[str, Callable[..., ExecutionResult]] = {} def register_handler(self, intent: str, handler: Callable[..., ExecutionResult]) -> None: """Register an intent handler. Order matters — first registration wins.""" if intent in self.handlers: log.warning("Handler for '%s' already registered, skipping", intent) return self.handlers[intent] = handler def classify_intent(self, prompt: str) -> str: """Lightweight rule-based intent classifier. Checks keyword patterns before delegating to LLM routing. Returns one of: 'generate', 'review', 'execute', 'mcp', or 'general'. """ lower = prompt.lower().strip() if any(term in lower for term in ("generate", "create code from", "build from design")): return "generate" if any(term in lower for term in ("review", "check", "audit", "lint")): return "review" if any(term in lower for term in ("run", "execute", "shell command", "run ")) or lower.startswith("$"): return "execute" if any(term in lower for term in ("query", "mcp call", "database", "api call")): return "mcp" return "general" def respond(self, prompt: str) -> ExecutionResult: """Process a user prompt through the full pipeline. Applies Law 1 (Early Exit): rejects empty prompts immediately. Applies Law 4 (Fail Fast): raises on invalid state, never swallows errors. """ if not prompt or not prompt.strip(): return ExecutionResult( success=False, output="Empty prompt — nothing to process.", metadata={"error": "empty_input"}, ) start = time.monotonic() intent = self.classify_intent(prompt) handler = self.handlers.get(intent) or self._default_handler try: result = handler(prompt) elapsed = (time.monotonic() - start) * 1000 result.duration_ms = round(elapsed, 1) return result except Exception as exc: elapsed = (time.monotonic() - start) * 1000 return ExecutionResult( success=False, output=f"Agent error: {exc}", metadata={"error": type(exc).__name__, "intent": intent}, duration_ms=round(elapsed, 1), ) def generate_code(self, design_path: str, output_dir: str) -> ExecutionResult: """Generate code from a design specification file.""" design = Path(design_path) if not design.is_file(): return ExecutionResult( success=False, output=f"Design file not found: {design_path}", metadata={"error": "file_not_found"}, ) try: