Team API

from agno.team import Team
from agno.agent import Agent

class Team:
    def __init__(
        self,
        *,
        members: Optional[List[Union[Agent, Team]]] = None,
        leader: Optional[Agent] = None,
        id: Optional[str] = None,
        name: Optional[str] = None,
        model: Optional[Union[Model, str]] = None,
        role: Optional[str] = None,
        description: Optional[str] = None,
        instructions: Optional[Union[str, List[str]]] = None,
        respond_directly: bool = False,
        determine_input_for_members: bool = False,
        delegate_task_to_all_members: bool = False,
        db: Optional[Union[BaseDb, AsyncBaseDb]] = None,
        knowledge: Optional[Knowledge] = None,
        memory_manager: Optional[MemoryManager] = None,
        session_id: Optional[str] = None,
        user_id: Optional[str] = None,
        add_history_to_context: bool = False,
        stream: Optional[bool] = None,
        **kwargs
    ): ...

def run(
    self,
    input: Union[str, List, Dict, Message, List[Message]],
    *,
    stream: Optional[bool] = None,
    session_id: Optional[str] = None,
    user_id: Optional[str] = None,
    **kwargs
) -> Union[TeamRunOutput, Iterator[Union[TeamRunOutputEvent, TeamRunOutput]]]:
    """
    Execute the team with the given input.
    
    Parameters:
        input: Text, messages, or structured input
        stream: Enable streaming responses
        session_id: Session identifier
        user_id: User identifier
        **kwargs: Additional parameters
    
    Returns:
        TeamRunOutput or Iterator: Team response
    """

async def arun(
    self,
    input: Union[str, List, Dict, Message, List[Message]],
    *,
    stream: Optional[bool] = None,
    **kwargs
) -> Union[TeamRunOutput, AsyncIterator[Union[TeamRunOutputEvent, TeamRunOutput]]]:
    """Async version of run()."""

@dataclass
class TeamRunOutput:
    """Response from team execution"""
    
    run_id: Optional[str]
    team_id: Optional[str]
    session_id: Optional[str]
    user_id: Optional[str]
    
    content: Optional[Any]
    messages: Optional[List[Message]]
    metrics: Optional[Metrics]
    
    status: RunStatus

from agno.agent import Agent
from agno.team import Team
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools

# Create specialized agents
researcher = Agent(
    name="Researcher",
    role="Research and gather information",
    model=OpenAIChat(id="gpt-4"),
    tools=[DuckDuckGoTools()],
    instructions=["Search for accurate information"]
)

analyst = Agent(
    name="Analyst",
    role="Analyze data and create insights",
    model=OpenAIChat(id="gpt-4"),
    tools=[YFinanceTools()],
    instructions=["Analyze data thoroughly"]
)

writer = Agent(
    name="Writer",
    role="Write clear, concise summaries",
    model=OpenAIChat(id="gpt-4"),
    instructions=["Create well-structured reports"]
)

# Create team
team = Team(
    members=[researcher, analyst, writer],
    leader=writer,
    name="Research Team",
    description="Collaborative research and analysis"
)

response = team.run("Research and analyze the latest AI trends")
print(response.content)