Structured Output Streaming

This example demonstrates streaming structured output from a team, using Pydantic models to ensure validated data structures while providing real-time streaming responses.

Code

cookbook/examples/teams/structured_input_output/04_structured_output_streaming.py

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team import Team
from agno.tools.exa import ExaTools
from pydantic import BaseModel


class StockAnalysis(BaseModel):
    symbol: str
    company_name: str
    analysis: str


stock_searcher = Agent(
    name="Stock Searcher",
    model=OpenAIChat("gpt-5-mini"),
    output_schema=StockAnalysis,
    role="Searches the web for information on a stock.",
    tools=[
        ExaTools(
            include_domains=["cnbc.com", "reuters.com", "bloomberg.com", "wsj.com"],
            text=False,
            highlights=False,
            show_results=True,
        )
    ],
)


class CompanyAnalysis(BaseModel):
    company_name: str
    analysis: str


company_info_agent = Agent(
    name="Company Info Searcher",
    model=OpenAIChat("gpt-5-mini"),
    role="Searches the web for information on a stock.",
    output_schema=CompanyAnalysis,
    tools=[
        ExaTools(
            include_domains=["cnbc.com", "reuters.com", "bloomberg.com", "wsj.com"],
            text=False,
            highlights=False,
            show_results=True,
        )
    ],
)


class StockReport(BaseModel):
    symbol: str
    company_name: str
    analysis: str


team = Team(
    name="Stock Research Team",
    model=OpenAIChat("gpt-5-mini"),
    members=[stock_searcher, company_info_agent],
    output_schema=StockReport,
    markdown=True,
    show_members_responses=True,
)

# Run with streaming and consume the generator to get the final response
stream_generator = team.run(
    "Give me a stock report for NVDA",
    stream=True,
    stream_events=True,
)

# Consume the streaming events and get the final response
run_response = None
for event_or_response in stream_generator:
    # The last item in the stream is the final TeamRunOutput
    run_response = event_or_response

assert isinstance(run_response.content, StockReport)
print(
    f"✅ Response content is correctly typed as StockReport: {type(run_response.content)}"
)
print(f"✅ Stock Symbol: {run_response.content.symbol}")
print(f"✅ Company Name: {run_response.content.company_name}")

Usage

1. Create a virtual environment

   Open the Terminal and create a python virtual environment.

   Mac

   python3 -m venv .venv
   source .venv/bin/activate
   
2. Install required libraries

   pip install agno exa_py pydantic
   
3. Set environment variables

   export OPENAI_API_KEY=****
   export EXA_API_KEY=****
   
4. Run the agent

   python cookbook/examples/teams/structured_input_output/04_structured_output_streaming.py