REST API Reference

The REST API exposes tasks and document pipelines via HTTP.

Starting the Server

# app.py
from redis_agent_kit.api import create_app

app = create_app(redis_url="redis://localhost:6379")

uvicorn app:app --reload
# Docs at http://localhost:8000/docs

Customization

app = create_app(
    redis_url="redis://localhost:6379",
    title="My Agent API",
    docs_url="/api/docs",
)

Endpoints

Health

GET /        → {"name": "Redis Agent Kit API", "version": "...", "docs": "/docs"}
GET /health  → {"status": "healthy"}

Tasks

List tasks:

GET /tasks
GET /tasks?status=in_progress&limit=10

Response:

{
  "tasks": [
    {
      "task_id": "01HXYZ...",
      "session_id": "01HABC...",
      "status": "in_progress",
      "metadata": {"message": "..."}
    }
  ],
  "total": 1
}

Get task:

GET /tasks/{task_id}

Response:

{
  "task_id": "01HXYZ...",
  "session_id": "01HABC...",
  "status": "done",
  "metadata": {"message": "What is Redis?"},
  "result": {"answer": "..."},
  "updates": [{"message": "Searching...", "update_type": "info"}]
}

Delete task:

DELETE /tasks/{task_id}

Cancel task (optionally cascade to descendants):

POST /tasks/{task_id}/cancel
POST /tasks/{task_id}/cancel?cascade=true

When cascade is omitted, the kit's SubTaskConfig.cascade_cancel is used. The response includes cancelled_count — the number of tasks transitioned to CANCELLED.

List children (sub-tasking):

GET /tasks/{task_id}/children

Returns the direct children of a task in spawn order, with each child's status, result, and spawn_key.

Get task tree (sub-tasking):

GET /tasks/{task_id}/tree
GET /tasks/{task_id}/tree?max_depth=5

Returns the recursive descendant tree:

{
  "task": {"task_id": "...", "status": "done", ...},
  "children": [
    {"task": {...}, "children": [...]}
  ]
}

See the Sub-Tasks guide for the data model and lifecycle.

Streaming

Server-Sent Events (SSE) endpoints push task progress and LLM tokens to clients in real time. Enable streaming by passing stream_config=StreamConfig(enabled=True) to both AgentKit and create_app.

from redis_agent_kit import AgentKit, ChannelScope, StreamConfig
from redis_agent_kit.api import create_app

stream_config = StreamConfig(
    enabled=True,
    channels={ChannelScope.TASK, ChannelScope.SESSION},
)
kit = AgentKit(agent_callable=my_handler, stream_config=stream_config)
app = create_app(kit=kit, stream_config=stream_config)

Stream a task:

GET /tasks/{task_id}/stream
GET /tasks/{task_id}/stream?replay=false
GET /tasks/{task_id}/stream?events=token,done,failed

Emits buffered updates on connect (?replay=true, default) and then new events as they happen. Returns 404 if streaming is not enabled.

Stream all tasks in a session (requires ChannelScope.SESSION):

GET /sessions/{session_id}/stream
GET /sessions/{session_id}/stream?events=token,done

Stream all events globally (requires ChannelScope.GLOBAL, useful for dashboards):

GET /stream/global
GET /stream/global?events=done,failed

Event types: update, token, done, failed, cancelled, input_required. When sub-tasking is enabled, the parent's channel also receives subtask_spawned, subtask_done, subtask_failed, and subtask_input_required events (see the Sub-Tasks guide).

JavaScript client:

const es = new EventSource(`/tasks/${taskId}/stream`);
es.addEventListener('update', (e) => console.log(JSON.parse(e.data).message));
es.addEventListener('token',  (e) => process.stdout.write(JSON.parse(e.data).message));
es.addEventListener('done',   (e) => { console.log(JSON.parse(e.data).result); es.close(); });
es.addEventListener('failed', (e) => { console.error(JSON.parse(e.data).error); es.close(); });

See the Streaming guide for the full configuration reference.

Pipelines

prepare, ingest, full, and documents submit background tasks and return task IDs.

Prepare (Stage 1):

POST /pipelines/prepare
{"source_path": "./docs"}

Response:

{"task_id": "01HXYZ..."}

Ingest (Stage 2):

POST /pipelines/ingest
{"batch_id": "01HXYZ..."}

Response:

{"task_id": "01HXYZ..."}

Full pipeline:

POST /pipelines/full
{"source_path": "./docs"}

Add document:

POST /pipelines/documents
{"filename": "my-doc.md", "content": "..."}

Run on inline documents:

POST /pipelines/run
{
  "documents": [{"title": "Doc 1", "content": "...", "source": "api"}],
  "chunk_size": 500
}

Status:

GET /pipelines/status

Response:

{"chunks": 142}

Clear:

DELETE /pipelines

Example Client

import httpx

async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
    # List tasks
    response = await client.get("/tasks", params={"status": "done"})
    tasks = response.json()["tasks"]

    # Get pipeline status
    response = await client.get("/pipelines/status")
    print(f"Chunks: {response.json()['chunks']}")

Error Responses

{"detail": "Task not found"}

Status codes: - 200 – Success - 201 – Created - 404 – Not found - 422 – Validation error - 500 – Server error