> ## Documentation Index
> Fetch the complete documentation index at: https://docs.promptguard.co/llms.txt
> Use this file to discover all available pages before exploring further.

# Streaming with Protection

> Secure streaming LLM responses in real-time

<Info>
  This example shows how to protect streaming LLM responses, detecting threats and PII as tokens arrive rather than waiting for the complete response.
</Info>

## Overview

Streaming presents unique security challenges:

* Responses arrive token-by-token
* Threats may span multiple chunks
* Users see partial content before full analysis

PromptGuard handles streaming with real-time scanning.

## How Streaming Protection Works

```mermaid theme={"system"}
sequenceDiagram
    participant App
    participant PromptGuard
    participant LLM

    App->>PromptGuard: Stream request
    PromptGuard->>PromptGuard: Scan input
    PromptGuard->>LLM: Forward request

    loop For each chunk
        LLM->>PromptGuard: Token chunk
        PromptGuard->>PromptGuard: Buffer & scan
        PromptGuard->>App: Safe chunk
    end

    PromptGuard->>App: Stream complete
```

## Implementation

### Auto-Instrumentation (Recommended)

```python theme={"system"}
import promptguard
promptguard.init(
    api_key="pg_live_xxxxxxxx",
    mode="enforce",
    scan_responses=True,
)

from openai import OpenAI
client = OpenAI()

# Streaming works exactly as before - protection is automatic
stream = client.chat.completions.create(
    model="gpt-5-nano",
    messages=[{"role": "user", "content": "Write a story about a hacker"}],
    stream=True,
)

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

### With Response Scanning

Enable `scan_responses=True` to scan the complete response after streaming:

```python theme={"system"}
promptguard.init(
    api_key="pg_live_xxxxxxxx",
    mode="enforce",
    scan_responses=True,  # Scan after stream completes
)

# If the full response contains threats, an error is raised
# after the stream completes
try:
    stream = client.chat.completions.create(
        model="gpt-5-nano",
        messages=[{"role": "user", "content": "Hello"}],
        stream=True,
    )

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

    # Response scan happens here (at stream end)
    print("\n\nStream completed safely!")

except promptguard.PromptGuardBlockedError as e:
    print(f"\n\nResponse contained: {e.decision.threat_type}")
```

### Real-Time Chunk Scanning

For immediate threat detection during streaming:

```python theme={"system"}
from promptguard import GuardClient

guard = GuardClient(api_key="pg_live_xxxxxxxx")

def secure_stream(messages: list):
    """Stream with real-time scanning."""
    from openai import OpenAI
    client = OpenAI()

    stream = client.chat.completions.create(
        model="gpt-5-nano",
        messages=messages,
        stream=True,
    )

    buffer = ""
    chunk_size = 50  # Scan every 50 characters

    for chunk in stream:
        content = chunk.choices[0].delta.content or ""
        buffer += content

        # Scan when buffer reaches threshold
        if len(buffer) >= chunk_size:
            decision = guard.scan(
                messages=[{"role": "assistant", "content": buffer}],
                direction="output",
            )

            if decision.blocked:
                yield "[CONTENT BLOCKED]"
                return

            if decision.redacted:
                yield decision.redacted_messages[0]["content"]
            else:
                yield buffer

            buffer = ""

    # Scan remaining buffer
    if buffer:
        decision = guard.scan(
            messages=[{"role": "assistant", "content": buffer}],
            direction="output",
        )

        if decision.blocked:
            yield "[CONTENT BLOCKED]"
        elif decision.redacted:
            yield decision.redacted_messages[0]["content"]
        else:
            yield buffer

# Usage
for safe_chunk in secure_stream([{"role": "user", "content": "Tell me a story"}]):
    print(safe_chunk, end="", flush=True)
```

## Node.js Streaming

### Auto-Instrumentation

```typescript theme={"system"}
import { init } from 'promptguard-sdk';
import OpenAI from 'openai';

init({
  apiKey: 'pg_live_xxxxxxxx',
  mode: 'enforce',
  scanResponses: true,
});

const client = new OpenAI();

async function streamChat() {
  const stream = await client.chat.completions.create({
    model: 'gpt-5-nano',
    messages: [{ role: 'user', content: 'Write a poem' }],
    stream: true,
  });

  for await (const chunk of stream) {
    const content = chunk.choices[0]?.delta?.content || '';
    process.stdout.write(content);
  }
}

streamChat();
```

### With Vercel AI SDK

```typescript theme={"system"}
import { init, promptGuardMiddleware } from 'promptguard-sdk';
import { streamText } from 'ai';
import { openai } from '@ai-sdk/openai';

init({ apiKey: 'pg_live_xxxxxxxx' });

async function handler(req: Request) {
  const { messages } = await req.json();

  const result = await streamText({
    model: openai('gpt-5-nano'),
    messages,
    experimental_middleware: promptGuardMiddleware({
      scanResponses: true,
    }),
  });

  return result.toDataStreamResponse();
}
```

### Server-Sent Events (SSE)

```typescript theme={"system"}
import { init, GuardClient } from 'promptguard-sdk';
import OpenAI from 'openai';
import { Response } from 'express';

init({ apiKey: 'pg_live_xxxxxxxx' });

async function streamSSE(res: Response, messages: any[]) {
  const client = new OpenAI();
  const guard = new GuardClient({ apiKey: 'pg_live_xxxxxxxx' });

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

  const stream = await client.chat.completions.create({
    model: 'gpt-5-nano',
    messages,
    stream: true,
  });

  let buffer = '';

  for await (const chunk of stream) {
    const content = chunk.choices[0]?.delta?.content || '';
    buffer += content;

    // Scan periodically
    if (buffer.length >= 100) {
      const decision = await guard.scanAsync({
        messages: [{ role: 'assistant', content: buffer }],
        direction: 'output',
      });

      if (decision.blocked) {
        res.write('data: [BLOCKED]\n\n');
        res.end();
        return;
      }

      const safeContent = decision.redacted
        ? decision.redactedMessages[0].content
        : buffer;

      res.write(`data: ${JSON.stringify({ content: safeContent })}\n\n`);
      buffer = '';
    }
  }

  // Send remaining
  if (buffer) {
    res.write(`data: ${JSON.stringify({ content: buffer })}\n\n`);
  }

  res.write('data: [DONE]\n\n');
  res.end();
}
```

## FastAPI Streaming

```python theme={"system"}
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import promptguard
from openai import OpenAI

promptguard.init(api_key="pg_live_xxxxxxxx", scan_responses=True)

app = FastAPI()
client = OpenAI()

@app.post("/chat/stream")
async def stream_chat(request: dict):
    messages = request.get("messages", [])

    async def generate():
        stream = client.chat.completions.create(
            model="gpt-5-nano",
            messages=messages,
            stream=True,
        )

        for chunk in stream:
            content = chunk.choices[0].delta.content or ""
            yield f"data: {content}\n\n"

        yield "data: [DONE]\n\n"

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

## Handling Blocked Streams

When a threat is detected mid-stream:

```python theme={"system"}
from promptguard import PromptGuardBlockedError

def safe_stream_handler(messages):
    try:
        stream = client.chat.completions.create(
            model="gpt-5-nano",
            messages=messages,
            stream=True,
        )

        for chunk in stream:
            yield chunk.choices[0].delta.content or ""

    except PromptGuardBlockedError as e:
        # Stream was blocked - notify user
        yield "\n\n[Response interrupted for security reasons]"

        # Log the event
        logging.warning(f"Stream blocked: {e.decision.event_id}")
```

## Performance Considerations

| Mode                 | Latency            | Security |
| -------------------- | ------------------ | -------- |
| Input-only scanning  | Minimal            | Good     |
| Full response scan   | +50-100ms at end   | Better   |
| Real-time chunk scan | +20-50ms per chunk | Best     |

### Recommendations

1. **For chat interfaces**: Use input scanning + end-of-stream response scan
2. **For sensitive data**: Use real-time chunk scanning
3. **For low-latency needs**: Use input-only scanning with async response analysis

## Best Practices

1. **Buffer appropriately** - Don't scan every token, batch into meaningful chunks
2. **Handle interruptions gracefully** - Users may see partial content
3. **Log blocked streams** - Track for security analysis
4. **Consider UX** - Decide if you show partial content before blocking

## Next Steps

<CardGroup cols={2}>
  <Card title="Streaming Reference" icon="stream" href="/guides/streaming">
    Full streaming documentation
  </Card>

  <Card title="Error Handling" icon="triangle-exclamation" href="/production/error-handling">
    Handle errors gracefully
  </Card>
</CardGroup>