The Team.run() function runs the team and generates a response, either as a TeamRunResponse object or a stream of TeamRunResponseEvent objects.

Many of our examples use team.print_response() which is a helper utility to print the response in the terminal. It uses team.run() under the hood.

Here’s how to run your team. The response is captured in the response and response_stream variables.

from agno.team import Team
from agno.models.openai import OpenAIChat

agent_1 = Agent(name="News Agent", role="Get the latest news")

agent_2 = Agent(name="Weather Agent", role="Get the weather for the next 7 days")

team = Team(name="News and Weather Team", mode="coordinate", members=[agent_1, agent_2])

response = team.run("What is the weather in Tokyo?")

# Synchronous execution
result = team.run("What is the weather in Tokyo?")

# Asynchronous execution
result = await team.arun("What is the weather in Tokyo?")

# Streaming responses
for chunk in team.run("What is the weather in Tokyo?", stream=True):
    print(chunk.content, end="", flush=True)

# Asynchronous streaming
async for chunk in await team.arun("What is the weather in Tokyo?", stream=True):
    print(chunk.content, end="", flush=True)

​
Streaming Intermediate Steps

Throughout the execution of a team, multiple events take place, and we provide these events in real-time for enhanced team transparency.

You can enable streaming of intermediate steps by setting stream_intermediate_steps=True.

# Stream with intermediate steps
response_stream = team.run(
    "What is the weather in Tokyo?",
    stream=True,
    stream_intermediate_steps=True
)

​
Handling Events

You can process events as they arrive by iterating over the response stream:

response_stream = team.run("Your prompt", stream=True, stream_intermediate_steps=True)

for event in response_stream:
    if event.event == "TeamRunResponseContent":
        print(f"Content: {event.content}")
    elif event.event == "TeamToolCallStarted":
        print(f"Tool call started: {event.tool}")
    elif event.event == "ToolCallStarted":
        print(f"Member tool call started: {event.tool}")
    elif event.event == "ToolCallCompleted":
        print(f"Member tool call completed: {event.tool}")
    elif event.event == "TeamReasoningStep":
        print(f"Reasoning step: {event.content}")
    ...

Team member events are yielded during team execution when a team member is being executed. You can disable this by setting stream_member_events=False.

​
Storing Events

You can store all the events that happened during a run on the RunResponse object.

from agno.team import Team
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response

team = Team(model=OpenAIChat(id="gpt-4o-mini"), members=[], store_events=True)

response = team.run("Tell me a 5 second short story about a lion", stream=True, stream_intermediate_steps=True)
pprint_run_response(response)

for event in agent.run_response.events:
    print(event.event)

By default the TeamRunResponseContentEvent and RunResponseContentEvent events are not stored. You can modify which events are skipped by setting the events_to_skip parameter.

For example:

team = Team(model=OpenAIChat(id="gpt-4o-mini"), members=[], store_events=True, events_to_skip=[TeamRunEvent.run_started.value])

​
Event Types

The following events are sent by the Team.run() and Team.arun() functions depending on team’s configuration:

​
Core Events

Event TypeDescription
TeamRunStartedIndicates the start of a run
TeamRunResponseContentContains the model’s response text as individual chunks
TeamRunCompletedSignals successful completion of the run
TeamRunErrorIndicates an error occurred during the run
TeamRunCancelledSignals that the run was cancelled

​
Tool Events

Event TypeDescription
TeamToolCallStartedIndicates the start of a tool call
TeamToolCallCompletedSignals completion of a tool call, including tool call results

​
Reasoning Events

Event TypeDescription
TeamReasoningStartedIndicates the start of the agent’s reasoning process
TeamReasoningStepContains a single step in the reasoning process
TeamReasoningCompletedSignals completion of the reasoning process

​
Memory Events

Event TypeDescription
TeamMemoryUpdateStartedIndicates that the agent is updating its memory
TeamMemoryUpdateCompletedSignals completion of a memory update

See detailed documentation in the TeamRunResponse documentation.

​
Structured Input

A team can be provided with structured input (i.e a pydantic model) by passing it in the Team.run() or Team.print_response() as the message parameter.

from typing import List

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team import Team
from agno.tools.hackernews import HackerNewsTools
from pydantic import BaseModel, Field


class ResearchTopic(BaseModel):
    """Structured research topic with specific requirements"""

    topic: str
    focus_areas: List[str] = Field(description="Specific areas to focus on")
    target_audience: str = Field(description="Who this research is for")
    sources_required: int = Field(description="Number of sources needed", default=5)


# Define agents
hackernews_agent = Agent(
    name="Hackernews Agent",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[HackerNewsTools()],
    role="Extract key insights and content from Hackernews posts",
)

team = Team(
    name="Hackernews Team",
    model=OpenAIChat(id="gpt-4o-mini"),
    members=[hackernews_agent],
    mode="collaborate",
)

team.print_response(
    message=ResearchTopic(
        topic="AI",
        focus_areas=["AI", "Machine Learning"],
        target_audience="Developers",
        sources_required=5,
    )
)