Skip to main content
Run your Team by calling Team.run() or Team.arun(). Below is a flowchart that explains how the team runs:
Team execution flow
  1. Pre-hooks execute (if configured) to perform validation or setup before the run starts.
  2. Reasoning agent runs (if enabled) to plan and break down the task.
  3. Context is built including system message, user message, chat history, user memories, session state, and other inputs.
  4. Model is invoked with the prepared context.
  5. Model decides whether to respond directly, call provided tools, or delegate requests to team members.
  6. If delegation occurs, member agents execute their tasks concurrently (in async mode) and return results to the team leader. The team-leader model processes these results and may delegate further or respond.
  7. Response is processed and parsed into an output_schema if provided.
  8. Post-hooks execute (if configured) to perform final validation or transformation of the final output.
  9. Session and metrics are stored in the database (if configured).
  10. TeamRunOutput is returned to the caller with the final response.
In the case of streaming responses, the team leader will stream responses from the model and from members to the caller. See the Streaming section for more details.

​
Basic Execution

The Team.run() function runs the team and returns the output — either as a TeamRunOutput object or as a stream of TeamRunOutputEvent and RunOutputEvent (for member agents) objects (when stream=True). For example: 
from agno.team import Team
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response

news_agent = Agent(
    name="News Agent",
    model=OpenAIChat(id="gpt-4o"),
    role="Get the latest news",
    tools=[DuckDuckGoTools()]
)
weather_agent = Agent(
    name="Weather Agent",
    model=OpenAIChat(id="gpt-4o"),
    role="Get the weather for the next 7 days",
    tools=[DuckDuckGoTools()]
)

team = Team(
    name="News and Weather Team",
    members=[news_agent, weather_agent],
    model=OpenAIChat(id="gpt-4o")
)

# Run team and return the response as a variable
response = team.run(input="What is the weather in Tokyo?")
# Print the response in markdown format
pprint_run_response(response, markdown=True)
You can also run the team asynchronously using Team.arun(). This means members will run concurrently if the team leader delegates to multiple members in one request.
See the Input & Output docs for more information, and to see how to use structured input and output with teams.

​
Run Output

The Team.run() function returns a TeamRunOutput object when not streaming. This object contains the output content, the list of messages sent to the model, the metrics of the run, the model used for the run, and an optional list of member responses. See the detailed schema in the TeamRunOutput documentation.

​
Streaming

To enable streaming, set stream=True when calling run(). This will return an iterator of TeamRunOutputEvent objects instead of a single TeamRunOutput object. 
from typing import Iterator
from agno.team import Team
from agno.agent import Agent
from agno.models.openai import OpenAIChat

news_agent = Agent(name="News Agent", role="Get the latest news")
weather_agent = Agent(name="Weather Agent", role="Get the weather for the next 7 days")

team = Team(
    name="News and Weather Team",
    members=[news_agent, weather_agent],
    model=OpenAIChat(id="gpt-4o")
)

# Run team and return the response as a stream
stream: Iterator[TeamRunOutputEvent] = team.run("What is the weather in Tokyo?", stream=True)
for chunk in stream:
    print(chunk.content, end="", flush=True)

​
Streaming team events

When you stream a response, only the TeamRunContent events will be streamed by default. You can also stream all run events by setting stream_events=True. This will provide real-time updates about the team’s internal processes, like tool calling or reasoning: 
# Stream all events
response_stream = team.run(
    "What is the weather in Tokyo?",
    stream=True,
    stream_events=True
)

​
Streaming member events

When streaming events with stream_events=True, the team leader also streams the events from team members to the caller. When your team is running asynchronously (using arun), the members will run concurrently if the team leader delegates to multiple members in one request. This means you will receive member events concurrently and the order of the events is not guaranteed. You can disable this by setting stream_member_events=False. 
...

team = Team(
    name="News and Weather Team",
    members=[news_agent, weather_agent],
    model=OpenAIChat(id="gpt-4o"),
    stream_member_events=False
)

response_stream = team.run(
    "What is the weather in Tokyo?",
    stream=True,
    stream_events=True)

​
Handling Events

You can process events as they arrive by iterating over the response stream: 
from agno.team import Team
from agno.run.team import TeamRunEvent
from agno.run.agent import RunEvent
from agno.models.openai import OpenAIChat
from agno.agent import Agent
from agno.tools.duckduckgo import DuckDuckGoTools

weather_agent = Agent(name="Weather Agent", role="Get the weather for the next 7 days", tools=[DuckDuckGoTools()])
team = Team(
    name="Test Team",
    members=[weather_agent],
    model=OpenAIChat(id="gpt-4o-mini")
)

response_stream = team.run("What is the weather in Tokyo?", stream=True, stream_events=True)

for event in response_stream:
    if event.event == TeamRunEvent.run_content:
        print(event.content, end="", flush=True)
    elif event.event == TeamRunEvent.tool_call_started:
        print(f"Team tool call started")
    elif event.event == TeamRunEvent.tool_call_completed:
        print(f"Team tool call completed")
    elif event.event == RunEvent.tool_call_started:
        print(f"Member tool call started")
    elif event.event == RunEvent.tool_call_completed:
        print(f"Member tool call completed")
    elif event.event == TeamRunEvent.run_started:
        print(f"Run started")
    elif event.event == TeamRunEvent.run_completed:
        print(f"Run completed")

​
Storing Events

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

team = Team(
    name="Story Team",
    members=[],
    model=OpenAIChat(id="gpt-4o"),
    store_events=True
)
By default the TeamRunContentEvent and RunContentEvent events are not stored. You can modify which events are skipped by setting the events_to_skip parameter. For example: 
team = Team(
    name="Story Team",
    members=[],
    model=OpenAIChat(id="gpt-4o"),
    store_events=True,
    events_to_skip=[]  # Include all events
)
See the full TeamRunOutput schema for more details.

​
Event Types

The following events are streamed when stream_events=True by the Team.run() and Team.arun() functions depending on team’s configuration:

​
Core Events

Event TypeDescription
TeamRunStartedIndicates the start of a run
TeamRunContentContains the model’s response text as individual chunks
TeamRunContentCompletedSignals completion of content streaming
TeamRunIntermediateContentContains the model’s intermediate response text as individual chunks. This is used when output_model is set.
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 team’s reasoning process
TeamReasoningStepContains a single step in the reasoning process
TeamReasoningCompletedSignals completion of the reasoning process

​
Memory Events

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

​
Session Summary Events

Event TypeDescription
TeamSessionSummaryStartedIndicates the start of session summary generation
TeamSessionSummaryCompletedSignals completion of session summary generation

​
Pre-Hook Events

Event TypeDescription
TeamPreHookStartedIndicates the start of a pre-run hook
TeamPreHookCompletedSignals completion of a pre-run hook execution

​
Post-Hook Events

Event TypeDescription
TeamPostHookStartedIndicates the start of a post-run hook
TeamPostHookCompletedSignals completion of a post-run hook execution

​
Parser Model events

Event TypeDescription
TeamParserModelResponseStartedIndicates the start of the parser model response
TeamParserModelResponseCompletedSignals completion of the parser model response

​
Output Model events

Event TypeDescription
TeamOutputModelResponseStartedIndicates the start of the output model response
TeamOutputModelResponseCompletedSignals completion of the output model response
See detailed documentation in the TeamRunOutput documentation.

​
Custom Events

If you are using your own custom tools, it will often be useful to be able to yield custom events. Your custom events will be yielded together with the rest of the expected Agno events. We recommend creating your custom event class extending the built-in CustomEvent class: 
from dataclasses import dataclass
from agno.run.team import CustomEvent

@dataclass
class CustomerProfileEvent(CustomEvent):
    """CustomEvent for customer profile."""

    customer_name: Optional[str] = None
    customer_email: Optional[str] = None
    customer_phone: Optional[str] = None
You can then yield your custom event from your tool. The event will be handled internally as an Agno event, and you will be able to access it in the same way you would access any other Agno event. 
from agno.tools import tool

@tool()
async def get_customer_profile():
    """Example custom tool that simply yields a custom event."""

    yield CustomerProfileEvent(
        customer_name="John Doe",
        customer_email="john.doe@example.com",
        customer_phone="1234567890",
    )
See the full example for more details.

​
Specify Run User and Session

You can specify which user and session to use when running the team by passing the user_id and session_id parameters. This ensures the current run is associated with the correct user and session. For example: 
team.run("Get me my monthly report", user_id="john@example.com", session_id="session_123")
For more information see the Sessions documentation.

​
Passing Images / Audio / Video / Files

You can pass images, audio, video, or files to the team by passing the images, audio, video, or files parameters. For example: 
team.run("Tell me a 5 second short story about this image", images=[Image(url="https://example.com/image.jpg")])
For more information see the Multimodal documentation.

​
Cancelling a Run

A run can be cancelled by calling the Team.cancel_run() method. See more details in the Cancelling a Run documentation.

​
Developer Resources