Billy Brenner

# Grok Swarm

**Multi-agent intelligence powered by Grok 4.20 Multi-Agent Beta**

Give any AI coding agent access to a 4-agent swarm with ~2M token context for code analysis, refactoring, generation, and complex reasoning.

- **Version:** 1.0.8
- **Platforms:** OpenClaw, Claude Code
- **Modes:** analyze, refactor, code, reason, orchestrate

---

## Overview

Grok 4.20 coordinates 4 agents (orchestrator + specialists + critics) to:
- Analyze codebases for security, architecture, and bugs
- Refactor code while preserving behavior
- Generate features, tests, and boilerplate
- Reason through complex architectural decisions

## Features

- **4-Agent Coordination** — Multi-perspective reasoning
- **Massive Context** — ~2M token window
- **File Writing** — Write annotated code blocks directly to disk
- **Tool Passthrough** — Use OpenAI-format tools with Grok

## Usage

### OpenClaw

```javascript
tools.grok_swarm({
  prompt: "Analyze security of this auth module",
  mode: "analyze",
  files: ["src/auth/*.ts"]
});
```

### Claude Code

```
/grok-swarm:analyze Review auth module security
/grok-swarm:refactor Convert to async/await
/grok-swarm:code Write FastAPI endpoint
```

## Task Modes

| Mode | Description |
|------|-------------|
| `analyze` | Security audits, architecture review |
| `refactor` | Modernization, migration |
| `code` | Feature generation, tests |
| `reason` | Multi-perspective reasoning |
| `orchestrate` | Custom agent handoff |

## Requirements

- Python 3.8+
- Node.js 18+
- `openai>=1.0.0`
- OpenRouter API key with Grok 4.20 access

## API Key

Set your API key:

```bash
export OPENROUTER_API_KEY=sk-or-v1-...
```

Or create `~/.config/grok-swarm/config.json`:

```bash
mkdir -p ~/.config/grok-swarm
echo '{"api_key": "sk-or-v1-..."}' > ~/.config/grok-swarm/config.json
chmod 600 ~/.config/grok-swarm/config.json
```

## Installation

```bash
# Via ClawHub
clawhub install grok-swarm

# Via npm
npm install @khaentertainment/grok-swarm
```

FILE:bridge/apply.py
#!/usr/bin/env python3
"""
apply.py - Parse code blocks from Grok responses and write them to files.

Handles markdown code blocks with optional language hints and file paths
from fenced code block attributes.
"""

import argparse
import re
import sys
from pathlib import Path


def parse_code_blocks(markdown_text):
    blocks = []
    lines = markdown_text.split('\n')
    i = 0

    while i < len(lines):
        line = lines[i]
        fence_match = re.match(r'^(`{3,})(.*)$', line)
        if fence_match:
            fence = fence_match.group(1)
            header = fence_match.group(2).strip()
            parts = header.split(None, 1)
            lang = parts[0] if parts else ""
            path_hint = parts[1] if len(parts) > 1 else ""

            code_lines = []
            i += 1
            while i < len(lines):
                if re.match(r'^' + re.escape(fence) + r'$', lines[i]):
                    i += 1  # Skip past closing fence
                    break
                code_lines.append(lines[i])
                i += 1

            code = '\n'.join(code_lines).strip()

            if code:
                blocks.append({
                    "language": lang,
                    "code": code,
                    "path_hint": path_hint
                })
        else:
            i += 1

    return blocks


def infer_filename(block, base_dir):
    if block.get("path_hint"):
        return block["path_hint"]

    code = block["code"]
    lang = block.get("language", "").lower()

    shebang_match = re.match(r'#!\S+/(\S+)', code.split('\n')[0])
    if shebang_match:
        name = shebang_match.group(1)
        return f"script.{name}"

    lang_map = {
        "python": "output.py", "py": "output.py",
        "javascript": "output.js", "js": "output.js",
        "typescript": "output.ts", "ts": "output.ts",
        "rust": "output.rs", "go": "output.go",
        "java": "Output.java", "c": "output.c",
        "cpp": "output.cpp", "c++": "output.cpp",
        "ruby": "output.rb", "rb": "output.rb",
        "php": "output.php", "shell": "output.sh",
        "bash": "output.sh", "sh": "output.sh",
        "yaml": "output.yaml", "yml": "output.yml",
        "json": "output.json", "toml": "output.toml",
        "html": "output.html", "css": "output.css",
        "sql": "output.sql", "markdown": "output.md",
        "md": "output.md",
    }

    if lang in lang_map:
        return lang_map[lang]

    return "output.txt"


def apply_blocks(blocks, base_dir, dry_run=True):
    base = Path(base_dir).resolve()
    if not dry_run:
        base.mkdir(parents=True, exist_ok=True)

    files_written = 0
    files_skipped = 0
    changes = []

    for block in blocks:
        filename = infer_filename(block, base_dir)
        filepath = (base / filename).resolve()

        # Validate containment using try/except instead of is_relative_to (Python 3.8 compat)
        try:
            rel = filepath.relative_to(base)
            rel_path = str(rel)
        except ValueError:
            # Path is outside base_dir
            files_skipped += 1
            changes.append({
                "path": str(filepath),
                "action": "skipped",
                "reason": "path outside base_dir"
            })
            continue

        if dry_run:
            files_skipped += 1
            action = "would write"
        else:
            filepath.parent.mkdir(parents=True, exist_ok=True)
            filepath.write_text(block["code"])
            files_written += 1
            action = "written"

        changes.append({
            "path": rel_path,
            "action": action,
            "language": block.get("language", ""),
            "size": len(block["code"])
        })

    return {
        "files_written": files_written,
        "files_skipped": files_skipped,
        "changes": changes
    }


def format_summary(result, base_dir):
    lines = []
    lines.append(f"\n{'='*60}")
    lines.append(f"Applied code blocks to: {base_dir}/")
    lines.append(f"{'='*60}")

    for change in result["changes"]:
        path = change["path"]
        action = change["action"]
        lang = change.get("language", "")
        size = change.get("size", 0)

        if action == "skipped":
            lines.append(f"  SKIPPED: {path} ({change.get('reason', 'n/a')})")
        else:
            lines.append(f"  {path} ({lang}, {size:,} chars) - {action}")

    lines.append(f"\nSummary: {result['files_written']} written, {result['files_skipped']} skipped")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="Parse code blocks from Grok responses and write to files")
    parser.add_argument("input", help="Input markdown file, or - for stdin")
    parser.add_argument("--base-dir", "-d", default="./grok-output", help="Base directory for output files")
    parser.add_argument("--yes", "-y", action="store_true", help="Actually write files (default is dry-run)")
    parser.add_argument("--json", action="store_true", help="Output JSON summary")

    args = parser.parse_args()

    if args.input == "-":
        markdown_text = sys.stdin.read()
    else:
        markdown_text = Path(args.input).read_text()

    blocks = parse_code_blocks(markdown_text)

    if not blocks:
        print("No code blocks found in input.", file=sys.stderr)
        sys.exit(0)

    print(f"Found {len(blocks)} code block(s)", file=sys.stderr)

    result = apply_blocks(blocks, args.base_dir, dry_run=not args.yes)

    if args.json:
        import json
        print(json.dumps(result, indent=2))
    else:
        print(format_summary(result, args.base_dir))


if __name__ == "__main__":
    main()
FILE:bridge/cli.py
#!/usr/bin/env python3
"""
Unified Grok Swarm CLI entrypoint.
Dispatches to refactor/analyze/code/reason modes using grok_bridge.py logic.

Supports file writing when Grok generates code:
    --output-dir <path>  Directory to write files to
    --apply              Actually write files (dry-run by default)
    --execute <cmd>      Run a command after generation
    --use-morph          Use Morph LLM MCP for file edits if available
"""

import argparse
import json
import subprocess
import sys
import time
from pathlib import Path

# Support both source and installed package layouts
current = Path(__file__).parent
if str(current) not in sys.path:
    sys.path.insert(0, str(current))

from grok_bridge import call_grok, read_files, MODE_PROMPTS


def check_morph_available():
    """Check if Morph LLM MCP is installed."""
    try:
        result = subprocess.run(
            ["claude", "mcp", "list"],
            capture_output=True,
            text=True,
            timeout=10
        )
        return "morph" in result.stdout.lower()
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return False


def apply_with_morph(blocks, base_dir):
    """
    Apply code blocks using Morph LLM MCP edit_file tool.
    Returns summary dict.
    """
    import uuid

    applied = 0
    errors = []
    base_resolved = Path(base_dir).resolve()

    for block in blocks:
        # For Morph, we prefer partial edits. Use path_hint as target.
        path_hint = block.get("path_hint", "")
        if not path_hint:
            # Try to infer from code
            inferred_path = block.get("inferred_path", "")
            if not inferred_path:
                # No valid path - skip this block
                errors.append("No path_hint or inferred_path provided - skipping partial edit")
                continue
            path_hint = inferred_path

        # Sanitize and validate path_hint to prevent path traversal
        # Check if path_hint is absolute
        path_obj = Path(path_hint)
        if path_obj.is_absolute():
            errors.append(f"{path_hint}: absolute paths not allowed")
            continue

        # Resolve the full path and ensure it's within base_dir
        try:
            target_path = (base_resolved / path_hint).resolve()
            # Validate containment
            target_path.relative_to(base_resolved)
            validated_path = str(target_path)
        except ValueError:
            errors.append(f"{path_hint}: path traversal detected - outside base_dir")
            continue

        # Execute via claude mcp
        try:
            result = subprocess.run(
                ["claude", "mcp", "call", "morphllm", "edit_file",
                 "--file", validated_path,
                 "--code", block["code"],
                 "--language", block.get("language", "")],
                capture_output=True,
                text=True,
                timeout=30
            )
            if result.returncode == 0:
                applied += 1
            else:
                errors.append(f"{path_hint}: {result.stderr}")
        except (OSError, subprocess.SubprocessError) as e:
            errors.append(f"{path_hint}: {e}")

    return {
        "applied": applied,
        "total": len(blocks),
        "errors": errors
    }


def parse_and_write(result_text, output_dir, dry_run=True):
    """
    Parse code blocks from Grok response and write to files.
    Returns summary string.
    """
    from apply import parse_code_blocks, apply_blocks, format_summary

    blocks = parse_code_blocks(result_text)

    if not blocks:
        return "No code blocks found in response."

    if dry_run:
        # Just show what would happen
        base = Path(output_dir)
        summaries = []
        for block in blocks:
            from apply import infer_filename
            filename = infer_filename(block, output_dir)
            summaries.append(f"  • {filename} ({block.get('language', 'text')}, {len(block['code']):,} chars)")
        return f"\nFound {len(blocks)} code block(s) — dry-run:\n" + "\n".join(summaries)

    # Actually write
    result = apply_blocks(blocks, output_dir, dry_run=False)
    return format_summary(result, output_dir)


def main():
    parser = argparse.ArgumentParser(
        description="Grok Swarm — Multi-agent CLI powered by Grok 4.20",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "mode",
        nargs="?",
        choices=list(MODE_PROMPTS.keys()),
        default="reason",
        help="Task mode (default: reason)",
    )
    parser.add_argument("--prompt", "-p", required=True, help="Task instruction or question")
    parser.add_argument("--files", "-f", nargs="*", default=[], help="Files to include as context")
    parser.add_argument("--system", "-s", help="Override system prompt")
    parser.add_argument("--tools", "-t", help="Path to OpenAI-format tools JSON")
    parser.add_argument("--timeout", type=int, default=120, help="Timeout in seconds")
    parser.add_argument("--output", help="Write raw output to file")
    parser.add_argument("--output-dir", "--od", "-d",
                        help="Directory to write generated files (used with --apply)")
    parser.add_argument("--apply", "-a", action="store_true",
                        help="Actually write files from code blocks (dry-run by default)")
    parser.add_argument("--execute", "-e", metavar="CMD",
                        help="Execute command after generation (shell string)")
    parser.add_argument("--use-morph", action="store_true",
                        help="Use Morph LLM MCP for file edits if available")

    args = parser.parse_args()

    if args.mode == "orchestrate" and not args.system:
        print("ERROR: orchestrate mode requires --system", file=sys.stderr)
        sys.exit(1)

    # Check Morph availability if --use-morph is set
    use_morph = False
    if args.use_morph:
        if check_morph_available():
            use_morph = True
            print("Morph LLM MCP detected — will use for file edits", file=sys.stderr)
        else:
            print("WARNING: --use-morph set but Morph LLM MCP not found", file=sys.stderr)
            print("  Install with: claude mcp add morphllm", file=sys.stderr)

    context = read_files(args.files) if args.files else ""

    if args.files:
        print(f"Read {len(args.files)} file(s) — {len(context):,} chars", file=sys.stderr)

    # Parse tools if provided
    tools = None
    if args.tools:
        with open(args.tools, 'r', encoding='utf-8') as f:
            tools = json.load(f)

    print(f"Calling Grok 4.20 (mode={args.mode}, 4 agents)...", file=sys.stderr)

    start = time.time()
    result = call_grok(
        prompt=args.prompt,
        mode=args.mode,
        context=context,
        system_override=args.system,
        tools=tools,
        timeout=args.timeout,
    )
    elapsed = time.time() - start

    # Preserve raw result for output
    raw_result = result

    # Handle JSON tool call responses
    response_data = None
    normalized_result = result
    if result.startswith("{") and '"tool_calls"' in result:
        try:
            response_data = json.loads(result)
            normalized_result = response_data.get("content", result)
        except json.JSONDecodeError:
            pass

    # File writing logic
    if args.apply or args.output_dir:
        output_dir = args.output_dir or "./grok-output"

        if use_morph and args.apply:
            # Use Morph LLM for edits
            from apply import parse_code_blocks
            blocks = parse_code_blocks(normalized_result)

            if not blocks:
                print("\nNo code blocks found to apply via Morph.", file=sys.stderr)
            else:
                morph_result = apply_with_morph(blocks, output_dir)
                print(f"\nApplied {morph_result['applied']}/{morph_result['total']} edits via Morph LLM",
                      file=sys.stderr)
                if morph_result['errors']:
                    for err in morph_result['errors']:
                        print(f"  Error: {err}", file=sys.stderr)
        else:
            # Use direct file writing
            summary = parse_and_write(normalized_result, output_dir, dry_run=not args.apply)
            print(summary, file=sys.stderr)

    # Write raw output if requested
    if args.output:
        output_path = Path(args.output)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_text(raw_result, encoding='utf-8')
        print(f"Output written to {args.output}", file=sys.stderr)

    # Execute command if requested
    if args.execute:
        print(f"\nExecuting: {args.execute}", file=sys.stderr)
        exec_result = subprocess.run(
            args.execute,
            shell=True,
            capture_output=True,
            text=True,
            timeout=300
        )
        if exec_result.stdout:
            print(exec_result.stdout)
        if exec_result.stderr:
            print(exec_result.stderr, file=sys.stderr)
        if exec_result.returncode != 0:
            print(f"Command exited with code {exec_result.returncode}", file=sys.stderr)
            sys.exit(exec_result.returncode)

    # Output the normalized response to stdout (unless we wrote files and nothing else)
    if not args.output and not args.execute:
        print(normalized_result)

    print(f"\nCompleted in {elapsed:.1f}s", file=sys.stderr)


if __name__ == "__main__":
    main()
FILE:bridge/grok_bridge.py
#!/usr/bin/env python3
"""
grok_bridge.py — General-purpose bridge to xAI Grok 4.20 Multi-Agent Beta via OpenRouter.

Supports multiple task modes, custom system prompts, file context ingestion,
and OpenAI-format tool use passthrough.

Usage:
    python3 grok_bridge.py --prompt "Refactor this" --mode refactor --files a.js b.js
    python3 grok_bridge.py --prompt "Analyze security" --mode analyze --files src/*.py
    python3 grok_bridge.py --prompt "Build feature" --mode orchestrate --system "You are a Go expert" --files main.go
"""

import argparse
import json
import os
import re
import sys
import time
from pathlib import Path

try:
    from openai import OpenAI
except ImportError:
    print("ERROR: openai package required. Install: pip3 install openai", file=sys.stderr)
    sys.exit(1)


OPENROUTER_BASE = "https://openrouter.ai/api/v1"
MODEL_ID = "x-ai/grok-4.20-multi-agent-beta"

# Mode-specific system prompts
MODE_PROMPTS = {
    "refactor": (
        "You are an expert code refactoring engineer. "
        "Improve code quality, maintainability, and performance while preserving behavior. "
        "Output refactored code with clear explanations of what changed and why."
    ),
    "analyze": (
        "You are a senior code analyst and security auditor. "
        "Examine the provided code for bugs, security vulnerabilities, performance issues, "
        "and architectural concerns. Be specific with file paths and line references. "
        "Prioritize findings by severity."
    ),
    "code": (
        "You are an expert software engineer. "
        "Write clean, production-ready code that follows best practices and idioms for the language. "
        "Include error handling, tests where appropriate, and clear comments."
    ),
    "reason": (
        "You are a collaborative multi-agent reasoning system. "
        "Consider multiple perspectives, weigh trade-offs, and provide well-reasoned analysis. "
        "Structure your response clearly with conclusions."
    ),
    "orchestrate": None,  # Requires --system flag
}


def get_api_key():
    """Resolve OpenRouter API key from config file, environment, or OpenClaw auth profiles."""

    # 1. Environment variables
    key = os.environ.get("OPENROUTER_API_KEY") or os.environ.get("XAI_API_KEY")
    if key:
        return key

    # 2. Grok Swarm config file (for Claude Code without secret management)
    grok_config = Path.home() / ".config" / "grok-swarm" / "config.json"
    if grok_config.exists():
        try:
            with open(grok_config) as f:
                data = json.load(f)
            key = data.get("api_key")
            if key:
                return key
        except (json.JSONDecodeError, KeyError):
            pass

    # 3. OpenClaw auth profiles (for OpenClaw integration)
    auth_paths = [
        Path.home() / ".openclaw" / "agents" / "coder" / "agent" / "auth-profiles.json",
        Path.home() / ".openclaw" / "agents" / "main" / "agent" / "auth-profiles.json",
        Path.home() / ".openclaw" / "auth-profiles.json",
        Path.home() / ".config" / "openclaw" / "auth-profiles.json",
    ]
    for auth_path in auth_paths:
        if auth_path.exists():
            try:
                with open(auth_path) as f:
                    data = json.load(f)
                profiles = data.get("profiles", {})
                or_profile = profiles.get("openrouter:default", {})
                key = or_profile.get("key") or or_profile.get("apiKey")
                if key:
                    return key
                default = profiles.get("default", {})
                key = default.get("openrouter", {}).get("apiKey") or default.get("openrouter", {}).get("key")
                if key:
                    return key
            except (json.JSONDecodeError, KeyError):
                continue
    return None


def read_files(file_paths):
    """Read and concatenate files with path headers."""
    chunks = []
    total_size = 0
    max_size = 1_500_000

    for fpath in file_paths:
        p = Path(fpath)
        if not p.exists():
            print(f"WARNING: File not found: {fpath}", file=sys.stderr)
            continue
        if not p.is_file():
            print(f"WARNING: Not a file: {fpath}", file=sys.stderr)
            continue

        content = p.read_text(errors="replace")
        header = f"\n{'='*60}\nFILE: {p.absolute()}\n{'='*60}\n"
        chunk = header + content

        if total_size + len(chunk) > max_size:
            print(f"WARNING: Size limit reached, skipping remaining files", file=sys.stderr)
            break

        chunks.append(chunk)
        total_size += len(chunk)

    return "\n".join(chunks)


def load_tools(tools_path):
    """Load OpenAI-format tool definitions from a JSON file."""
    if not tools_path:
        return None
    p = Path(tools_path)
    if not p.exists():
        print(f"ERROR: Tools file not found: {tools_path}", file=sys.stderr)
        sys.exit(1)
    with open(p) as f:
        tools = json.load(f)
    if not isinstance(tools, list):
        print(f"ERROR: Tools file must be a JSON array", file=sys.stderr)
        sys.exit(1)
    return tools


def _safe_dest(output_path, file_path):
    """
    Resolve ``file_path`` relative to ``output_path`` and verify the result
    stays inside ``output_path``.  Returns the resolved Path or raises
    ValueError for unsafe paths (absolute, containing ``..``, etc.).
    """
    raw = Path(file_path)
    if raw.is_absolute():
        raise ValueError(f"Absolute paths are not allowed: {file_path!r}")
    if ".." in raw.parts:
        raise ValueError(f"Path traversal not allowed: {file_path!r}")
    dest = (output_path / raw).resolve()
    resolved_root = output_path.resolve()
    try:
        dest.relative_to(resolved_root)
    except ValueError as exc:
        raise ValueError(f"Path escapes output directory: {file_path!r}") from exc
    return dest


def parse_and_write_files(response_text, output_dir):
    """
    Scan response for fenced code blocks with filename annotations and write to disk.
    
    Supports patterns:
      ```lang:path/to/file ... ```
      ```lang
      // FILE: path/to/file
      ...
      ```
    
    Returns list of (relative_path, byte_count) tuples written, where
    byte_count is the number of UTF-8 bytes written.
    """
    written = []
    output_path = Path(output_dir)
    
    # Pattern for lang:path at start of block (language tag contains path)
    lang_path_pattern = re.compile(r'^(\w+):([^\s\n]+)\n', re.MULTILINE)
    # Pattern for // FILE: or # FILE: markers
    file_marker_pattern = re.compile(r'^\s*(?://|#)\s*FILE:\s*(.+?)\s*$', re.MULTILINE)
    
    def _write_file(file_path, content):
        """Validate path, write content, and record result. Returns True on success."""
        try:
            dest = _safe_dest(output_path, file_path)
        except ValueError as exc:
            print(f"WARNING: Skipping unsafe path — {exc}", file=sys.stderr)
            return False
        dest.parent.mkdir(parents=True, exist_ok=True)
        encoded = content.strip().encode("utf-8", errors="replace")
        dest.write_bytes(encoded)
        written.append((file_path, len(encoded)))
        return True

    # Split into code blocks by ``` fences.
    # Even indices are fence markers or text between fences; skip them.
    # Odd indices are the actual code block contents.
    parts = re.split(r'```', response_text)
    
    for i, part in enumerate(parts):
        if i % 2 == 0:
            # Skip even-indexed parts (fences/text between fences)
            continue
        
        # Check for lang:path at start (language tag contains the path)
        lang_match = lang_path_pattern.match(part)
        if lang_match:
            _write_file(lang_match.group(2), part[lang_match.end():])
            continue
        
        # Check for // FILE: or # FILE: marker within the block
        marker_match = file_marker_pattern.search(part)
        if marker_match:
            _write_file(marker_match.group(1).strip(), part[marker_match.end():])
    
    return written

def call_grok(prompt, mode="reason", context="", system_override=None, tools=None, timeout=120):
    """Make the API call to Grok 4.20 Multi-Agent Beta."""
    api_key = get_api_key()
    if not api_key:
        print("ERROR: No OpenRouter API key found.", file=sys.stderr)
        print("Set OPENROUTER_API_KEY env var or configure in OpenClaw auth-profiles.json", file=sys.stderr)
        sys.exit(1)

    # Resolve system prompt
    if system_override:
        system_content = system_override
    else:
        system_content = MODE_PROMPTS.get(mode)
        if system_content is None:
            print(f"ERROR: Mode '{mode}' requires --system flag", file=sys.stderr)
            sys.exit(1)

    # Append context to system prompt
    if context:
        system_content += f"\n\n## Codebase Context\n{context}"

    client = OpenAI(
        base_url=OPENROUTER_BASE,
        api_key=api_key,
        timeout=timeout,
    )

    messages = [
        {"role": "system", "content": system_content},
        {"role": "user", "content": prompt},
    ]

    kwargs = {
        "model": MODEL_ID,
        "messages": messages,
        "max_tokens": 16384,
        "temperature": 0.3,
        "extra_body": {"agent_count": 4},
    }

    if tools:
        kwargs["tools"] = tools

    start = time.time()

    try:
        response = client.chat.completions.create(**kwargs)
    except Exception as e:
        elapsed = time.time() - start
        print(f"ERROR after {elapsed:.1f}s: {e}", file=sys.stderr)
        sys.exit(1)

    elapsed = time.time() - start

    if not response.choices:
        print(f"ERROR: Empty response after {elapsed:.1f}s", file=sys.stderr)
        sys.exit(1)

    choice = response.choices[0]

    # Log usage
    if hasattr(response, 'usage') and response.usage:
        u = response.usage
        print(f"USAGE: mode={mode} prompt={u.prompt_tokens} completion={u.completion_tokens} "
              f"total={u.total_tokens} time={elapsed:.1f}s", file=sys.stderr)

    # Handle content filtering
    if choice.finish_reason == "content_filter":
        print(f"WARNING: Response blocked by content filter after {elapsed:.1f}s", file=sys.stderr)
        if not choice.message.content:
            print("No content returned. Try rephrasing the prompt.", file=sys.stderr)
            sys.exit(2)

    content = choice.message.content or ""

    # Handle tool calls (return as JSON if present)
    if choice.message.tool_calls:
        tool_calls = []
        for tc in choice.message.tool_calls:
            tool_calls.append({
                "id": tc.id,
                "type": tc.type,
                "function": {
                    "name": tc.function.name,
                    "arguments": tc.function.arguments,
                }
            })
        result = {
            "content": content,
            "tool_calls": tool_calls,
        }
        return json.dumps(result, indent=2)

    return content.strip()


def main():
    parser = argparse.ArgumentParser(
        description="General-purpose bridge to xAI Grok 4.20 Multi-Agent Beta"
    )
    parser.add_argument("--prompt", required=True, help="Task instruction or question")
    parser.add_argument("--mode", default="reason", choices=list(MODE_PROMPTS.keys()),
                        help="Task mode (default: reason)")
    parser.add_argument("--files", nargs="*", default=[], help="Local file paths for context")
    parser.add_argument("--system", help="Override system prompt (for orchestrate mode)")
    parser.add_argument("--tools", help="Path to JSON file with OpenAI-format tool definitions")
    parser.add_argument("--timeout", type=int, default=120, help="Timeout in seconds (default: 120)")
    parser.add_argument("--output", help="Output file path (default: stdout)")
    parser.add_argument("--write-files", action="store_true",
                        help="Parse response for annotated code blocks and write to --output-dir")
    parser.add_argument("--output-dir", default="./grok-output/",
                        help="Directory for file writes (default: ./grok-output/)")

    args = parser.parse_args()

    # Validate orchestrate mode
    if args.mode == "orchestrate" and not args.system:
        print("ERROR: --mode orchestrate requires --system flag", file=sys.stderr)
        sys.exit(1)

    # Read context files
    context = ""
    if args.files:
        print(f"Reading {len(args.files)} files...", file=sys.stderr)
        context = read_files(args.files)
        print(f"Context size: {len(context):,} chars", file=sys.stderr)

    # Load tools
    tools = load_tools(args.tools)

    # Call Grok
    print(f"Calling {MODEL_ID} (mode={args.mode}, 4 agents, timeout={args.timeout}s)...", file=sys.stderr)
    result = call_grok(
        prompt=args.prompt,
        mode=args.mode,
        context=context,
        system_override=args.system,
        tools=tools,
        timeout=args.timeout,
    )

    # Output
    if args.output:
        Path(args.output).write_text(result)
        print(f"Written to: {args.output}", file=sys.stderr)

    if args.write_files:
        written = parse_and_write_files(result, args.output_dir)
        if written:
            total_bytes = sum(b for _, b in written)
            print(f"Wrote {len(written)} files to {args.output_dir}")
            for rel_path, byte_count in written:
                print(f"  {rel_path} ({byte_count:,} bytes)")
            print(f"Total: {total_bytes:,} bytes")
        else:
            print(
                "No annotated files found in model response to write to disk.\n"
                "Re-run without --write-files to see the full response.",
                file=sys.stderr,
            )
    elif not args.output:
        print(result)


if __name__ == "__main__":
    main()
FILE:bridge/index.js
#!/usr/bin/env node
/**
 * index.js — Node.js wrapper for grok_bridge.py
 *
 * Enforces timeout at the process level to prevent 502 Bad Gateway errors.
 * Supports multiple task modes, custom system prompts, file context, and tool use.
 *
 * Usage:
 *   node index.js --prompt "Analyze security" --mode analyze --files src/*.js
 *   node index.js --prompt "Build feature" --mode orchestrate --system "You are a Go expert" --files main.go
 */

const { spawn } = require('child_process');
const path = require('path');

const BRIDGE_SCRIPT = path.join(__dirname, 'grok_bridge.py');
const PYTHON = path.join(__dirname, '.venv', 'bin', 'python3');

const VALID_MODES = ['refactor', 'analyze', 'code', 'reason', 'orchestrate'];

function parseArgs() {
  const args = process.argv.slice(2);
  const parsed = {
    prompt: null,
    mode: 'reason',
    files: [],
    system: null,
    tools: null,
    timeout: 120,
    output: null,
    writeFiles: false,
    outputDir: './grok-output/',
  };

  for (let i = 0; i < args.length; i++) {
    switch (args[i]) {
      case '--prompt':
        parsed.prompt = args[++i];
        break;
      case '--mode':
        parsed.mode = args[++i];
        break;
      case '--files':
        while (i + 1 < args.length && !args[i + 1].startsWith('--')) {
          parsed.files.push(args[++i]);
        }
        break;
      case '--system':
        parsed.system = args[++i];
        break;
      case '--tools':
        parsed.tools = args[++i];
        break;
      case '--timeout':
        parsed.timeout = parseInt(args[++i], 10) || 120;
        break;
      case '--output':
        parsed.output = args[++i];
        break;
      case '--write-files':
        parsed.writeFiles = true;
        break;
      case '--output-dir':
        parsed.outputDir = args[++i];
        break;
      case '--help':
        console.log(`
grok_swarm — Bridge to xAI Grok 4.20 Multi-Agent Beta (4-agent swarm)

Usage:
  node index.js --prompt "instruction" [options]

Options:
  --prompt <text>     Task instruction (required)
  --mode <mode>       Task mode: refactor|analyze|code|reason|orchestrate (default: reason)
  --files <path...>   Files to include as context
  --system <text>     Override system prompt (required for orchestrate mode)
  --tools <path>      JSON file with OpenAI-format tool definitions
  --timeout <secs>    Timeout in seconds (default: 120)
  --output <path>     Output file (default: stdout)
  --write-files       Parse response for annotated code blocks and write files
  --output-dir <path> Directory for file writes (default: ./grok-output/)
  --help              Show this help

Modes:
  refactor   Large-scale code refactoring, modernization, migration
  analyze    Code review, security audit, architecture assessment
  code       Generate new code, implement features, write tests
  reason     Complex reasoning, research synthesis, multi-perspective analysis
  orchestrate  Full prompt control — requires --system flag
`);
        process.exit(0);
    }
  }

  if (!parsed.prompt) {
    console.error('ERROR: --prompt is required');
    process.exit(1);
  }

  if (!VALID_MODES.includes(parsed.mode)) {
    console.error(`ERROR: Invalid mode 'parsed.mode'. Valid: VALID_MODES.join(', ')`);
    process.exit(1);
  }

  if (parsed.mode === 'orchestrate' && !parsed.system) {
    console.error('ERROR: --mode orchestrate requires --system flag');
    process.exit(1);
  }

  return parsed;
}

function run() {
  const opts = parseArgs();

  // Build Python args
  const pyArgs = [
    BRIDGE_SCRIPT,
    '--prompt', opts.prompt,
    '--mode', opts.mode,
    '--timeout', String(opts.timeout),
  ];

  if (opts.files.length > 0) {
    pyArgs.push('--files', ...opts.files);
  }

  if (opts.system) {
    pyArgs.push('--system', opts.system);
  }

  if (opts.tools) {
    pyArgs.push('--tools', opts.tools);
  }

  if (opts.output) {
    pyArgs.push('--output', opts.output);
  }

  if (opts.writeFiles) {
    pyArgs.push('--write-files');
  }

  // Skip --output-dir if it matches the Python bridge default.
  // This optimization reduces CLI noise; CLI invocations bypass this wrapper
  // via src/plugin/index.ts, which sets defaults at the plugin layer.
  if (opts.outputDir && opts.outputDir !== './grok-output/') {
    pyArgs.push('--output-dir', opts.outputDir);
  }

  // Spawn Python process
  const child = spawn(PYTHON, pyArgs, {
    stdio: ['inherit', 'pipe', 'pipe'],
    env: { ...process.env },
  });

  let stdout = '';
  let stderr = '';
  let timedOut = false;

  // Enforce timeout at process level
  const timer = setTimeout(() => {
    timedOut = true;
    child.kill('SIGTERM');
    setTimeout(() => child.kill('SIGKILL'), 5000);
  }, opts.timeout * 1000);

  child.stdout.on('data', (data) => {
    stdout += data.toString();
  });

  child.stderr.on('data', (data) => {
    stderr += data.toString();
    process.stderr.write(data);
  });

  child.on('close', (code) => {
    clearTimeout(timer);

    if (timedOut) {
      console.error(`\nTIMEOUT: Grok call exceeded opts.timeouts limit`);
      process.exit(124);
    }

    if (code !== 0) {
      console.error(`\nBridge exited with code code`);
      process.exit(code);
    }

    if (opts.output) {
      console.error(`Output written to: opts.output`);
    } else {
      process.stdout.write(stdout);
    }
  });

  child.on('error', (err) => {
    clearTimeout(timer);
    console.error(`Failed to spawn bridge: err.message`);
    process.exit(1);
  });
}

run();
FILE:install.sh
#!/bin/bash
# install.sh — Install Grok Swarm from ClawHub skill
# Run this after: clawhub install grok-swarm

set -e

SCRIPT_DIR="$(cd "$(dirname "BASH_SOURCE[0]")" && pwd)"
OPENCLAW_HOME="-$HOME/.openclaw"

RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'

log() { echo -e "GREEN[+]NC $1"; }
warn() { echo -e "YELLOW[!]NC $1"; }
error() { echo -e "RED[✗]NC $1" >&2; }

echo "=========================================="
echo "  Grok Swarm Installer (from ClawHub)"
echo "=========================================="
echo

# Check prerequisites
if ! command -v python3 &> /dev/null; then
    error "Python 3 not found. Please install Python 3.8+"
    exit 1
fi

# Install plugin
log "Installing OpenClaw plugin..."
PLUGIN_DIR="$OPENCLAW_HOME/extensions/grok-swarm"
mkdir -p "$PLUGIN_DIR"
if [ -f "$SCRIPT_DIR/openclaw.plugin.json" ]; then
    cp "$SCRIPT_DIR/openclaw.plugin.json" "$PLUGIN_DIR/"
    log "Plugin manifest installed"
else
    warn "No plugin manifest found, skipping"
fi

# Install bridge skill
log "Installing Grok Swarm bridge..."
SKILL_DIR="$OPENCLAW_HOME/skills/grok-refactor"
mkdir -p "$SKILL_DIR"

if [ -d "$SCRIPT_DIR/bridge" ]; then
    cp -r "$SCRIPT_DIR/bridge/"* "$SKILL_DIR/"
    log "Bridge files installed"
else
    error "Bridge files not found in skill!"
    exit 1
fi

# Set up Python venv
log "Setting up Python environment..."
if [ ! -d "$SKILL_DIR/.venv" ]; then
    python3 -m venv "$SKILL_DIR/.venv"
fi
"$SKILL_DIR/.venv/bin/pip" install -q openai>=1.0.0
log "Python packages installed"

# Create config directory
log "Setting up API key configuration..."
CONFIG_DIR="$HOME/.config/grok-swarm"
mkdir -p "$CONFIG_DIR"
if [ ! -f "$CONFIG_DIR/config.json" ]; then
    echo '{"api_key": ""}' > "$CONFIG_DIR/config.json"
    chmod 600 "$CONFIG_DIR/config.json"
    warn "Created $CONFIG_DIR/config.json — add your OpenRouter API key"
fi

echo
log "Installation complete!"
echo
echo "Next steps:"
echo "  1. Add your API key to $CONFIG_DIR/config.json"
echo "     Or set: export OPENROUTER_API_KEY=sk-or-v1-..."
echo "  2. Update ~/.openclaw/openclaw.json to enable the plugin:"
echo "     - Add 'grok-swarm' to plugins.allow"
echo "     - Add 'grok_swarm' to agent tools.allow"
echo "  3. Restart OpenClaw: openclaw gateway restart"
echo
echo "Usage:"
echo "  node $SKILL_DIR/index.js --prompt 'Your task' --mode analyze"

FILE:openclaw.plugin.json
{
  "id": "grok-swarm",
  "name": "Grok Multi-Agent Swarm",
  "description": "Bridge to xAI Grok 4.20 Multi-Agent Beta (4-agent swarm) for codebase analysis, refactoring, reasoning, and code generation.",
  "version": "1.0.0",
  "configSchema": {
    "type": "object",
    "additionalProperties": false,
    "properties": {
      "defaultTimeout": {
        "type": "number",
        "description": "Default timeout in seconds (default: 120)"
      },
      "pythonPath": {
        "type": "string",
        "description": "Path to Python 3 binary"
      },
      "bridgeScript": {
        "type": "string",
        "description": "Path to grok_bridge.py script"
      },
      "defaultOutputDir": {
        "type": "string",
        "description": "Default output directory for file writes (default: ./grok-output/)"
      }
    }
  }
}