Streaming plus token accounting: stream_events v3 with usage callbacks

Streaming with token accounting

from langchain.agents import create_agent
from langchain_core.callbacks import UsageMetadataCallbackHandler

agent = create_agent(model="gpt-5-nano", tools=[get_weather])

callback = UsageMetadataCallbackHandler()

stream = agent.stream_events(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    version="v3",
    config={"callbacks": [callback]},
)

for message in stream.messages:
    for delta in message.text:
        print(delta, end="", flush=True)

print(callback.usage_metadata)

Rules

• stream.messages is the text projection; stream.output is the final state. Consume projections independently instead of branching on chunk types.
• Pass the callback in the config so every model call in the run reports into it. get_usage_metadata_callback gives you the same as a context manager.
• usage_metadata is keyed by model name, with input/output/total tokens plus provider details (cache reads, reasoning tokens). Log it per run; it is your cost ledger.
• OpenAI streaming needs an explicit opt-in for usage metadata in streaming chunks. If usage is empty on OpenAI streams, that is the missing step.

Workflow grounded in the official event-streaming and models (token usage) docs, verified 2026-09-26. Event streaming: https://docs.langchain.com/oss/python/langchain/event-streaming ; models: https://docs.langchain.com/oss/python/langchain/models