# Skill: Register a Pipeline with the API

Use when you want to expose your pipeline over HTTP — async fire & poll, SSE streaming,
or WebSocket.

## How the API Works

`arccrew/api/app.py` creates a FastAPI app with:

| Endpoint | What it does |
|---|---|
| `POST /api/runs` | Start a run — returns `run_id` **immediately** (202 Accepted) |
| `GET /api/runs/{run_id}` | Poll status + partial results while it runs |
| `GET /api/runs/{run_id}/stream` | Real-time SSE events per agent node |
| `WS /api/ws` | WebSocket — same as above, bidirectional |
| `GET /api/pipelines` | List all registered pipelines |
| `GET /playground` | Interactive browser UI (pipelines, events, code examples) |
| `POST /api/auth/token` | JWT token (only when `API_AUTH_ENABLED=true`) |

When `arccrew serve` starts, it auto-loads `pipeline.py` from the project root.
Any `pipeline_registry.register()` calls at module level are executed automatically.

## Registering Your Pipeline

At the bottom of your `pipeline.py`:

```python
from arccrew.api.deps import pipeline_registry

pipeline_registry.register("my_pipeline", create_my_pipeline)

# Multiple pipelines in the same project:
pipeline_registry.register("another_pipeline", create_another_pipeline)
```

Do NOT modify `arccrew/api/app.py` — it is a library file.

`pipeline_registry.register(name, factory_fn)` takes:
- `name`: string identifier used in API requests
- `factory_fn`: a callable that returns a compiled LangGraph — called once on first use

## `PipelineRequest` Schema

The request body for `POST /api/runs` accepts **three equivalent forms**:

### Simple string input (most common)

```json
{
  "pipeline": "my_pipeline",
  "input": "Research AI trends in healthcare and write a short report"
}
```

### Structured dict input (for form data or structured briefs)

```json
{
  "pipeline": "my_pipeline",
  "input": {
    "nombre": "Raíces Vivas",
    "industria": "Moda sostenible",
    "publico_objetivo": "Mujeres 25-40 años",
    "duracion_dias": 7
  }
}
```

The dict is JSON-encoded as `tasks[0].description` and also stored in `tasks[0].metadata`.
Your agent reads it from `state["tasks"][0]["description"]` — the JSON string — and can
call `self.extract_json()` to parse it back into a dict.

### Multi-task list (batch processing)

```json
{
  "pipeline": "my_pipeline",
  "tasks": [
    {"description": "Task one description"},
    {"description": "Task two description"}
  ]
}
```

The pipeline runs the full agent chain once per task. All results accumulate in `results[]`.

Optional `options`:
```json
{
  "pipeline": "my_pipeline",
  "input": "...",
  "options": {
    "dry_run": false,
    "workspace_dir": "/tmp/my_run"
  }
}
```

## The Async Pattern

`POST /api/runs` always returns immediately — the pipeline runs in the background.
Use one of two strategies to get results:

### Strategy A — Fire & Poll (simplest, works everywhere)

```bash
# 1. Start run
RUN_ID=$(curl -s -X POST http://localhost:8000/api/runs \
  -H "Content-Type: application/json" \
  -d '{"pipeline": "my_pipeline", "input": "Research AI trends"}' \
  | jq -r .run_id)

# 2. Poll until done
while true; do
  STATUS=$(curl -s http://localhost:8000/api/runs/$RUN_ID | jq -r .status)
  echo "Status: $STATUS"
  [ "$STATUS" = "completed" ] || [ "$STATUS" = "failed" ] && break
  sleep 3
done

# 3. Get results
curl -s http://localhost:8000/api/runs/$RUN_ID | jq .results
```

`GET /api/runs/{run_id}` response:
```json
{
  "run_id": "abc-123",
  "status": "completed",
  "pipeline": "my_pipeline",
  "task_count": 1,
  "results": [{"task_index": 0, "summary": "...", "files_created": [...]}],
  "started_at": "2026-04-08T10:00:00Z",
  "completed_at": "2026-04-08T10:00:12Z",
  "elapsed_seconds": 12.4
}
```

`status` is one of: `running` | `completed` | `failed`

Partial `results` are populated as each task finishes — useful for long multi-task runs.

### Strategy B — SSE Streaming (real-time progress)

```bash
# 1. Start run
RUN_ID=$(curl -s -X POST http://localhost:8000/api/runs \
  -H "Content-Type: application/json" \
  -d '{"pipeline": "my_pipeline", "input": "Research AI trends"}' \
  | jq -r .run_id)

# 2. Stream events
curl -N http://localhost:8000/api/runs/$RUN_ID/stream
```

Each SSE event is a JSON-encoded `StreamEvent`:

```
data: {"event": "agent_completed", "agent": "researcher", "run_id": "...", "data": {...}}
data: {"event": "agent_completed", "agent": "writer",     "run_id": "...", "data": {...}}
data: {"event": "pipeline_completed", "run_id": "...", "data": {"results": [...], "elapsed_seconds": 12.4}}
```

On failure:
```
data: {"event": "pipeline_failed", "run_id": "...", "data": {"error": "..."}}
```

### Python — fire & poll

```python
import httpx, time

BASE = "http://localhost:8000"

run = httpx.post(f"{BASE}/api/runs", json={
    "pipeline": "my_pipeline",
    "input": "Research AI trends in healthcare",
}).json()

run_id = run["run_id"]

while True:
    status = httpx.get(f"{BASE}/api/runs/{run_id}").json()
    if status["status"] in ("completed", "failed"):
        break
    time.sleep(2)

print(status["results"])
```

### Python — SSE streaming

```python
import httpx, json

BASE = "http://localhost:8000"

run = httpx.post(f"{BASE}/api/runs", json={
    "pipeline": "my_pipeline",
    "input": "Research AI trends",
}).json()

with httpx.stream("GET", f"{BASE}/api/runs/{run['run_id']}/stream") as r:
    for line in r.iter_lines():
        if line.startswith("data:"):
            event = json.loads(line[5:])
            print(event["event"], event.get("agent", ""))
            if event["event"] in ("pipeline_completed", "pipeline_failed"):
                break
```

## WebSocket

```bash
# Using wscat (npm install -g wscat)
wscat -c ws://localhost:8000/api/ws

# Send a run request
{"type": "run", "payload": {"pipeline": "my_pipeline", "input": "test"}}

# Cancel a run
{"type": "cancel", "run_id": "abc-123"}
```

WebSocket event types (server → client):
- `started` — pipeline beginning
- `agent_completed` — one node finished (payload includes partial state)
- `complete` — pipeline done (payload includes `results`)
- `cancelled` — run was cancelled
- `error` — unrecoverable error

## Viewing Registered Pipelines

```bash
curl http://localhost:8000/api/pipelines
# → {"pipelines": ["researcher_writer", "my_pipeline"]}
```

If your pipeline is not listed, the registration in `pipeline.py` did not run — check
for import errors in server logs on startup.

## Interactive Playground

```bash
arccrew serve --reload
open http://localhost:8000/playground
```

The playground lets you select a pipeline, send text or JSON input, and watch SSE
events live in the browser. Curl and Python code examples are generated automatically.

## Authentication (Optional)

Auth is disabled by default. To enable for production:

```bash
# .env
API_AUTH_ENABLED=true
API_SECRET_KEY=$(python3 -c "import secrets; print(secrets.token_hex(32))")
API_TOKEN_EXPIRE_MINUTES=1440
```

Get a token and use it:

```bash
curl -X POST http://localhost:8000/api/auth/token \
  -H "Content-Type: application/json" \
  -d '{"username": "admin", "password": "changeme"}'

curl -X POST http://localhost:8000/api/runs \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer eyJ..." \
  -d '{"pipeline": "my_pipeline", "input": "test"}'
```

## Rate Limiting

Default: 60 requests/minute per IP, burst of 10.

```bash
# .env
API_RATE_LIMIT=60
API_RATE_LIMIT_BURST=10
```

## Starting the API Server

```bash
arccrew serve --reload   # dev (hot reload)
arccrew serve            # production
arccrew serve --port 9000
```

Swagger UI: http://localhost:8000/docs
Playground: http://localhost:8000/playground

## CORS

```bash
# .env
API_CORS_ORIGINS=https://myapp.com,https://staging.myapp.com
API_CORS_ORIGINS=*   # dev only
```
