Run your Team by calling Team.run() or Team.arun(). Here’s how they work:
- 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).
- The team leader sends this context to the model.
- The model processes the input and decides whether to delegate to team members, call other tools, or respond directly.
- If delegation occurs, team members execute their tasks and return results to the team leader.
- The team leader processes the updated context and provides a final response.
- 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.
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.
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 == "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_intermediate_steps=True)
pprint_run_response(response)
for event in response.events:
print(event.event)
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 |
TeamRunCompleted | Signals successful completion of the run |
TeamRunError | Indicates an error occurred during the run |
TeamRunCancelled | Signals that the run was cancelled |
| 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 |
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
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",
)
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")
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")])
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
