arccrew 0.13.0

Framework for building multi-agent LangGraph pipelines with Claude Code skills

ArcCrew

Multi-agent LangGraph pipelines — scaffold, build, and ship in minutes.

ArcCrew is a Python framework for building production-ready multi-agent pipelines on LangGraph. It ships with a CLI, a 3-layer prompt system, a FastAPI server, an MCP server, and AI coding skills that let you generate entire pipelines from a description.

---

Install

pip install arccrew

Optional dependencies

Extra	Installs	When you need it
arccrew[production]	Postgres + Redis drivers	Any server/VPS deployment — included in the generated requirements.txt
arccrew[postgres]	psycopg2-binary	Postgres run store (API_RUN_STORE=postgres)
arccrew[redis]	redis	Redis event bus (API_PUBSUB_BACKEND=redis)
arccrew[supabase]	supabase	Supabase Realtime event bus (API_PUBSUB_BACKEND=supabase_realtime)
arccrew[mcp]	mcp, langchain-mcp-adapters	Connecting MCP servers as agent tools or exposing pipelines via MCP
arccrew[sheets]	gspread, google-auth	Google Sheets read/write tools
arccrew[otel]	OpenTelemetry SDK	Tracing via Grafana, Datadog, Jaeger, etc.
arccrew[all]	Everything above	Development, CI, or when in doubt

Projects scaffolded with arccrew init already use arccrew[production] in requirements.txt. Run arccrew sync on existing projects to migrate automatically.

Quick start

# 1. Scaffold a new project
arccrew init my-project
cd my-project

# 2. Set your API key
cp .env.example .env
# edit .env → set ANTHROPIC_API_KEY

# 3. Verify everything is configured
arccrew check

# 4. Open in Claude Code and run:
# /build-agents  ← describe your pipeline, get all files generated

Or build manually:

# agents/researcher.py
from arccrew import BaseAgent, track_timing
from arccrew.tools import get_research_tools
from langgraph.types import Command
from pathlib import Path

class ResearcherAgent(BaseAgent):
    def __init__(self):
        super().__init__(name="researcher", prompts_dir=Path("prompts"))

    @property
    def system_prompt(self) -> str:
        return self.get_prompt_manager().assemble_prompt("researcher")

    @track_timing
    async def execute(self, state: dict) -> Command:
        task = state["tasks"][state["current_task_index"]]["description"]
        result = await self.run_react(task=task, tools=get_research_tools())
        return Command(goto="writer", update={"context": self.extract_json(result)})

# pipeline.py
from arccrew import create_pipeline, PipelineState
from arccrew.api.deps import pipeline_registry
from arccrew.mcp_server import register_pipeline
from agents.researcher import ResearcherAgent
from agents.writer import WriterAgent

def create_my_pipeline():
    researcher = ResearcherAgent()
    writer = WriterAgent()

    # Pure Python node — no LLM needed for deterministic transformation
    async def format_node(state: dict) -> dict:
        ctx = state.get("context", {})
        return {"context": {**ctx, "findings": ctx.get("findings", [])[:5]}}

    return create_pipeline(
        state_class=PipelineState,
        nodes={
            "researcher": lambda s: researcher.execute(s),  # LLM agent
            "format": format_node,                          # pure Python
            "writer": lambda s: writer.execute(s),          # LLM agent
        },
        flow=["researcher", "format", "writer"],
    )

pipeline_registry.register("my_pipeline", create_my_pipeline)
# Optional: reject bad input before a run is created (returns 422 immediately)
# pipeline_registry.register("my_pipeline", create_my_pipeline, validator=validate_my_input)
register_pipeline("my_pipeline", create_my_pipeline)

arccrew serve       # REST API on :8000
arccrew serve-mcp   # MCP server for Claude Desktop and other MCP clients

Once the server is running, open http://localhost:8000/playground to explore your pipelines interactively, or call them via API:

# Submit a run — returns 202 immediately with a run_id
curl -X POST http://localhost:8000/api/runs \
  -H "Content-Type: application/json" \
  -d '{"pipeline": "my_pipeline", "input": "write a blog post about AI"}'

# → {"run_id": "abc-123", "status": "running", ...}

# Poll until done
curl http://localhost:8000/api/runs/abc-123

# Stream events in real time (SSE)
curl -N http://localhost:8000/api/runs/abc-123/stream

# Fire-and-forget — get notified via webhook when done
curl -X POST http://localhost:8000/api/runs \
  -H "Content-Type: application/json" \
  -d '{"pipeline": "my_pipeline", "input": "...", "options": {"callback_url": "https://yourapp.com/webhook"}}'

# Cancel a running pipeline
curl -X POST http://localhost:8000/api/runs/abc-123/cancel

# List recent runs (filter by status, limit results)
curl "http://localhost:8000/api/runs?status=running&limit=20"

---

API integration

The REST API is designed for async workloads: submit a run and receive a run_id immediately (HTTP 202). Results arrive through one of three patterns depending on your use case.

Pattern 1 — Poll for results

Best for: simple clients, scripts, internal tooling.

POST /api/runs   →  202 + run_id
GET  /api/runs/{run_id}   (repeat until status != "running")

import httpx, time

BASE = "http://localhost:8000"

resp = httpx.post(f"{BASE}/api/runs", json={"pipeline": "default", "input": "your task"})
run_id = resp.json()["run_id"]

while True:
    data = httpx.get(f"{BASE}/api/runs/{run_id}").json()
    if data["status"] != "running":
        break
    time.sleep(3)

print(data["status"], data["results"])

Pattern 2 — Real-time SSE streaming

Best for: frontends and dashboards that need to show progress as agents finish.

POST /api/runs      →  202 + run_id
GET  /api/runs/{run_id}/stream   (event stream, closes on pipeline_completed / pipeline_failed)

import httpx

BASE = "http://localhost:8000"

run_id = httpx.post(f"{BASE}/api/runs", json={"pipeline": "default", "input": "your task"}).json()["run_id"]

with httpx.stream("GET", f"{BASE}/api/runs/{run_id}/stream") as r:
    for line in r.iter_lines():
        if line.startswith("data:"):
            print(line[5:].strip())  # JSON StreamEvent

Each event is a JSON StreamEvent:

{"event": "agent_completed", "run_id": "...", "agent": "researcher", "data": {...}}
{"event": "pipeline_completed", "run_id": "...", "data": {"status": "completed", "results": [...]}}

Pattern 3 — Fire-and-forget with webhook (recommended for production)

Best for: external services that submit jobs and want to be notified when done without holding a connection open.

POST /api/runs  {callback_url}  →  202 + run_id   (your service is free to do other work)
                                    ↓  pipeline runs async
POST callback_url  {run_id, status, results}        (arccrew notifies you when done)

import httpx

resp = httpx.post("http://localhost:8000/api/runs", json={
    "pipeline": "default",
    "input": "your task",
    "options": {
        "callback_url": "https://yourapp.com/webhook",  # arccrew POSTs here when done
        "timeout": 300,                                  # optional: cancel after 5 min
    },
})
run_id = resp.json()["run_id"]
# → your service can do other work; the webhook will arrive when the pipeline finishes

Webhook payload (POSTed to your callback_url):

{
  "run_id": "abc-123",
  "status": "completed",
  "elapsed_seconds": 42.1,
  "results": [{"summary": "...", "files_created": [...]}]
}

See examples/webhook_demo.py for a self-contained runnable demo (minimal HTTP receiver included, stdlib only).

Cancelling a run

POST /api/runs/{run_id}/cancel

Returns 200 immediately — cancellation is async. Poll GET /api/runs/{run_id} until status == "cancelled". Returns 409 if the run is already finished.

Run timeouts

Set a per-run timeout (seconds) via options.timeout. Runs that exceed the limit are cancelled and marked failed:

{"pipeline": "default", "input": "...", "options": {"timeout": 120}}

Set a server-wide default in .env:

API_RUN_TIMEOUT=300   # 5 minutes; 0 = no limit (default)

Run store backends

Backend	API_RUN_STORE	Best for
SQLite (default)	sqlite	single-instance deployments, persistent history, zero config
PostgreSQL	postgres	multi-instance production deployments, shared persistent storage
In-memory	memory	stateless deployments, testing, ephemeral containers

SQLite is the default — run history survives server restarts automatically, and runs.db is excluded from version control by default. Runs accumulate forever; set API_RUN_RETENTION to evict old terminal runs automatically.

# .env — SQLite (default, no config needed)
API_RUN_STORE_PATH=runs.db   # path to the SQLite file (default: runs.db)
API_RUN_RETENTION=0          # seconds to keep completed/failed runs (0 = keep forever)

# PostgreSQL — shared persistent storage for multi-instance deployments
# Requires: pip install "arccrew[postgres]"
API_RUN_STORE=postgres
API_RUN_STORE_URL=postgresql://user:password@host:5432/dbname

# In-memory — no disk writes, history lost on restart
API_RUN_STORE=memory

Event bus (SSE streaming)

Controls how real-time agent events are delivered to playground and API clients.

Backend	API_PUBSUB_BACKEND	Best for
Memory (default)	memory	single-instance deployments, zero config
Redis	redis	multi-instance deployments, multiple API workers
Supabase Realtime	supabase_realtime	browser clients subscribing directly

# Memory (default) — asyncio queue, no extra deps
# API_PUBSUB_BACKEND=memory

# Redis — pub/sub across multiple workers or servers
# Requires: pip install "arccrew[redis]"
API_PUBSUB_BACKEND=redis
API_REDIS_URL=redis://localhost:6379

# Supabase Realtime — browsers subscribe directly to Supabase channels
# Requires: pip install "arccrew[supabase]"
API_PUBSUB_BACKEND=supabase_realtime
SUPABASE_URL=https://<project>.supabase.co
SUPABASE_SERVICE_ROLE_KEY=<service-role-key>

Extending arccrew — custom event bus backends

You can add your own event bus backend in two ways.

Option 1 — register at runtime:

from arccrew.api.event_bus import register_bus_backend

register_bus_backend("my_bus", lambda: MyEventBus(url="..."))

Then set API_PUBSUB_BACKEND=my_bus and arccrew will call your factory on startup.

Option 2 — package entry point (distributable plugins):

# In your package's pyproject.toml
[project.entry-points."arccrew.event_bus"]
my_bus = "mypackage.bus:MyEventBus"

MyEventBus is instantiated with no arguments, so all configuration should come from environment variables or constructor defaults.

Fan-out to multiple backends with CompositeEventBus:

from arccrew.api.event_bus import CompositeEventBus, MemoryEventBus
from arccrew.extras.supabase import SupabaseRealtimeEventBus

bus = CompositeEventBus(
    MemoryEventBus(),                                      # SSE via ArcCrew API
    SupabaseRealtimeEventBus(url=..., key=...),            # browser direct subscribe
)

publish and close are sent to all backends concurrently. A single failing backend never blocks the others. subscribe reads from the first (primary) bus.

Docker (production)

arccrew init generates a Dockerfile and docker-compose.yml in your project directory. To start Postgres + Redis + your pipeline:

# Add to .env: POSTGRES_PASSWORD=... and PROJECT_NAME=...
docker compose up -d

Scaling arccrew

arccrew supports horizontal scaling across multiple workers or containers. The key requirement is a shared run store (Postgres) and a shared event bus (Redis) so every worker can see every run.

Environment variables

Variable	Default	Description
API_MAX_CONCURRENT_RUNS	0 (unlimited)	Max concurrent pipeline runs per worker. Returns 503 when full.
API_MAX_QUEUE_DEPTH	50	Max requests that can be waiting when the limit is reached.
API_ORPHAN_THRESHOLD_SECONDS	1800	Seconds before a stuck RUNNING run is reaped as FAILED.
API_RUN_STORE	sqlite	Set to postgres for shared persistent state across workers.
API_PUBSUB_BACKEND	memory	Set to redis to enable SSE and cross-worker cancel.
API_REDIS_URL	—	Redis connection URL, e.g. redis://localhost:6379.

Cross-worker cancel

When API_PUBSUB_BACKEND=redis, cancelling a run via POST /api/runs/{id}/cancel publishes a signal to Redis channel arccrew:cancel:{run_id}. Every worker subscribes to this pattern at startup and cancels its local asyncio task if it owns that run. This means cancel works correctly regardless of which worker receives the HTTP request.

Multi-worker Docker deployment

docker-compose.prod.yml (included in the project root) runs 3 replicas with Postgres + Redis:

# Set POSTGRES_PASSWORD and API_SECRET_KEY in .env, then:
docker compose -f docker-compose.prod.yml up -d

# Scale to 5 replicas at runtime:
docker compose -f docker-compose.prod.yml up -d --scale arccrew=5

Key design decisions:

• 1 uvicorn worker per container — Docker handles replicas; multiple uvicorn workers per container share asyncio state unpredictably.
• API_MAX_CONCURRENT_RUNS — per-container back-pressure. With 3 replicas × 10 slots = 30 concurrent pipelines total.
• Redis cancel — required for multi-replica deployments so any replica can cancel any run.

Local multi-worker test

# Terminal 1: infrastructure
docker compose up redis db

# Terminal 2: worker on port 8001
API_RUN_STORE=postgres API_RUN_STORE_URL=postgresql://arccrew:changeme@localhost:5432/arccrew \
API_PUBSUB_BACKEND=redis API_REDIS_URL=redis://localhost:6379 \
API_MAX_CONCURRENT_RUNS=2 API_PORT=8001 arccrew serve

# Terminal 3: worker on port 8002
API_RUN_STORE=postgres API_RUN_STORE_URL=postgresql://arccrew:changeme@localhost:5432/arccrew \
API_PUBSUB_BACKEND=redis API_REDIS_URL=redis://localhost:6379 \
API_MAX_CONCURRENT_RUNS=2 API_PORT=8002 arccrew serve

# Test cross-worker cancel: start run on worker 1, cancel via worker 2
RUN_ID=$(curl -s -X POST localhost:8001/api/runs -H "Content-Type: application/json" \
  -d '{"pipeline":"my_pipeline","input":{}}' | jq -r .run_id)
curl -X POST localhost:8002/api/runs/$RUN_ID/cancel

Authentication

Auth is disabled by default. Enable for production:

API_AUTH_ENABLED=true
API_SECRET_KEY=<run: python -c "import secrets; print(secrets.token_hex(32))">
API_USERS=alice:s3cr3t,bob:hunter2   # comma-separated user:password pairs

Get a token, then pass it as a Bearer header:

curl -X POST http://localhost:8000/api/auth/token \
  -H "Content-Type: application/json" \
  -d '{"username": "alice", "password": "s3cr3t"}'
# → {"access_token": "eyJ..."}

curl http://localhost:8000/api/runs \
  -H "Authorization: Bearer eyJ..."

WebSocket auth — connect, then send the token as the first message:

ws://localhost:8000/api/ws
→ {"type": "auth", "token": "eyJ..."}
← {"type": "auth_ok", ...}

(Passing the token as a query param — /api/ws?token=eyJ... — still works but is deprecated: URLs end up in reverse-proxy access logs.)

---

Features

• 3-layer prompt system — library base + your project globals + per-agent role
• AI coding skills — pre-installed in every scaffolded project, work natively in Claude Code
• Built-in tools — web search (Tavily or DDG Lite), file management, shell execution, MCP server adapters, ready to plug into any agent
• FastAPI server — non-blocking REST API (POST /api/runs → 202 + run_id, poll with GET /api/runs/{id}), SSE streaming, WebSocket, interactive playground at /playground
• MCP dual role — expose your pipelines as MCP tools AND connect external MCP servers (GitHub, Notion, Slack…) as agent tools
• Multi-provider — Claude by default; OpenAI, Gemini, Groq, OpenRouter (200+ models), and Ollama (local) via env var
• Supervisor pattern — LLM-driven routing as an alternative to manual graph wiring
• Retry / verification loops — built-in worker → verifier → retry pattern

---

Prompt layers

Every agent's system prompt is assembled in this order:

Layer	File	Who controls it
Base	bundled in arccrew	library — universal agent rules
Global	prompts/global.md	you — project-wide rules (tone, domain, language)
Agent	prompts/{agent}.md	you — role, tools, output schema

---

Built-in tools

Every agent has access to these tools out of the box:

from arccrew.tools import get_research_tools, create_workspace_tools, get_mcp_tools

# Research tools — web_search
# Uses Tavily if TAVILY_API_KEY is set (recommended), DDG Lite otherwise (free, no key)
tools = get_research_tools()

# Workspace tools — write_file, read_file, list_files, run_shell
from pathlib import Path
tools = create_workspace_tools(Path("workspace"))

# MCP tools — connect any MCP server (GitHub, Notion, Slack, Google, custom…)
# Requires: pip install "arccrew[mcp]"
# Configure servers once in arccrew.json, then call from any agent:
tools = await get_mcp_tools()              # all servers in arccrew.json
tools = await get_mcp_tools(["github"])    # specific servers only

arccrew.json (project root) — define servers once, use everywhere:

{
  "mcpServers": {
    "github": {
      "transport": "sse",
      "url": "https://api.githubcopilot.com/mcp/",
      "headers": { "Authorization": "Bearer ${GITHUB_TOKEN}" }
    }
  }
}

${ENV_VAR} is interpolated from .env at runtime — API keys never go in arccrew.json.

Combine freely in any agent:

result = await self.run_react(
    task=task,
    tools=(
        get_research_tools()
        + create_workspace_tools(Path("workspace"))
        + await get_mcp_tools()
    )
)

---

Adding your own tools

Create a file in tools/ named after your domain:

# tools/calendar_tools.py
from langchain_core.tools import tool

@tool
async def get_availability(date: str) -> str:
    """Check calendar availability for a given date (YYYY-MM-DD).

    Use for: tasks that require checking free/busy slots.
    Do NOT use for: general research (use web_search instead).

    Args:
        date: Date to check in YYYY-MM-DD format.

    Returns:
        Available time slots as a formatted string.
    """
    try:
        # your implementation here
        return f"Available slots for {date}: 9am, 2pm, 4pm"
    except Exception as e:
        return f"ERROR: {e}"

def get_calendar_tools() -> list:
    return [get_availability]

Combine with arccrew built-ins in any agent:

from arccrew.tools import get_research_tools
from tools.calendar_tools import get_calendar_tools

result = await self.run_react(
    task=task,
    tools=get_research_tools() + get_calendar_tools(),
)

Use /add-tool in Claude Code to generate a new tool from a description.

---

Environment variables

All configuration lives in .env. Copy .env.example after scaffolding:

# LLM provider (pick one or mix via per-agent overrides)
ANTHROPIC_API_KEY=your_key_here
OPENAI_API_KEY=your_key_here
GROQ_API_KEY=your_key_here
GOOGLE_CLOUD_PROJECT=my-gcp-project  # Google Vertex AI (auth via ADC or service account)
OPENROUTER_API_KEY=sk-or-v1-...      # 200+ models under one key (optional)

# Model (default for all agents)
AGENT_MODEL=anthropic/claude-haiku-4-5-20251001

# OpenRouter — access any model with one key (free tier available)
# AGENT_MODEL=openrouter/meta-llama/llama-3.1-8b-instruct:free

# Ollama — local models, no cost, no data leaves machine
# AGENT_MODEL=ollama/llama3.2   (requires: ollama serve + ollama pull llama3.2)

# Per-agent overrides — mix cheap and expensive models per agent
RESEARCHER_MODEL=anthropic/claude-sonnet-4-6
RESEARCHER_MAX_ROUNDS=10
WRITER_MAX_ROUNDS=5

# Web search backend (optional — DDG Lite is free and works without a key)
TAVILY_API_KEY=tvly-...   # recommended for production (1k free req/month)

# Workspace (where agents write files)
WORKSPACE_DIR=workspace

# API server
API_HOST=0.0.0.0
API_PORT=8000
API_AUTH_ENABLED=false   # set true + API_SECRET_KEY for production

# Observability (optional)
LANGSMITH_API_KEY=your_key_here
LANGSMITH_PROJECT=my-project
LANGSMITH_TRACING=true

---

Skills

Every project created with arccrew init gets skills pre-installed in .claude/commands/ and detailed references in skills/. They work natively in Claude Code as slash commands.

After upgrading arccrew, run arccrew sync to get new and updated skills without touching your project files.

Core

Skill	What it does
/build-agents	Generate a full pipeline from a description
/add-agent	Add a single agent to an existing pipeline
/add-tool	Add a tool to an agent
/add-state-field	Add a custom state field with the right reducer
/add-prompt	Add or update an agent prompt

Patterns

Skill	What it does
/add-retry-loop	Add retry + verification loop
/add-checkpointing	Enable durable execution (crash recovery + resume)
/add-review-gate	Add a human-in-the-loop review gate
/add-supervisor	Add LLM-driven supervisor orchestration

Infrastructure

Skill	What it does
/add-api-endpoint	Add a REST endpoint to the FastAPI server
/add-mcp-pipeline	Register a pipeline as an MCP tool
/add-google-sheets	Add Google Sheets read/write tools (OAuth2 auth, requires pip install "arccrew[sheets]")

Configuration

Skill	What it does
/configure-claude	Configure Claude as LLM provider
/configure-openai	Configure OpenAI as LLM provider
/configure-gemini	Configure Gemini as LLM provider
/configure-openrouter	Configure OpenRouter — 200+ models, free tier available
/configure-ollama	Configure Ollama for local models (no API key, no cost)
/switch-provider	Switch between LLM providers
/configure-mcp	Connect external MCP servers as agent tools

Quality

Skill	What it does
/enable-langsmith	Set up LangSmith tracing
/enable-otel	Set up OpenTelemetry (Grafana, Datadog, Jaeger…)
/debug-pipeline	Diagnose pipeline errors
/write-tests	Generate tests for agents and tools

---

CLI

arccrew init <name>                    # scaffold a new project
arccrew init <name> --dir .           # scaffold into the current directory (empty or dotfiles only)
arccrew init <name> --dir . --force   # scaffold into an existing directory (skips existing files)
arccrew check                          # verify config and dependencies
arccrew sync                           # update skills after upgrading
arccrew run "describe your task"       # run a pipeline from the terminal
arccrew run "task" --pipeline my_name  # run a specific pipeline
arccrew visualize                      # print Mermaid diagram of your graph
arccrew visualize -o graph.png         # save diagram as PNG (requires pyppeteer)
arccrew visualize -o graph.mmd         # save raw Mermaid code
arccrew serve                          # start FastAPI server
arccrew serve-mcp                      # start MCP server (stdio)

---

Local development

To work on arccrew itself and test changes immediately:

git clone https://github.com/amonrreal/arccrew
cd arccrew

# Install in editable mode — changes take effect without reinstalling
pip install -e ".[dev]"
cp .env.example .env
# Set ANTHROPIC_API_KEY (or another provider) in .env

# Verify the CLI uses your local version
arccrew --help

# Run the built-in example
python -m examples.researcher_writer.pipeline
python -m examples.researcher_writer.pipeline --supervisor

# Scaffold a test project and try the new commands
cd /tmp
arccrew init test-project
cd test-project
cp .env.example .env
# edit .env → set API key

arccrew check
# → now requires a pipeline.py in the current directory ✓

# Run tests
cd /path/to/arccrew
pytest
pytest tests/test_base_agent.py -v
pytest -k "test_name"

---

MCP server

Expose your pipelines as tools in any MCP-compatible client (Claude Desktop, Cursor, Claude Code):

arccrew serve-mcp   # local stdio transport
arccrew serve       # also serves /mcp for remote HTTP transport

Connecting clients

Local (Claude Desktop / Cursor) — add to your MCP config:

{
  "mcpServers": {
    "my-project": {
      "command": "arccrew",
      "args": ["serve-mcp"],
      "cwd": "/path/to/your/project"
    }
  }
}

Remote (after deploying):

{
  "mcpServers": {
    "my-project": {
      "url": "https://your-deployment-url/mcp"
    }
  }
}

Available tools

Once connected, the MCP client sees these tools automatically:

Tool	Description
run_pipeline	Run any registered pipeline by name
run_{name}	Shortcut tool per pipeline (e.g. run_default)
list_pipelines	List all available pipeline names

Calling a pipeline

Both input (string shorthand) and tasks (array) are accepted — same ergonomics as the REST API:

# Single task — plain string
run_default(input="Research the top 5 open-source LLM frameworks and summarize them")

# Multiple tasks — explicit list
run_pipeline(pipeline="default", tasks=[
  {"description": "Research open-source LLM frameworks"},
  {"description": "Write a comparison report"}
])

Progress notifications

Pipelines stream progress in real time — the client sees agent-by-agent updates instead of waiting silently for the full result:

Starting pipeline 'default' (2 task(s))
Agent 'researcher' running…
Agent 'researcher' completed (1/2 task(s) done)
Agent 'writer' running…
Agent 'writer' completed (2/2 task(s) done)

Progress is sent automatically when the client supplies a progressToken (Claude Desktop and Cursor do this by default). No configuration needed.

---

Utility helpers

from arccrew.utils.helpers import truncate_text, extract_json_safe, slugify

# Truncate long LLM output before storing
short = truncate_text(long_response, max_chars=4000)

# Safely extract JSON from any LLM response
data = extract_json_safe(response, fallback={})

# Generate URL-safe slugs
slug = slugify("My Agent Result — 2026")  # "my-agent-result-2026"

---

Examples

Researcher + Writer pipeline

examples/researcher_writer/ — complete pipeline with two agents (Researcher + Writer) in both manual graph and supervisor patterns.

python -m examples.researcher_writer.pipeline            # manual graph
python -m examples.researcher_writer.pipeline --supervisor  # supervisor pattern

Webhook integration demo

examples/webhook_demo.py — end-to-end demo of the fire-and-forget pattern: submit a job with a callback_url, receive results via webhook when the pipeline finishes. Includes a minimal HTTP receiver (stdlib only, no extra deps).

# Terminal 1 — start the API
arccrew serve

# Terminal 2 — run the demo
python -m examples.webhook_demo

---

License

ArcCrew is licensed under the MIT License.