Skip to content

Streaming Responses

Streaming responses (stream=True) work transparently with Paygent. Your code looks the same — you iterate the generator, Paygent captures token counts when the stream ends, and the cost tracking happens after the final chunk.

How it works

When the patched client.chat.completions.create(stream=True) runs, Paygent doesn't return the original generator. It returns a StreamWrapper that:

  1. Yields every chunk from the original stream unchanged (zero latency overhead)
  2. Accumulates chunks in a list as they pass through
  3. When StopIteration is raised (the stream ends), inspects the accumulated chunks for usage data
  4. Builds a UsageEvent and pushes it to the queue

From your code's perspective the wrapper is indistinguishable from the original — same __iter__, same __next__, same context manager support.

Capturing tokens: include_usage

OpenAI's streaming API only includes usage data in the final chunk if you ask for it. You ask by passing stream_options={"include_usage": True}.

from openai import OpenAI
from paygent import Paygent, paygent_context

pg = Paygent.init(api_key="pg_live_...")
client = OpenAI()

with paygent_context(user_id="user_123"):
    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Tell me a story"}],
        stream=True,
        stream_options={"include_usage": True},   # ← this
    )

    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()

import OpenAI from "openai";
import { Paygent, paygentContext } from "@paygentjs/sdk";

const pg = await Paygent.init({ apiKey: "pg_live_..." });
const openai = pg.instrument(new OpenAI());

await paygentContext({ userId: "user_123" }, async () => {
  const stream = await openai.chat.completions.create({
    model: "gpt-4o-mini",
    messages: [{ role: "user", content: "Tell me a story" }],
    stream: true,
    // No `stream_options` needed — see note below.
  });

  for await (const chunk of stream) {
    const delta = chunk.choices[0]?.delta?.content;
    if (delta) process.stdout.write(delta);
  }
  process.stdout.write("\n");
});

The JS SDK injects include_usage for you

On an instrumented streaming call, Paygent automatically adds stream_options: { include_usage: true } before the request goes out, so metering works without you remembering the flag. (If you bypass instrument() and use the raw client, you still need to set it yourself.)

After the loop ends, Paygent extracts chunk.usage from the final chunk, calculates cost, and meters the call.

Without include_usage

If you don't pass stream_options={"include_usage": True}, the OpenAI API doesn't include usage info in any chunk. Paygent has no way to know how many tokens were consumed.

What happens:

  • The stream still works correctly — you get every chunk, Paygent doesn't interfere
  • Paygent fires update_cache with total_tokens=0 and cost_total=0
  • No usage event reflects the real cost
  • Your guards never see the spend → user can effectively bypass spend caps via streaming

Always set include_usage=True for streamed calls

Without it, streamed calls don't contribute to spend tracking. The fix is one line: 

stream_options={"include_usage": True}

For Anthropic (anthropic SDK), the message-stream API includes usage in message_delta events automatically — no equivalent flag needed.

Async streaming

Identical pattern with AsyncOpenAI:

from openai import AsyncOpenAI
import asyncio

async_client = AsyncOpenAI()

async def stream_response(user_id: str, prompt: str):
    with paygent_context(user_id=user_id):
        stream = await async_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
            stream_options={"include_usage": True},
        )

        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                print(chunk.choices[0].delta.content, end="", flush=True)
        print()

asyncio.run(stream_response("user_123", "Tell me a poem"))

// In Node there is a single async path — the same instrumented client
// streams via `for await`, no separate async client needed.
async function streamResponse(userId: string, prompt: string) {
  await paygentContext({ userId }, async () => {
    const stream = await openai.chat.completions.create({
      model: "gpt-4o-mini",
      messages: [{ role: "user", content: prompt }],
      stream: true,
    });

    for await (const chunk of stream) {
      const delta = chunk.choices[0]?.delta?.content;
      if (delta) process.stdout.write(delta);
    }
    process.stdout.write("\n");
  });
}

await streamResponse("user_123", "Tell me a poem");

The JS stream wrapper is an async iterator: it yields every chunk through unchanged and fires metering when the iterator is exhausted.

Streaming with FastAPI

A common production pattern — streaming server-sent events to the browser:

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from openai import OpenAI

from paygent import Paygent, PaygentLimitExceeded, paygent_context

pg = Paygent.init(api_key=os.environ["PAYGENT_API_KEY"])
client = OpenAI()
app = FastAPI()

@app.get("/chat/stream")
def stream(user_id: str, prompt: str):
    def event_stream():
        try:
            with paygent_context(user_id=user_id):
                response = client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{"role": "user", "content": prompt}],
                    stream=True,
                    stream_options={"include_usage": True},
                )
                for chunk in response:
                    if chunk.choices and chunk.choices[0].delta.content:
                        yield f"data: {chunk.choices[0].delta.content}\n\n"
                yield "data: [DONE]\n\n"
        except PaygentLimitExceeded as e:
            yield f"event: limit\ndata: {e.guard_result.message}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

// Express — stream server-sent events to the browser.
import express from "express";
import OpenAI from "openai";
import { Paygent, paygentContext, PaygentLimitExceeded } from "@paygentjs/sdk";

const pg = await Paygent.init({ apiKey: process.env.PAYGENT_API_KEY });
const openai = pg.instrument(new OpenAI());
const app = express();

app.get("/chat/stream", async (req, res) => {
  const userId = String(req.query.user_id);
  const prompt = String(req.query.prompt);

  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");

  try {
    await paygentContext({ userId }, async () => {
      // Hard gate throws HERE, before any chunk is produced.
      const stream = await openai.chat.completions.create({
        model: "gpt-4o-mini",
        messages: [{ role: "user", content: prompt }],
        stream: true,
      });

      for await (const chunk of stream) {
        const delta = chunk.choices[0]?.delta?.content;
        if (delta) res.write(`data: ${delta}\n\n`);
      }
      res.write("data: [DONE]\n\n");
    });
  } catch (err) {
    if (err instanceof PaygentLimitExceeded) {
      res.write(`event: limit\ndata: ${err.guardResult.message}\n\n`);
    } else {
      res.write(`event: error\ndata: ${String(err)}\n\n`);
    }
  } finally {
    res.end();
  }
});

The hard gate raises before the stream starts (during client.chat.completions.create(...), before any chunks come back), so you can convert it to an SSE error event and close the stream cleanly.

Streaming with pg.wrap()

The wrap() and awrap() paths also handle streams — same StreamWrapper returns, same behavior:

stream = pg.wrap(
    lambda: client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "..."}],
        stream=True,
        stream_options={"include_usage": True},
    ),
    user_id="user_123",
    model="gpt-4o-mini",
)

for chunk in stream:
    ...

const stream = await pg.wrap(
  () =>
    openai.chat.completions.create({
      model: "gpt-4o-mini",
      messages: [{ role: "user", content: "..." }],
      stream: true,
      stream_options: { include_usage: true }, // wrap() doesn't auto-inject
    }),
  { userId: "user_123", model: "gpt-4o-mini", stream: true },
);

for await (const chunk of stream) {
  // ...
}

Pass stream: true to wrap() and set include_usage yourself

Unlike instrument(), the wrap() path doesn't rewrite your request, so add stream_options: { include_usage: true } to the OpenAI call and stream: true to the wrap() options so it knows to wrap the iterator.

Edge cases

  • Stream is consumed only partially. If you break out of the loop early without exhausting the stream, the metering callback does not fire (it's tied to StopIteration). The call is unmetered. To force metering, fully iterate the stream or close it explicitly via stream.close().
  • Stream raises mid-flight. If the network drops between chunks, the metering happens with whatever was accumulated so far. You'll get a usage event with the partial token count visible up to the failure.
  • Multiple streams concurrently for the same user. Each stream has its own StreamWrapper. They each independently call update_cache after their own StopIteration. The cache update is atomic per-event, so concurrent streams don't race.

Next steps