@clawhub-khaentertainment-f72dc150bc
Refactors code using Grok 4.20 multi-agent swarm to improve readability, maintain behavior, add modern patterns, and explain changes.
# 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/)"
}
}
}
}