Structured Output in 'agent.stream_async' streaming method

[JayRabari17]

I’m trying to use agent.stream_async and would like to understand how to retrieve structured output (e.g. intermediate steps, tool calls, final answer) from it, similar to what stream_events provides. Could you clarify the best way to achieve this?

[zastrowm]

Can you clarify what you mean by "similar to what stream_events offers? AFAICT, that's not an api that strands supports.

[yonib05]

Hi @JayRabari17, thanks for the question, and apologies for the very delayed follow-up here.

As @zastrowm noted, stream_events isn't a Strands API (I'm guessing you were thinking of LangChain's astream_events), but everything you listed is available from stream_async today. The iterator yields intermediate events as the run progresses: event["data"] for streamed text, event["current_tool_use"] for tool calls as they happen, event["message"] each time a message is added to the conversation (including tool results), and a final event["result"] holding the complete AgentResult.

For typed structured output specifically, this landed since you asked: pass structured_output_model directly to the invocation and read the parsed model off the final result.

async for event in agent.stream_async(prompt, structured_output_model=WeatherForecast):
    if "current_tool_use" in event and event["current_tool_use"].get("name"):
        print(f"tool: {event['current_tool_use']['name']}")
    elif "data" in event:
        print(event["data"], end="")
    elif "result" in event:
        forecast = event["result"].structured_output  # WeatherForecast instance

So intermediate steps and tool calls come through as events during the stream, and the validated final answer arrives on the result event. Note that the older Agent.structured_output() method is deprecated in favor of this parameter.

Full event reference is in the async iterators docs, and the streaming example above is adapted from the structured output docs. Hope this gets you going!