Team Events

Inspect team and member lifecycle and tool events in an async stream.

team_events.py
"""
Team Events
===========

Demonstrates monitoring team and member events in sync-like and async event streams.
"""

import asyncio
from uuid import uuid4

from agno.agent import Agent, RunEvent
from agno.models.openai import OpenAIResponses
from agno.team import Team, TeamRunEvent
from agno.tools.hackernews import HackerNewsTools
from agno.tools.websearch import WebSearchTools

# ---------------------------------------------------------------------------
# Create Members
# ---------------------------------------------------------------------------
hacker_news_agent = Agent(
    id="hacker-news-agent",
    name="Hacker News Agent",
    role="Search Hacker News for information",
    tools=[HackerNewsTools()],
    instructions=[
        "Find articles about the company in the Hacker News",
    ],
)

website_agent = Agent(
    id="website-agent",
    name="Website Agent",
    role="Search the website for information",
    model=OpenAIResponses(id="gpt-5-mini"),
    tools=[WebSearchTools()],
    instructions=[
        "Search the website for information",
    ],
)

# ---------------------------------------------------------------------------
# Create Team
# ---------------------------------------------------------------------------
user_id = str(uuid4())
team_id = str(uuid4())

company_info_team = Team(
    name="Company Info Team",
    id=team_id,
    user_id=user_id,
    model=OpenAIResponses(id="gpt-5.2"),
    members=[hacker_news_agent, website_agent],
    markdown=True,
    instructions=[
        "You are a team that finds information about a company.",
        "First search the web and Hacker News for information about the company.",
        "If you can find the company's website URL, then scrape the homepage and the about page.",
    ],
    show_members_responses=True,
    events_to_skip=[TeamRunEvent.run_started, TeamRunEvent.run_completed],
)


# ---------------------------------------------------------------------------
# Run Team
# ---------------------------------------------------------------------------
async def run_team_with_events(prompt: str) -> None:
    content_started = False
    async for run_output_event in company_info_team.arun(
        prompt,
        stream=True,
        stream_events=True,
    ):
        if run_output_event.event in [
            TeamRunEvent.run_started,
            TeamRunEvent.run_completed,
        ]:
            print(f"\nTEAM EVENT: {run_output_event.event}")

        if run_output_event.event in [TeamRunEvent.tool_call_started]:
            print(f"\nTEAM EVENT: {run_output_event.event}")
            print(f"TOOL CALL: {run_output_event.tool.tool_name}")
            print(f"TOOL CALL ARGS: {run_output_event.tool.tool_args}")

        if run_output_event.event in [TeamRunEvent.tool_call_completed]:
            print(f"\nTEAM EVENT: {run_output_event.event}")
            print(f"TOOL CALL: {run_output_event.tool.tool_name}")
            print(f"TOOL CALL RESULT: {run_output_event.tool.result}")

        if run_output_event.event in [RunEvent.tool_call_started]:
            print(f"\nMEMBER EVENT: {run_output_event.event}")
            print(f"AGENT ID: {run_output_event.agent_id}")
            print(f"TOOL CALL: {run_output_event.tool.tool_name}")
            print(f"TOOL CALL ARGS: {run_output_event.tool.tool_args}")

        if run_output_event.event in [RunEvent.tool_call_completed]:
            print(f"\nMEMBER EVENT: {run_output_event.event}")
            print(f"AGENT ID: {run_output_event.agent_id}")
            print(f"TOOL CALL: {run_output_event.tool.tool_name}")
            print(f"TOOL CALL RESULT: {run_output_event.tool.result}")

        if run_output_event.event in [TeamRunEvent.run_content]:
            if not content_started:
                print("CONTENT")
                content_started = True
            else:
                print(run_output_event.content, end="")


if __name__ == "__main__":
    # Async event streaming
    asyncio.run(
        run_team_with_events(
            "Write me a full report on everything you can find about Agno, the company building AI agent infrastructure.",
        )
    )

Print the first content chunk

The source's else branch drops the first team content chunk. Replace that final content-event branch with:

if run_output_event.event == TeamRunEvent.run_content:
    if not content_started:
        print("CONTENT")
        content_started = True
    if run_output_event.content:
        print(run_output_event.content, end="", flush=True)

events_to_skip controls which events are stored when store_events=True; it does not remove events from the live iterator. This example has no database or event-storage setting, so it only observes the stream.

HackerNewsTools reads top stories and user profiles; it does not search arbitrary company terms. WebSearchTools searches the web but does not scrape the homepage or About page requested by the instructions. Add a page-reading tool to the website member if you need the full page text.

Run the Example

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

Install dependencies

uv pip install -U agno ddgs openai

Export your OpenAI API key

export OPENAI_API_KEY="your_openai_api_key_here"

Run the example

Save the code above as team_events.py, then run:

python team_events.py

Full source: cookbook/03_teams/08_streaming/team_events.py