Skip to main content

Streaming

Capture streaming responses from OpenAI and Anthropic. Auto-instrumentation records the underlying streaming calls automatically -- you just consume the tokens.

Environment variables

This example requires OPENAI_API_KEY, ANTHROPIC_API_KEY, WAXELL_API_KEY, and WAXELL_API_URL.

import asyncio

import waxell_observe as waxell

waxell.init()

from openai import OpenAI
import anthropic

openai_client = OpenAI()
anthropic_client = anthropic.Anthropic()


@waxell.observe(agent_name="streaming-demo")
async def streaming_comparison(prompt: str):
    # OpenAI streaming -- auto-captured
    openai_chunks = []
    stream = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    for chunk in stream:
        if chunk.choices[0].delta.content:
            openai_chunks.append(chunk.choices[0].delta.content)
    openai_result = "".join(openai_chunks)

    # Anthropic streaming -- auto-captured
    anthropic_chunks = []
    with anthropic_client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            anthropic_chunks.append(text)
    anthropic_result = "".join(anthropic_chunks)

    return {
        "openai_length": len(openai_result),
        "anthropic_length": len(anthropic_result),
    }


asyncio.run(streaming_comparison("Explain machine learning in simple terms."))

What this demonstrates​

  • Auto-instrumentation handles streaming -- waxell.init() captures streaming calls automatically for OpenAI, Anthropic, Gemini, Cohere, and Bedrock. No manual recording required.
  • @waxell.observe -- declarative trace lifecycle around the agent function. Both streaming calls land under one trace.
  • Streaming token capture -- chunk-by-chunk consumption works normally; the SDK records the underlying stream behind the scenes.

Run it​

export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export WAXELL_API_KEY="your-waxell-api-key"
export WAXELL_API_URL="https://api.waxell.ai"

python streaming.py