Running Teams

Learn how to run Agno Teams.

Run your Team by calling Team.run() or Team.arun(). Here’s how they work:

1. The team leader builds the context to send to the model (system message, user message, chat history, user memories, session state and other relevant inputs).
2. The team leader sends this context to the model.
3. The model processes the input and decides whether to use the delegate_task_to_members tool to delegate to team members, call other tools, or respond directly.
4. If delegation occurs, team members execute their tasks and return results to the team leader.
5. The team leader processes the updated context and provides a final response.
6. The team returns this final response to the caller.

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. Here are some of the core attributes:

• run_id: The id of the run.
• team_id: The id of the team.
• team_name: The name of the team.
• session_id: The id of the session.
• user_id: The id of the user.
• content: The response content.
• content_type: The type of content. In the case of structured output, this will be the class name of the pydantic model.
• reasoning_content: The reasoning content.
• messages: The list of messages sent to the model.
• metrics: The metrics of the run. For more details see Metrics.
• model: The model used for the run.
• member_responses: The list of member responses. Optional to add when store_member_responses=True on the Team.

Team members inherit the model from their parent team if no model is specified. The reasoning_model, parser_model, and output_model must be explicitly set for each team or team member. See the model inheritance example.

See detailed documentation 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 response.

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:
    if chunk.event == "TeamRunContent":
        print(chunk.content)

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.

Streaming all events

By default, when you stream a response, only the RunContent events will be streamed.

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
)

Handling Events

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

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

for event in response_stream:
    if event.event == "TeamRunContent":
        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 RunOutput object.

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

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

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

for event in response.events:
    print(event.event)

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=["TeamRunStarted"]
)

Event Types

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

Core Events

Event Type	Description
TeamRunStarted	Indicates the start of a run
TeamRunContent	Contains the model’s response text as individual chunks
TeamRunContentCompleted	Signals completion of content streaming
TeamRunIntermediateContent	Contains the model’s intermediate response text as individual chunks. This is used when output_model is set.
TeamRunCompleted	Signals successful completion of the run
TeamRunError	Indicates an error occurred during the run
TeamRunCancelled	Signals that the run was cancelled

Tool Events

Event Type	Description
TeamToolCallStarted	Indicates the start of a tool call
TeamToolCallCompleted	Signals completion of a tool call, including tool call results

Reasoning Events

Event Type	Description
TeamReasoningStarted	Indicates the start of the team’s reasoning process
TeamReasoningStep	Contains a single step in the reasoning process
TeamReasoningCompleted	Signals completion of the reasoning process

Memory Events

Event Type	Description
TeamMemoryUpdateStarted	Indicates that the team is updating its memory
TeamMemoryUpdateCompleted	Signals completion of a memory update

Session Summary Events

Event Type	Description
TeamSessionSummaryStarted	Indicates the start of session summary generation
TeamSessionSummaryCompleted	Signals completion of session summary generation

Pre-Hook Events

Event Type	Description
TeamPreHookStarted	Indicates the start of a pre-run hook
TeamPreHookCompleted	Signals completion of a pre-run hook execution

Post-Hook Events

Event Type	Description
TeamPostHookStarted	Indicates the start of a post-run hook
TeamPostHookCompleted	Signals completion of a post-run hook execution

Parser Model events

Event Type	Description
TeamParserModelResponseStarted	Indicates the start of the parser model response
TeamParserModelResponseCompleted	Signals completion of the parser model response

Output Model events

Event Type	Description
TeamOutputModelResponseStarted	Indicates the start of the output model response
TeamOutputModelResponseCompleted	Signals 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="[e-mail]",
        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="[e-mail]", session_id="session_123")

For more information see the Team 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

• View the Team reference
• View the TeamRunOutput schema
• View Team Cookbook