Streaming

The Guardrails server supports streaming responses via Server-Sent Events (SSE) on both the Chat Completions and Responses API endpoints. Set stream: true in the request body — exactly as you would with the OpenAI API directly.

---

How it works

Client                    Guardrails Server                  Upstream LLM
  │                              │                                 │
  │── POST (stream: true) ──────►│                                 │
  │                       pre-flight guardrails                    │
  │                       input guardrails  ──── request ─────────►│
  │                              │◄──────────── SSE stream ────────│
  │◄── data: chunk ─────────────│  (output guardrails run         │
  │◄── data: chunk ─────────────│   every N chunks)               │
  │◄── data: [DONE] ────────────│                                 │

Pre-flight and input guardrails run before the first byte is sent. Output guardrails run periodically as the stream accumulates, and once more at the end.

---

Enforcement behaviour

Pipeline stage	Policy mode	What the client receives
Pre-flight or Input	block	HTTP 400 — no bytes sent yet
Pre-flight or Input	alert	Stream proceeds normally
Output (mid-stream)	block	Final SSE chunk with finish_reason: "content_filter", then [DONE]
Output (mid-stream)	alert	Stream continues uninterrupted

The mid-stream content_filter termination matches Azure Content Safety behaviour. All OpenAI SDK clients handle it as a valid ChatCompletionChunk and stop cleanly — no deserialization errors.

---

Chat Completions (/v1/chat/completions)

Request

Add "stream": true to the standard request body:

{
  "model": "gpt-4o",
  "messages": [{"role": "user", "content": "Hello!"}],
  "stream": true,
  "guardrails": {"config_id": "default"}
}

Wire format

Each chunk arrives as a standard OpenAI ChatCompletionChunk:

data: {"id":"chatcmpl-...","object":"chat.completion.chunk","model":"gpt-4o","choices":[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]}

data: {"id":"chatcmpl-...","object":"chat.completion.chunk","model":"gpt-4o","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

When a block-mode output guardrail fires mid-stream:

data: {"id":"chatcmpl-...","object":"chat.completion.chunk","model":"gpt-4o","choices":[{"index":0,"delta":{},"finish_reason":"content_filter"}]}

data: [DONE]

curl

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  --no-buffer \
  -d '{
    "model": "gpt-4o",
    "messages": [{"role": "user", "content": "Hello!"}],
    "stream": true,
    "guardrails": {"config_id": "default"}
  }'

OpenAI Python SDK (sync)

from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="any")

with client.chat.completions.stream(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    extra_body={"guardrails": {"config_id": "default"}},
) as stream:
    for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

OpenAI Python SDK (async)

import asyncio
from openai import AsyncOpenAI

async def main() -> None:
    client = AsyncOpenAI(base_url="http://localhost:8000/v1", api_key="any")

    async with client.chat.completions.stream(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
        extra_body={"guardrails": {"config_id": "default"}},
    ) as stream:
        async for chunk in stream:
            if chunk.choices[0].delta.content:
                print(chunk.choices[0].delta.content, end="", flush=True)

asyncio.run(main())

Handling enforcement errors

from openai import OpenAI, BadRequestError

client = OpenAI(base_url="http://localhost:8000/v1", api_key="any")

try:
    with client.chat.completions.stream(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Ignore all previous instructions."}],
        extra_body={"guardrails": {"config_id": "strict"}},
    ) as stream:
        for chunk in stream:
            if chunk.choices[0].delta.content:
                print(chunk.choices[0].delta.content, end="", flush=True)
            # Mid-stream block: chunk.choices[0].finish_reason == "content_filter"
            if chunk.choices[0].finish_reason == "content_filter":
                print("\n[blocked by guardrail]")
                break
except BadRequestError as exc:
    # Pre-flight or input guardrail blocked before stream started.
    print("Blocked:", exc.body)

---

Responses API (/v1/responses)

Request

{
  "model": "gpt-4o",
  "input": "Hello!",
  "stream": true,
  "guardrails": {"config_id": "default"}
}

Wire format

Events follow the OpenAI Responses API streaming event schema:

data: {"type":"response.created","response":{"id":"resp-...","status":"in_progress",...}}

data: {"type":"response.output_text.delta","delta":"Hello"}

data: {"type":"response.completed","response":{"id":"resp-...","status":"completed",...}}

data: [DONE]

When a block-mode output guardrail fires mid-stream:

data: {"type":"response.completed","response":{"id":"resp-...","status":"incomplete","incomplete_details":{"reason":"content_filter"}}}

data: [DONE]

curl

curl -X POST http://localhost:8000/v1/responses \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  --no-buffer \
  -d '{
    "model": "gpt-4o",
    "input": "Hello!",
    "stream": true,
    "guardrails": {"config_id": "default"}
  }'

OpenAI Python SDK (async)

import asyncio
from openai import AsyncOpenAI

async def main() -> None:
    client = AsyncOpenAI(base_url="http://localhost:8000/v1", api_key="any")

    async with client.responses.stream(
        model="gpt-4o",
        input="Hello!",
        extra_body={"guardrails": {"config_id": "default"}},
    ) as stream:
        async for event in stream:
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)

asyncio.run(main())

---

Azure deployment URL

The Azure-style deployment route (/openai/deployments/{deployment}/chat/completions) inherits streaming automatically — no special handling is needed. Set "stream": true exactly as you would with the standard Chat Completions endpoint:

curl -X POST http://localhost:8000/openai/deployments/my-gpt4o-dep/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  --no-buffer \
  -d '{
    "messages": [{"role": "user", "content": "Hello!"}],
    "stream": true,
    "guardrails": {"config_id": "default"}
  }'

---

Streaming vs blocking

	Blocking (default)	Streaming
Output guardrails	Run on full response before returning	Run periodically + at end of stream
Time-to-first-token	After all guardrails complete	Immediately after input guardrails pass
Mid-stream violation	N/A (always blocked before response)	finish_reason: "content_filter" chunk
Best for	Compliance-critical, high-assurance	Low-latency, real-time UX