Streaming Chat

Category: AI & Agents

This page is generated from the Air Pipe marketplace. Browse it live to install into your organization.

Stream a model's answer to the browser as it is generated, instead of showing a spinner for twenty seconds and then a wall of text. Air Pipe reads the provider's token stream and forwards it to a realtime channel, so the UI renders text while the model is still writing.

Works with any OpenAI-compatible provider — OpenAI, DeepSeek, Groq, Together, Mistral, OpenRouter, or a local vLLM/Ollama endpoint — by pointing one variable at its base URL.

Why this is hard without it

Streaming a model to a browser normally means running your own service: hold the upstream SSE connection, keep a socket per user, fan out across instances, and stop a slow reader from stalling the model call. That is the service this pack replaces.

	Roll your own	This pack
Read the provider's token stream	your code	✅ `stream: true`
Push tokens to the browser live	your socket server	✅ realtime channel
Fan out across nodes	Redis / NATS	✅ built in
Per-user channel authorisation	your code	✅ `subscribe_authorizer`
Slow reader can't stall the model	your backpressure	✅ handled

What's included

Interface	Method	What it does
`session`	POST	Starts a chat session — returns a `session_id` and the JWT that may watch it
`live`	WS	The socket the browser subscribes on
`chat`	POST	Ask, stream tokens to the channel, return the full text when done
`authorize-session`	internal	Channel ACL — binds a channel to the session in the token

Platform variables

Variable	Example
`LLM_BASE_URL`	`https://api.openai.com`
`LLM_API_KEY`	your provider key
`LLM_MODEL`	`gpt-4o-mini`, `deepseek-chat`, `llama-3.3-70b-versatile`
`JWT_SECRET`	signs and verifies the session token

Using it

1 — start a session

curl -X POST https://<host>/session \
  -H 'content-type: application/json' \
  -d '{"username":"alice"}'
# -> { "data": { "Issue": { "data": { "session_id": "8f2a...", "token": "eyJ..." } } } }
#    Actions appear under `data` keyed by name -- read the token at data.Issue.data.token.

2 — open the socket and subscribe

const ws = new WebSocket(`wss://<host>/ws/<ORG_UUID>/production/live?token=${token}`);
ws.onopen = () => ws.send(JSON.stringify({ ap_subscribe: `chat:${username}` }));

let answer = "";
ws.onmessage = (e) => {
  const { data } = JSON.parse(e.data);
  if (!data || data.delta === undefined) return;   // acks and other frames
  answer += data.delta;                            // append in seq order
  render(answer);
  if (data.done) finish(data.dropped);             // dropped > 0 = frames were missed
};

3 — ask

curl -X POST https://<host>/chat \
  -H 'content-type: application/json' -H "airpipe-jwt: $token" \
  -d '{"session_id":"8f2a...","messages":[{"role":"user","content":"Write a haiku about streaming."}]}'

Tokens appear on the socket while that request is still running.

What arrives on the channel

{"ap_channel":"chat:alice","data":{"seq":0,"delta":"Once","done":false}}
{"ap_channel":"chat:alice","data":{"seq":1,"delta":" upon a","done":false}}
{"ap_channel":"chat:alice","data":{"seq":2,"delta":"","done":true}}

seq increases by one per frame — a gap means a frame was missed.
delta is a run of characters, not one token. Deltas are batched (~50ms or 200 characters) because each publish reaches every node holding a subscriber; per-token publishing would be enormously wasteful. About twenty frames a second reads as smooth.
done marks the end. Treat it as the signal to stop, not silence.
dropped appears on the final frame only if a slow reader missed frames. A slow subscriber never stalls the model call — it loses frames instead, and is told so.

Channel naming — read this before you copy the pattern

The channel is per user (chat:<username>), derived from the verified token, and the session id rides in the request rather than the channel name. Two reasons, both of which matter more than they look:

A channel is not free to create. The broker keeps a commit log per subscribed filter and never releases it — unsubscribing does not free it. chat:<new-uuid> per conversation leaks one log per conversation for the life of the process. chat:<username> is bounded by how many users you have. Name channels from a bounded set — user, org, room — never from a per-request or per-session identifier. Hosted platforms hide this from you (Ably garbage collects idle channels, Pusher has no channel state at all); a self-hosted broker does not.

The channel must come from the token, not the body. If the interface took the channel from the request, a caller could make the server publish their stream onto someone else's channel. Auth verifies the JWT and the channel is built from the claim.

One consequence to know, and it is a hard one rather than a style note: the envelope is fixed at {seq, delta, done} (plus dropped) and a config cannot add a field to it — the session id is not in the frame and cannot be put there. So a client with two conversations open at once on the same user cannot tell the streams apart. This pack assumes one active conversation per user. Carrying a correlation id in the frame is an open engine request.

Notes

Two stream keys, different jobs. stream: true inside http: READS the provider's response as a stream. stream: on the interface says WHERE those pieces go. You need both.
The channel ACL is real. authorize-session binds a channel to the session_id claim in the token, so guessing another user's session id gets you nothing. is_valid_jwt decodes the token and exposes the claims as jwt_claims, which is how the check reads it.
Swap the demo login in session for a real check — a database: lookup plus a bcrypt compare; see the rest-api-starter pack.

Configuration

chat.yml

name: StreamingChat
description: >
  Stream a model's answer to the browser as it is generated. The pipeline reads the provider's
  token stream and forwards it to a realtime channel, so a chat UI renders text while the model
  is still writing — instead of showing a spinner and then a wall of text. Works with any
  OpenAI-compatible provider.

docs: true

# ── How it fits together ────────────────────────────────────────────────────────
# The browser does two things:
#
#   1. Opens ONE WebSocket to the `live` interface and subscribes to its own channel:
#        {"ap_subscribe": "chat:alice"}
#   2. POSTs the prompt to `chat` with that same session id.
#
# Tokens then arrive on the socket as they are generated:
#
#   {"ap_channel":"chat:alice","data":{"seq":0,"delta":"Once","done":false}}
#   {"ap_channel":"chat:alice","data":{"seq":1,"delta":" upon a","done":false}}
#   {"ap_channel":"chat:alice","data":{"seq":2,"delta":"","done":true}}
#
# NAME CHANNELS FROM A BOUNDED SET — per USER here, never per session or per request. A
# channel is not free to create: the broker keeps a commit log per subscribed filter and
# never releases it, so `chat:<new-uuid>` on every conversation leaks one log per
# conversation for the life of the process. `chat:<user>` is bounded by how many users you
# have.
#
# The trade this makes, stated plainly because it bites at runtime: the frame envelope is
# fixed at {seq, delta, done} (+ `dropped`) and a config cannot add to it, so ONE ACTIVE
# CONVERSATION PER USER. Two at once on the same user are indistinguishable on the wire.
# Carrying a correlation id in the frame is filed as an engine request.
#
# Append every `delta` in `seq` order; stop on `done`. A `dropped` count on the final frame
# means a slow client missed frames — reload rather than assume the text is complete.
#
# TWO `stream` keys, doing different jobs. Do not confuse them:
#   `stream: true` INSIDE `http:`   — read the provider's response as a stream.
#   `stream:` on the INTERFACE      — forward those pieces to a channel.
# You need both. The first without the second still works; it just keeps the pieces to itself.
#
# Deltas are batched (~50ms or 200 characters), not published per token. A publish reaches
# every node holding a subscriber, so per-token publishing would be enormously wasteful. ~20
# frames a second reads as smooth.
#
# Set these platform variables when installing:
#   LLM_BASE_URL   e.g. https://api.openai.com  (or api.deepseek.com, api.groq.com, ...)
#   LLM_API_KEY    the provider key
#   LLM_MODEL      e.g. gpt-4o-mini / deepseek-chat / llama-3.3-70b-versatile
#   JWT_SECRET     signs + verifies the session token

interfaces:

  # ── The socket the browser watches ──────────────────────────────────────────────
  # `subscribe_authorizer` binds a session channel to the token that owns it, so one user
  # cannot subscribe to another user's conversation by guessing an id.
  live:
    ws: "on"                       # realtime — no `output` needed
    subscribe_authorizer: authorize-session
    summary: Realtime socket carrying the model's tokens
    description: Subscribe to `chat:<username>` and receive answers as they are generated.
    tags: [websocket, realtime, streaming, chat, ai]
    actions:
      - name: Ping
        input: a|body|

  # Channel ACL — runs on every `ap_subscribe` with {"channel":"<name>"} as a|body| and the
  # socket's token as a|headers::airpipe-jwt|. Allowed iff it returns 2xx.
  authorize-session:
    actions:
      - name: Verify
        input: a|headers|
        assert:
          http_code_on_error: 403
          error_message: "not authorized for this channel"
          tests:
            - value: airpipe-jwt
              is_valid_jwt: a|ap_var::JWT_SECRET|
        # `is_valid_jwt` also DECODES the token and attaches the claims to this action's data
        # as `jwt_claims` — no read_jwt needed to reach one.

      # Bind the channel to the USER in the token, so a token only opens its own chat and a
      # guessed name gets nothing.
      - name: Bind
        run_when_succeeded:
          actions: [Verify]
          http_code_on_error: 403
        input: a|body|
        assert:
          error_message: "channel not allowed for this user"
          tests:
            - value: channel
              is_equal_to: "chat:a|Verify::jwt_claims.username|"

  # ── Issue a session token ───────────────────────────────────────────────────────
  # Mints the session id AND the token that may watch it. Swap the demo check for a real one
  # (a database: lookup + bcrypt compare — see the `rest-api-starter` pack).
  session:
    output: http
    method: POST
    summary: Start a chat session and get its realtime token
    description: Returns a session id plus the JWT used to open the socket and post prompts.
    tags: [auth, chat, jwt]
    accepts:
      example:
        username: alice
    actions:
      - name: Validate
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "username required"
          tests:
            - value: username
              is_not_empty: true

      - name: Issue
        run_when_succeeded: [Validate]
        input: a|Validate|
        post_transforms:
          - add_attribute:
              session_id: a|uuid|
          - add_jwt:
              key: token
              secret: a|ap_var::JWT_SECRET|
              exp: 2h
              data: [username, session_id]
          - keep_attributes: ["session_id", "token"]

  # ── Ask, and stream the answer (the common case) ────────────────────────────────
  # The request stays open for as long as the model takes, and the caller also gets the
  # complete text back in the response — while the browser has already rendered it live off
  # the channel. Use this unless the answer can outlive the caller's patience.
  chat:
    output: http
    method: POST
    stream:
      from: AskModel               # the action reading the provider's stream
      to: channel
      channel: "chat:a|Auth::jwt_claims.username|"
    summary: Ask the model and stream the answer to the session channel
    description: Tokens arrive on `chat:<username>` as they are generated.
    tags: [ai, streaming, chat, realtime]
    accepts:
      example:
        session_id: "8f2a1c3e-..."
        messages:
          - role: user
            content: "Write a haiku about streaming."
    actions:
      # The channel is derived from the TOKEN, never from the body. Taking it from the body
      # would let a caller make the server publish their stream onto someone else's channel.
      - name: Auth
        input: a|headers|
        hide_data_on_success: true      # else the response echoes the request headers back
        assert:
          http_code_on_error: 401
          error_message: "a valid session token is required"
          tests:
            - value: airpipe-jwt
              is_valid_jwt: a|ap_var::JWT_SECRET|

      - name: CheckInput
        run_when_succeeded: [Auth]
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 422
          error_message: "session_id and messages are required"
          tests:
            - value: session_id
              is_not_empty: true
            - value: messages
              is_not_null: true

      - name: AskModel
        run_when_succeeded: [CheckInput]
        hide_data_on_success: true      # the provider's whole raw response, otherwise
        timeout: 300000
        http:
          url: "a|ap_var::LLM_BASE_URL|/v1/chat/completions"
          method: POST
          stream: true             # READ the provider's stream
          idle_timeout: 45s        # a stalled provider fails fast; a slow one is left alone
          headers:
            content-type: application/json
            authorization: "Bearer a|ap_var::LLM_API_KEY|"
          body: |
            {
              "model": a|ap_var::LLM_MODEL->double_quote|,
              "stream": true,
              "messages": a|CheckInput::messages|
            }

      - name: Answer
        run_when_succeeded: [AskModel]
        input: a|AskModel::body.choices|[0].message.content|

Streaming Chat

Why this is hard without it​

What's included​

Platform variables​

Using it​

What arrives on the channel​

Channel naming — read this before you copy the pattern​

Notes​

Configuration​

chat.yml​