Server-Sent Events (SSE) - Prowler Documentation

This guide explains how to add a Server-Sent Events (SSE) endpoint to the Prowler API. SSE lets the backend push a one-way stream of events to a client over a single long-lived HTTP connection — ideal for live progress, token-by-token LLM output, or any “the server has news for you” use case where the client should not poll.

The platform ships the SSE infrastructure (api.sse) and wiring. No feature endpoint streams over SSE out of the box — this guide shows how to build one on top of the shared base.

When to use SSE

Need	Use
Server pushes incremental updates, client only reads	SSE
Bidirectional, low-latency messaging (chat both ways, games)	WebSocket
Client asks, server answers once	Plain REST

SSE is the right tool when the client only consumes: scan progress, long-running job checkpoints, streamed LLM tokens, cross-client resource-sync notifications. It rides on plain HTTP, reconnects automatically in the browser via the native EventSource API, and needs no extra protocol.

How it works

SSE is wired through django-eventstream and a small platform layer in api/src/backend/api/sse/:

Piece	File	Responsibility
`BaseSSEViewSet`	`api/sse/base_views.py`	Base DRF viewset a feature subclasses. The feature implements `get_channels`; the base handles auth, the tenant transaction, and delegates streaming to `django-eventstream`.
`SSEChannelManager`	`api/sse/channelmanager.py`	Registered in `settings.EVENTSTREAM_CHANNELMANAGER_CLASS`. Reads the channel set off the request and enforces the platform-wide tenant gate.
`SSEAuthentication`	`api/authentication.py`	Same JWT/API-key stack as the rest of the API, plus an `?access_token=<jwt>` fallback for browser `EventSource` clients. Lives with the other authentication classes, not in the `sse` package.
`make_channel_name` / `tenant_id_from_channel`	`api/sse/utils.py`	Single source of truth for the channel-name format, so publishers and the channel manager agree byte-for-byte.
Settings	`config/settings/eventstream.py`	Valkey Pub/Sub backend (dedicated DB), channel manager, allowed headers.

Transport: the server runs on ASGI

SSE connections are long-lived. Holding one open per synchronous worker would exhaust the worker pool, so the API runs under Gunicorn’s native asgi worker (config.asgi:application). Streams are parked on the event loop while ordinary CRUD endpoints keep their synchronous execution (Django runs sync views in a thread-sensitive executor under ASGI). This is configured in config/guniconf.py and used by both the dev and production entrypoints — no separate server process is needed.

The data flow

publisher (Celery task / view)             subscriber (browser, CLI)
        │                                          │
        │  send_event(channel, "scan.progress", …) │  GET …/event-stream
        ▼                                          ▼
   Valkey Pub/Sub  ◄────────────────────►  BaseSSEViewSet.list
   (EVENTSTREAM_VALKEY_DB)                  → get_channels() (RLS-scoped)
                                            → SSEChannelManager (tenant gate)
                                            → StreamingHttpResponse (text/event-stream)

A publisher anywhere in the system (most often a Celery task) calls send_event(channel, event_type, payload). django-eventstream fans it out over Valkey Pub/Sub to every connection subscribed to that channel.

Adding an SSE endpoint to your feature

The example below streams progress for a long-running scan. Adapt the resource, prefix, and event names to your feature.

Pick a channel prefix

Channels follow the format <prefix>:<tenant_id>:<resource_id>, built only through make_channel_name. The prefix is owned by your feature and may contain hyphens but never colons (the parser splits on :).

CHANNEL_PREFIX = "scan-progress"

The tenant id is baked into every channel name. That is what lets the platform enforce cross-tenant isolation without knowing anything about your feature.

Subclass BaseSSEViewSet

Create the viewset for the SSE sub-resource. The only required method is get_channels; it runs inside the tenant transaction set up by the base class, so any database lookup inside it is automatically RLS-scoped.

# scans/event_streams.py
from api.sse import BaseSSEViewSet, make_channel_name
from django.shortcuts import get_object_or_404
from scans.models import Scan

CHANNEL_PREFIX = "scan-progress"


class ScanEventStreamViewSet(BaseSSEViewSet):
    def get_queryset(self):
        # RLS already scopes to the tenant; narrow further as needed
        # (e.g. only scans the requesting user may see).
        return Scan.objects.filter(tenant_id=self.request.tenant_id)

    def get_channels(self) -> set[str]:
        scan = get_object_or_404(self.get_queryset(), pk=self.kwargs["scan_pk"])
        return {make_channel_name(CHANNEL_PREFIX, scan.tenant_id, scan.id)}

get_channels must raise the relevant DRF exception (NotFound, PermissionDenied, NotAuthenticated) when authorization fails — get_object_or_404 does this for you. Returning an empty set surfaces as django-eventstream’s confusing “No channels specified” error instead of the real cause.

Wire the URL as a sub-resource

Mount the endpoint as an event-stream sub-resource. Keep it outside the DRF router, which would force the URL into a list/detail convention. Route the get method to the viewset’s list action.

# scans/urls.py
path(
    "scans/<uuid:scan_pk>/event-stream",
    ScanEventStreamViewSet.as_view({"get": "list"}),
    name="scan-event-stream",
),

Define your event vocabulary

A feature owns its event types in <app>/<domain>/events.py: one publish_<event> function per event type, each body a single send_event call so the wire-level string lives in exactly one place.

# scans/events.py
from django_eventstream import send_event


def publish_progress(channel: str, checked: int, total: int) -> None:
    send_event(channel, "scan.progress", {"checked": checked, "total": total})


def publish_end(channel: str, scan_id: str) -> None:
    # Terminal event carries the canonical id so reconnecting clients
    # can refetch the persisted resource over REST.
    send_event(channel, "scan.end", {"scan_id": scan_id})


def publish_error(channel: str, code: str, detail: str) -> None:
    send_event(channel, "scan.error", {"code": code, "detail": detail})

There is no platform-side enum, registry, or dispatch table — the naming convention is the contract (see below).

Publish from the producer

Wherever the work happens — usually a Celery task — build the channel the same way and publish:

from api.sse import make_channel_name
from scans.events import publish_progress, publish_end

channel = make_channel_name("scan-progress", scan.tenant_id, scan.id)
publish_progress(channel, checked=42, total=100)
...
publish_end(channel, scan_id=str(scan.id))

Event naming convention

Every event uses an event type of the form <resource>.<verb> (lowercased, dot-separated). The verb comes from this platform-wide vocabulary — if you need a verb that is not listed, document the addition in this guide so the catalog stays discoverable.

Verb	When to use
`delta`	An incremental piece of a stream the client concatenates (LLM text tokens, audio chunks). Standard term across OpenAI / Anthropic / LiteLLM / Vercel AI SDK.
`start`	Begin marker for a compound operation (e.g. a tool call whose execution will be reported by a matching `end`).
`end`	Terminal marker. Carries the canonical resource id so reconnecting clients can refetch persisted state via REST.
`progress`	Periodic checkpoint with quantifiable completion, e.g. `{"checked": 42, "total": 100}`.
`created` / `updated` / `deleted`	Resource-lifecycle events for cross-client sync streams.
`error`	Terminal failure. Carries a stable `code` for client switching and a human-readable `detail`.

Payloads are flat JSON. The wire-level event: field already names the event type, so do not wrap the payload in {"type": ..., "data": ...}. Include the canonical resource UUID on terminal events so reconnecting clients can reconcile via REST.

Authentication

SSE endpoints use the same authentication stack as the rest of the API. Non-browser clients (CLI, programmatic) send the standard Authorization header — JWT or API key. Browser EventSource is the only widely available SSE client API and it cannot set custom headers. For that case only, the endpoint accepts a JWT via the ?access_token=<jwt> query parameter. The header always wins when present — a header is intentional, while a query parameter can leak into referers and logs, so it is consulted only as a fallback.

// Browser
const es = new EventSource(
  `/api/v1/scans/${scanId}/event-stream?access_token=${jwt}`
);

# CLI / programmatic — header, exactly like every other endpoint
curl -N -H "Authorization: Bearer $JWT" \
  https://<host>/api/v1/scans/$SCAN_ID/event-stream

Tenant isolation & security model

Authorization is enforced at two layers:

At connect, get_channels runs under the regular DRF stack inside the tenant transaction (rls_transaction). Resource lookups are RLS-scoped, so a user cannot even resolve a channel for a resource they cannot see. Narrow the queryset further (e.g. created_by=request.user) when a resource is per-user within a tenant.
After connect, SSEChannelManager.can_read_channel re-verifies tenant membership by parsing the tenant id embedded in the channel name. Cross-tenant subscription is rejected even if a URL-level check ever has a bug. A malformed channel name is treated as “not authorized”.

Because the tenant id lives inside the channel name, this gate works for any feature without the platform knowing anything about it.

Reconnect & state recovery

The platform deliberately ships without server-side replay (is_channel_reliable returns False). When a client reconnects, it does not receive missed events. Instead:

Terminal events (*.end) carry the canonical resource UUID.
On reconnect, the client refetches the authoritative state from the normal REST endpoint using that id.

Design your event payloads accordingly: deltas are ephemeral and concatenated in-flight; the durable truth always lives behind a REST resource.

Local development

The dev and production entrypoints both launch Gunicorn with the asgi worker (config.asgi:application). In dev, DJANGO_DEBUG=True enables hot reload; preload_app is automatically disabled under debug so edited code is picked up.
SSE uses a dedicated Valkey database (EVENTSTREAM_VALKEY_DB, default 2) kept separate from the Celery broker so a noisy broker cannot crowd out streaming traffic. It reuses the same VALKEY_* connection settings as the rest of the platform.

Env var	Default	Purpose
`EVENTSTREAM_VALKEY_DB`	`2`	Valkey DB index for the SSE Pub/Sub bus
`DJANGO_WORKER_CLASS`	`asgi`	Gunicorn worker class

Test the stream end to end with curl -N (disable buffering) and an auth header:

curl -N -H "Authorization: Bearer $JWT" \
  http://localhost:8080/api/v1/scans/$SCAN_ID/event-stream

Testing

The platform basis is covered by api/tests/test_sse.py (channel parsing, the tenant gate, and auth precedence). For a feature endpoint, test:

get_channels returns the expected channel for an authorized resource and raises NotFound/PermissionDenied otherwise.
Each publish_<event> helper emits the correct event type and flat payload (mock send_event).
The producer builds the channel with make_channel_name using the resource’s own tenant_id.

​When to use SSE

​How it works

​Transport: the server runs on ASGI

​The data flow

​Adding an SSE endpoint to your feature

​Event naming convention

​Authentication

​Tenant isolation & security model

​Reconnect & state recovery

​Local development

​Testing

When to use SSE

How it works

Transport: the server runs on ASGI

The data flow

Adding an SSE endpoint to your feature

Event naming convention

Authentication

Tenant isolation & security model

Reconnect & state recovery

Local development

Testing