team_streaming.py
"""
Team Streaming
==============
Demonstrates sync and async streaming responses from a team.
"""
import asyncio
from agno.agent import Agent
from agno.models.openai import OpenAIResponses
from agno.team import Team
from agno.tools.yfinance import YFinanceTools
from agno.utils.pprint import apprint_run_response
# ---------------------------------------------------------------------------
# Create Members
# ---------------------------------------------------------------------------
stock_searcher = Agent(
name="Stock Searcher",
model=OpenAIResponses(id="gpt-5-mini"),
role="Searches the web for information on a stock.",
tools=[
YFinanceTools(
enable_stock_price=True,
enable_analyst_recommendations=True,
)
],
)
company_info_agent = Agent(
name="Company Info Searcher",
model=OpenAIResponses(id="gpt-5-mini"),
role="Searches the web for information on a company.",
tools=[
YFinanceTools(
enable_stock_price=False,
enable_company_info=True,
enable_company_news=True,
)
],
)
# ---------------------------------------------------------------------------
# Create Team
# ---------------------------------------------------------------------------
team = Team(
name="Stock Research Team",
model=OpenAIResponses(id="gpt-5-mini"),
members=[stock_searcher, company_info_agent],
markdown=True,
show_members_responses=True,
)
# ---------------------------------------------------------------------------
# Run Team
# ---------------------------------------------------------------------------
async def streaming_with_arun() -> None:
await apprint_run_response(
team.arun(input="What is the current stock price of NVDA?", stream=True)
)
async def streaming_with_aprint_response() -> None:
await team.aprint_response("What is the current stock price of NVDA?", stream=True)
if __name__ == "__main__":
# Sync streaming
team.print_response(
"What is the current stock price of NVDA?",
stream=True,
)
team.print_response(
"What is the latest news for TSLA?",
stream=True,
show_member_responses=False,
)
# Async streaming
asyncio.run(streaming_with_arun())
# asyncio.run(streaming_with_aprint_response())
Run the Example
1
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activate
uv venv --python 3.12
.venv\Scripts\activate
2
Install dependencies
uv pip install -U agno openai yfinance
3
Export your OpenAI API key
export OPENAI_API_KEY="your_openai_api_key_here"
$Env:OPENAI_API_KEY="your_openai_api_key_here"
4
Run the example
Save the code above as
team_streaming.py, then run:python team_streaming.py