Real-time Streaming

RAK can push task progress and LLM tokens to clients in real time over Server-Sent Events (SSE), backed by Redis Pub/Sub. Streaming is off by default and turned on with a single StreamConfig object.

Enabling streaming

Pass a StreamConfig to both AgentKit (so the worker publishes) and create_app (so the API exposes the SSE endpoints):

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

stream_config = StreamConfig(enabled=True)

kit = AgentKit(agent_callable=my_handler, stream_config=stream_config)
app = create_app(kit=kit, stream_config=stream_config)

With this minimal setup, GET /tasks/{task_id}/stream is live.

Channel scopes

StreamConfig.channels controls which Pub/Sub channels an event fans out to. Each scope unlocks a corresponding SSE endpoint.

from redis_agent_kit import ChannelScope, StreamConfig

# Per-task only (default)
StreamConfig(enabled=True)

# Per-task + per-session (one EventSource per chat session)
StreamConfig(enabled=True, channels={ChannelScope.TASK, ChannelScope.SESSION})

# Per-task + per-session + global dashboard
StreamConfig(
    enabled=True,
    channels={ChannelScope.TASK, ChannelScope.SESSION, ChannelScope.GLOBAL},
)

Scope	Channel	Endpoint
TASK	rak:task:{task_id}:updates	GET /tasks/{task_id}/stream
SESSION	rak:session:{session_id}:updates	GET /sessions/{session_id}/stream
GLOBAL	rak:updates	GET /stream/global

Event types

Event	Meaning	Persisted?
update	Milestone progress message	Yes
token	A single LLM token	No (ephemeral)
done	Task completed, includes result	Yes
failed	Task failed, includes error	Yes
cancelled	Task was cancelled	Yes
input_required	Task is waiting for user input	Yes

Non-token events are written to the task's update history so clients that connect late can ?replay=true and catch up. Tokens are intentionally publish-only to keep token-by-token streaming fast.

Token streaming

Use ctx.emitter.emit_token(...) inside your handler to stream LLM tokens:

from openai import AsyncOpenAI

async def handler(ctx):
    client = AsyncOpenAI()
    await ctx.emitter.emit("Thinking...")

    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": ctx.message}],
        stream=True,
    )

    chunks: list[str] = []
    async for event in stream:
        delta = event.choices[0].delta.content
        if delta:
            chunks.append(delta)
            await ctx.emitter.emit_token(delta)

    return {"response": "".join(chunks)}

emit_token() publishes directly to Pub/Sub without touching the task object in Redis, so per-token overhead stays minimal.

Filtering events

All three endpoints accept an events= query parameter:

# Only tokens and terminal events
curl -N 'http://localhost:8000/tasks/{task_id}/stream?events=token,done,failed'

You can also filter at the source via StreamConfig.publish_events:

StreamConfig(
    enabled=True,
    publish_events={"token", "done", "failed", "cancelled"},  # suppress 'update'
)

Persistence policy

By default, every event except token is persisted to the task's update list. Override with persist_events:

# Persist only milestones and terminal statuses — tokens stay ephemeral
StreamConfig(
    enabled=True,
    persist_events={"update", "done", "failed", "cancelled"},
)

Consuming from JavaScript

const es = new EventSource(`/tasks/${taskId}/stream`);

es.addEventListener('update', (e) => {
    const data = JSON.parse(e.data);
    console.log(data.message);           // milestone progress
});
es.addEventListener('token', (e) => {
    const data = JSON.parse(e.data);
    process.stdout.write(data.message);  // individual LLM token
});
es.addEventListener('done', (e) => {
    const data = JSON.parse(e.data);
    console.log(data.result.response);   // final answer
    es.close();
});
es.addEventListener('failed', (e) => {
    console.error(JSON.parse(e.data).error);
    es.close();
});

The per-task endpoint supports ?replay=true (default) to emit buffered updates on connect so late subscribers don't miss progress that already happened.

Example

A complete runnable example — OpenAI token streaming wired through SSE — lives in examples/sse_streaming/agent.py.

Related

• Tasks — task lifecycle and the TaskEmitter API
• API Reference — endpoint reference and query parameters
• Middleware — EmitterMiddleware publishes through the same stream