Responses API advanced: structured outputs, streaming SSE, Batch API, prompt caching

Responses API (the input field, not messages) is OpenAI's current primary interface - most developers only use basic chat. This article covers four production-grade features that upgrade Responses API from 'can call' to 'production-grade high-availability, low-cost'.

1. JSON Schema strict mode (structured outputs)

Make the model output 100% to schema, zero parse failure.

from pydantic import BaseModel
from openai import OpenAI

class OrderStatus(BaseModel):
    order_id: str
    status: str  # 'pending' / 'shipped' / 'delivered'
    eta: str     # ISO date
    tracking_url: str | None = None

client = OpenAI()
response = client.responses.create(
    model="gpt-5.6-terra",
    input=[{"role": "user", "content": "Query order 12345 status"}],
    text={
        "format": {
            "type": "json_schema",
            "name": "order_status",        # name is required
            "strict": True,                # strict mode: 100% to schema
            "schema": OrderStatus.model_json_schema(),
        }
    },
)

# Direct parse, no try/except
order = OrderStatus.model_validate_json(response.output_text)
print(order.order_id, order.status)

Key constraints:

• name is required - only schema without name errors.
• strict: True makes the model output strictly conform to schema (no extra fields, no missing fields, no type errors).
• Schema must be JSON Schema 2020-12 compatible - Pydantic / Zod generated schemas usually work.
• Cost: slight latency bump (model does schema validation), but 'no errors' in production is worth tens of ms.

2. Streaming SSE parsing

Real-time UI + mid-stream function calling.

import sseclient
import json

def stream_response(prompt):
    response = requests.post(
        "https://api.openai.com/v1/responses",
        headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
        json={
            "model": "gpt-5.6-terra",
            "input": [{"role": "user", "content": prompt}],
            "stream": True,
        },
        stream=True,
    )
    client = sseclient.SSEClient(response.iter_content(chunk_size=1024))

    text_buffer = ""
    for event in client.events():
        data = json.loads(event.data)

        if data["type"] == "response.output_text.delta":
            # text fragment
            text_buffer += data["delta"]
            yield ("text", data["delta"])

        elif data["type"] == "response.function_call_arguments.delta":
            # function params streaming
            yield ("function_arg_delta", data["delta"])

        elif data["type"] == "response.function_call_arguments.done":
            # function params complete, call function
            args = json.loads(data["arguments"])
            result = my_function(**args)
            # Feed result back (need to maintain session / conversation_id)
            yield ("function_done", result)

Gotchas:

• Text output and function calling interleave - do not assume 'text first, then function'.
• Mid-stream function calling: listen to function_call_arguments.done, call function, feed result back via conversation.item.create (must maintain conversation state).
• Client must maintain full state (prior message history + current token accumulation) for correct assembly.

3. Prompt caching (three-tier cache)

Repeat prompts - 80%+ cost cut.

# === implicit caching (automatic) ===
# When prompt > 1k tokens and prefix is same, OpenAI auto-hits cache
# Cache TTL: 5-10 min
response = client.responses.create(
    model="gpt-5.6-terra",
    input=[
        {"role": "system", "content": LONG_SYSTEM_PROMPT},  # > 1k token
        {"role": "user", "content": "User question"},
    ],
)
# First call: cache miss, full price
# Same system prompt within 5 min: cache hit, cache price (~1/4 of input)

# === explicit caching (custom key) ===
response = client.responses.create(
    model="gpt-5.6-terra",
    input=[...],
    prompt_cache_key="user_12345",  # custom cache key, isolate by user / tenant
)

# === Control cache mode ===
response = client.responses.create(
    model="gpt-5.6-terra",
    input=[...],
    prompt_cache_options={
        "mode": "explicit",  # 'implicit' / 'explicit'
    },
)

When effective:

• Long system prompt (>1k token) + multiple calls
• Multi-turn conversation - history accumulates in prompt prefix
• Batch similar tasks - all calls share one large prompt

When ineffective:

• Every call prompt fully different
• prompt < 1k tokens
• Call interval exceeds 5-10 min (cache expires)

Quota limits: ≤4 writes per request, ≤50 breakpoints, ≤15 per key per minute.

4. Batch API (offline -50%)

Batch tasks that don't need realtime use Batch API.

# === Create batch task ===
batch = client.batches.create(
    input_file_id="file-abc123",  # uploaded JSONL request file
    endpoint="/v1/responses",
    completion_window="24h",       # max 24h return
    metadata={"description": "doc-summarize-batch-2026-08-14"},
)

# === Query batch status ===
status = client.batches.retrieve(batch.id)
print(f"Status: {status.status}, completed: {status.request_counts.completed}/{status.request_counts.total}")

# === Download result (when complete) ===
if status.status == "completed":
    output_file = client.files.content(status.output_file_id)
    for line in output_file.text.split("\n"):
        result = json.loads(line)
        print(result["custom_id"], result["response"]["body"])

Use cases:

• Batch doc summary (thousands of news)
• Batch data labeling (100k items)
• Batch translation (no realtime needed)
• Batch eval (generate test cases)
• Nightly offline jobs

Hard limits:

• No streaming
• No web search / file search
• No Realtime API
• Only responses.create / chat.completions.create synchronous endpoints

Cost comparison: Batch API price -50%, but scheduling cycle is long (up to 24h).

5. Cost optimization combo

Three-piece combo: drop monthly cost from $100K to $30K.

# === Three-piece combo: batch doc summary ===

# Step 1: Build JSONL requests (one per request)
requests_jsonl = []
for doc_id, doc_text in documents:
    requests_jsonl.append({
        "custom_id": f"doc-{doc_id}",
        "method": "POST",
        "url": "/v1/responses",
        "body": {
            "model": "gpt-5.6-luna",  # cheapest model for summary
            "input": [
                {"role": "system", "content": LONG_SUMMARIZE_PROMPT},  # shared prompt via cache
                {"role": "user", "content": doc_text},
            ],
            "text": {
                "format": {
                    "type": "json_schema",
                    "name": "summary",
                    "strict": True,
                    "schema": {
                        "type": "object",
                        "properties": {
                            "summary": {"type": "string"},
                            "key_points": {"type": "array", "items": {"type": "string"}},
                        },
                        "required": ["summary", "key_points"],
                        "additionalProperties": False,
                    },
                }
            },
        },
    })

# Step 2: Upload + create
file = client.files.create(file=("\n".join(json.dumps(r) for r in requests_jsonl)).encode(), purpose="batch")
batch = client.batches.create(input_file_id=file.id, endpoint="/v1/responses", completion_window="24h")

Cost savings estimate (100k doc summary scenario):

• Without combo: $100K/month
• Luna model only: $20K/month
• • JSON Schema (avoid re-parse): $20K/month (saved parse failure retries)
• • Batch API: $10K/month (-50%)
• • Prompt caching: $8K/month (shared prompt halved)

Total saving: 92%.

FAQ

1. Must structured outputs use strict mode?

Strongly recommended. Strict mode ({'strict': True}) makes the model output 100% to JSON Schema - no extra fields, no missing fields, no type errors. Non-strict (default) may output 'roughly matching' JSON, causing occasional client parse failures. Cost is slight latency bump, but production-grade strict mode rarely errors.

2. Mid-stream function calling?

Three steps: (1) listen to response.output_text.delta to accumulate text; (2) listen to response.function_call_arguments.delta to stream function params; (3) on function_call_arguments.done, call function and feed result via conversation.item.create. Gotcha: text and function interleave.

3. When to use Batch API?

Criteria: (1) realtime required? (yes -> Realtime API or normal Responses); (2) can wait 24h? (yes -> Batch API, -50%). Scenarios: batch doc summary, batch labeling, batch translation, batch eval, nightly offline jobs. Hard limits: no streaming, no web search / file search.

4. When is prompt caching effective?

Three scenarios: (1) long system prompt (>1k tokens) + multiple calls; (2) multi-turn conversation (history in prefix); (3) batch similar tasks (shared prompt). Failure scenarios: every call prompt fully different / prompt < 1k tokens / call interval exceeds 5-10 min.

5. Can Batch API + prompt caching stack?

Yes with trade-offs. Batch API already -50%, add prompt caching for further savings (cache-hit portion at cache rate, ~1/4 of input). But Batch API scheduling cycle is long (up to 24h), cache TTL may expire. Recommendation: long-running high-frequency calls use prompt caching realtime; low-frequency / large-batch use Batch API.

6. Monitoring metrics

Production must-watch numbers:

# === Cache hit rate ===
cached = response.usage.input_tokens_details.cached_tokens
total_input = response.usage.input_tokens
cache_hit_rate = cached / total_input if total_input > 0 else 0
# Target ≥ 50%

# === Batch completion rate ===
batch = client.batches.retrieve(batch_id)
print(f"completed: {batch.request_counts.completed}/{batch.request_counts.total}")
# Target = 100%

# === Function call trigger rate ===
# Count function_call events / total response from logs

Next steps

• New to Responses API? Read OpenAI API Beginner: Your First GPT-5.6 Call Explained.
• Want function calling? Read Function Calling with the OpenAI API: A Complete Guide to Tool Use in the Responses API.
• Want error handling? Read OpenAI API Error Handling and Retry: 401/429/5xx Patterns.