Streaming

Capture streaming responses from OpenAI and Anthropic, accumulating chunks while auto-instrumentation records the underlying API calls.

Environment variables

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

import asyncio

import waxell_observe
from waxell_observe import WaxellContext

waxell_observe.init()

from openai import OpenAI
import anthropic

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


async def streaming_comparison(prompt: str):
    async with WaxellContext(agent_name="streaming-demo") as ctx:
        # OpenAI streaming
        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)
        ctx.record_step("openai_stream", output={"chunks": len(openai_chunks)})

        # Anthropic streaming
        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)
        ctx.record_step("anthropic_stream", output={"chunks": len(anthropic_chunks)})

        ctx.set_result({
            "openai_length": len(openai_result),
            "anthropic_length": len(anthropic_result),
        })


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

What this demonstrates

• Streaming token capture -- chunk-by-chunk accumulation for both OpenAI and Anthropic streaming APIs.
• Auto-instrumentation -- waxell_observe.init() captures the underlying streaming calls automatically, even though you consume tokens manually.
• Step recording -- each provider's stream is recorded as a distinct step with chunk count metadata.
• WaxellContext -- explicit async trace lifecycle wrapping both streaming calls in a single trace.

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