Agno

Agno Skill

Build production-ready AI agents with Agno - a lightweight, model-agnostic framework for agents, teams, workflows, and MCP integration.

When to Use This Skill

This skill should be triggered when:

• Building AI agents with tools, memory, structured outputs, or knowledge
• Creating multi-agent teams with role-based delegation
• Implementing workflows with sequential, parallel, conditional, or routing steps
• Integrating MCP servers (stdio, SSE, or Streamable HTTP)
• Deploying agents with AgentOS (FastAPI-based runtime)
• Working with the LearningMachine (user profiles, entity memory, session context)
• Debugging agent behavior or optimizing performance

Architecture Overview

Agent          - Single autonomous AI unit (model + tools + instructions)
Team           - Multiple agents coordinated by a leader (route/broadcast/tasks modes)
Workflow       - Pipeline-based execution (Step, Parallel, Condition, Loop, Router)
AgentOS        - FastAPI runtime for deploying agents as production APIs
LearningMachine - Persistent learning across sessions (profiles, memory, knowledge)

Quick Reference

1. Basic Agent with Tools

from agno.agent import Agent
from agno.models.google import Gemini
from agno.tools.yfinance import YFinanceTools

agent = Agent(
    name="Finance Agent",
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
    add_datetime_to_context=True,
    markdown=True,
)

agent.print_response("Give me a quick brief on NVIDIA", stream=True)

2. Structured Output with Pydantic

from typing import List, Optional
from agno.agent import Agent
from agno.models.google import Gemini
from agno.tools.yfinance import YFinanceTools
from pydantic import BaseModel, Field

class StockAnalysis(BaseModel):
    ticker: str = Field(..., description="Stock ticker symbol")
    company_name: str = Field(..., description="Full company name")
    current_price: float = Field(..., description="Current price in USD")
    summary: str = Field(..., description="One-line summary")
    key_drivers: List[str] = Field(..., description="2-3 key growth drivers")
    recommendation: str = Field(..., description="Buy, Hold, or Sell")

agent = Agent(
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
    output_schema=StockAnalysis,
)

response = agent.run("Analyze NVIDIA")
analysis: StockAnalysis = response.content
print(f"{analysis.company_name}: {analysis.recommendation}")

3. Agent with Storage (Session Persistence)

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.google import Gemini

agent = Agent(
    model=Gemini(id="gemini-3-flash-preview"),
    db=SqliteDb(db_file="tmp/agents.db"),
    add_history_to_context=True,
    num_history_runs=5,
    markdown=True,
)

# Same session_id = continuous conversation across runs
agent.print_response("Analyze NVDA", session_id="my-session", stream=True)
agent.print_response("Compare that to Tesla", session_id="my-session", stream=True)

4. Agent with Memory (User Preferences)

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.memory import MemoryManager
from agno.models.google import Gemini

db = SqliteDb(db_file="tmp/agents.db")

agent = Agent(
    model=Gemini(id="gemini-3-flash-preview"),
    db=db,
    memory_manager=MemoryManager(
        model=Gemini(id="gemini-3-flash-preview"),
        db=db,
    ),
    enable_agentic_memory=True,  # Agent decides when to store/recall
    markdown=True,
)

# Agent remembers user preferences across sessions
agent.print_response(
    "I'm interested in AI stocks. My risk tolerance is moderate.",
    user_id="alice@example.com",
    stream=True,
)

5. Multi-Agent Team

from agno.agent import Agent
from agno.models.google import Gemini
from agno.team.team import Team
from agno.tools.yfinance import YFinanceTools

bull = Agent(
    name="Bull Analyst",
    role="Make the investment case FOR a stock",
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
)

bear = Agent(
    name="Bear Analyst",
    role="Make the investment case AGAINST a stock",
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
)

team = Team(
    name="Investment Research",
    model=Gemini(id="gemini-3-flash-preview"),
    members=[bull, bear],
    instructions=["Get both perspectives, then synthesize a balanced recommendation"],
    show_members_responses=True,
    markdown=True,
)

team.print_response("Should I invest in NVIDIA?", stream=True)

6. Sequential Workflow

from agno.agent import Agent
from agno.models.google import Gemini
from agno.tools.yfinance import YFinanceTools
from agno.workflow import Step, Workflow

data_agent = Agent(name="Data Gatherer", model=Gemini(id="gemini-3-flash-preview"), tools=[YFinanceTools()])
analyst = Agent(name="Analyst", model=Gemini(id="gemini-3-flash-preview"))
writer = Agent(name="Report Writer", model=Gemini(id="gemini-3-flash-preview"), markdown=True)

workflow = Workflow(
    name="Research Pipeline",
    steps=[
        Step(name="Gather Data", agent=data_agent),
        Step(name="Analyze", agent=analyst),
        Step(name="Write Report", agent=writer),
    ],
)

workflow.print_response("Analyze NVIDIA for investment", stream=True)

7. MCP Server Integration (stdio)

import asyncio
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.mcp import MCPTools

async def run_agent(message: str) -> None:
    async with MCPTools(command="uvx mcp-server-git") as mcp_tools:
        agent = Agent(model=Claude(id="claude-sonnet-4-5-20250929"), tools=[mcp_tools])
        await agent.aprint_response(message, stream=True)

asyncio.run(run_agent("What is the license for this project?"))

8. MCP Server (Streamable HTTP)

import asyncio
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.mcp import MCPTools

async def run_agent(message: str) -> None:
    async with MCPTools(
        transport="streamable-http",
        url="https://docs.agno.com/mcp",
    ) as mcp_tools:
        agent = Agent(model=Claude(id="claude-sonnet-4-5-20250929"), tools=[mcp_tools], markdown=True)
        await agent.aprint_response(message, stream=True)

asyncio.run(run_agent("What is Agno?"))

9. Multiple MCP Servers

import asyncio
from os import getenv
from agno.agent import Agent
from agno.tools.mcp import MultiMCPTools

async def run_agent(message: str) -> None:
    mcp_tools = MultiMCPTools(
        commands=["npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt"],
        urls=["http://localhost:8000/mcp"],
        urls_transports=["streamable-http"],
        timeout_seconds=30,
    )
    await mcp_tools.connect()
    agent = Agent(tools=[mcp_tools], markdown=True)
    await agent.aprint_response(message, stream=True)
    await mcp_tools.close()

asyncio.run(run_agent("Find listings in Barcelona"))

10. LearningMachine (Persistent Learning)

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.learn import LearningMachine, LearningMode, UserProfileConfig
from agno.models.openai import OpenAIResponses

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    db=db,
    learning=LearningMachine(
        user_profile=UserProfileConfig(mode=LearningMode.ALWAYS),
    ),
    markdown=True,
)

agent.print_response("Hi! I'm Alice, call me Ali.", user_id="alice@example.com", stream=True)
# Profile fields (name, preferred_name) captured automatically

Key Patterns

Pattern: MCP Connection Lifecycle

Always close MCP connections. Use async context managers or try/finally:

# Preferred: context manager
async with MCPTools(command="uvx mcp-server-git") as tools:
    agent = Agent(tools=[tools])
    await agent.aprint_response("query")

# Alternative: manual lifecycle
tools = MCPTools(command="uvx mcp-server-git")
await tools.connect()
try:
    agent = Agent(tools=[tools])
    await agent.aprint_response("query")
finally:
    await tools.close()

Pattern: Production Database (PostgreSQL)

from agno.db.postgres import PostgresDb
db = PostgresDb(db_url="postgresql+psycopg://user:pass@localhost:5432/agno")
agent = Agent(db=db, add_history_to_context=True)

Pattern: Debug Mode

agent = Agent(debug_mode=True)  # Detailed logs of messages, tools, tokens

Pattern: Custom Tools

from agno.tools.decorator import tool

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"Weather in {city}: 72F, sunny"

agent = Agent(tools=[get_weather])

Important Rules

• Never create agents in loops - reuse agents for performance
• Use `output_schema` for structured responses (not free-form parsing)
• PostgreSQL for production, SQLite only for development
• Both sync and async - all public methods have async variants (prefix with a)
• Always close MCP connections - use try/finally or async context managers
• Enable `debug_mode=True` when troubleshooting

Reference Files

Detailed documentation is available in references/:

• agents.md - Agent parameters, configuration, tools, memory, knowledge, guardrails
• teams.md - Team modes (route/broadcast/tasks), member coordination
• workflows.md - Step types (Step, Parallel, Condition, Loop, Router)
• mcp.md - MCP integration (stdio, SSE, Streamable HTTP), MultiMCPTools
• tools.md - Built-in tools list, custom tool creation, tool hooks
• learning.md - LearningMachine stores (profile, memory, session, knowledge, entity)
• models.md - Supported model providers and configuration

Resources

• Documentation: https://docs.agno.com
• GitHub: https://github.com/agno-agi/agno
• Cookbook Examples: https://github.com/agno-agi/agno/tree/main/cookbook
• Install: pip install agno

Agent Reference

Creating an Agent

from agno.agent import Agent

agent = Agent(
    # --- Identity ---
    name="My Agent",                    # Display name
    id="my-agent",                      # Unique identifier
    model="openai:gpt-4o",             # Model (string shorthand or Model instance)

    # --- Instructions ---
    description="Agent description",    # Added to system message
    instructions=["Rule 1", "Rule 2"], # List of strings or single string
    system_message="Full override",     # Replaces auto-generated system message
    expected_output="Format spec",      # Output format guidance
    additional_context="Extra info",    # Appended to system message

    # --- Tools ---
    tools=[YFinanceTools()],           # List of Toolkit, Callable, or Function
    tool_call_limit=10,                # Max tool calls per run
    tool_choice="auto",                # "auto", "none", "required", or {"type": "function", "function": {"name": "..."}}
    tool_hooks=[my_hook],              # Hooks called on tool execution

    # --- Structured Output ---
    output_schema=MyPydanticModel,     # Pydantic model for typed responses
    structured_outputs=True,           # Use native structured outputs (provider support required)
    use_json_mode=False,               # Force JSON mode

    # --- Session & Storage ---
    db=SqliteDb(db_file="agents.db"),  # Database for session persistence
    session_id="session-123",          # Persistent session identifier
    user_id="user@example.com",        # User identifier for memory/learning
    add_history_to_context=True,       # Include conversation history
    num_history_runs=5,                # Number of past runs to include

    # --- Memory ---
    memory_manager=MemoryManager(...), # User memory manager
    enable_agentic_memory=True,        # Agent decides when to store/recall (efficient)
    update_memory_on_run=False,        # Auto-extract after every run (guaranteed but costly)

    # --- Knowledge (RAG) ---
    knowledge=knowledge_base,          # KnowledgeBase instance
    add_knowledge_to_context=True,     # Add retrieved docs to context

    # --- Learning ---
    learning=LearningMachine(...),     # Or learning=True for defaults

    # --- State ---
    session_state={"key": "value"},    # Shared state dict
    add_session_state_to_context=True, # Include state in context

    # --- Reasoning ---
    reasoning=True,                    # Enable chain-of-thought
    reasoning_model=Model(...),        # Separate model for reasoning
    reasoning_min_steps=1,
    reasoning_max_steps=10,

    # --- Hooks & Guardrails ---
    pre_hooks=[guardrail_fn],          # Run before agent response
    post_hooks=[eval_fn],              # Run after agent response

    # --- Context Enrichment ---
    add_datetime_to_context=True,      # Add current date/time
    add_location_to_context=False,     # Add user location
    add_name_to_context=False,         # Add agent name

    # --- Retry & Reliability ---
    retries=0,                         # Number of retries on failure
    delay_between_retries=1,           # Seconds between retries
    exponential_backoff=False,         # Exponential backoff on retries

    # --- Streaming ---
    stream=True,                       # Enable streaming
    stream_events=False,               # Enable event-based streaming

    # --- Debug ---
    debug_mode=False,                  # Detailed logging
    telemetry=True,                    # Usage telemetry (set False to disable)
    markdown=True,                     # Format output as markdown
)

Key Methods

run() / arun()

Execute the agent and get a RunOutput.

# Synchronous
response = agent.run("Your message")
print(response.content)  # String or Pydantic model if output_schema set

# Asynchronous
response = await agent.arun("Your message")

# With streaming
for chunk in agent.run("Your message", stream=True):
    print(chunk)

# With multimodal inputs
from agno.media import Image
response = agent.run(
    "Describe this image",
    images=[Image(url="https://example.com/photo.jpg")],
)

# Override parameters per-run
response = agent.run(
    "Your message",
    session_id="custom-session",
    user_id="user@example.com",
    debug_mode=True,
)

print_response() / aprint_response()

Execute and print formatted output to console.

# Basic
agent.print_response("Your message", stream=True)

# Async
await agent.aprint_response("Your message", stream=True)

# With options
agent.print_response(
    "Your message",
    stream=True,
    markdown=True,
    show_reasoning=True,
    session_id="my-session",
    user_id="user@example.com",
)

Memory Methods

# Get user memories
memories = agent.get_user_memories(user_id="user@example.com")

RunOutput

The response object from agent.run():

response = agent.run("message")

response.content          # str or BaseModel (if output_schema)
response.messages         # List of messages exchanged
response.metrics          # Token usage, timing, etc.
response.run_id           # Unique run identifier
response.session_id       # Session identifier

Input Types

Agents accept flexible input:

# String
agent.run("Hello")

# Message object
from agno.models.message import Message
agent.run(Message(role="user", content="Hello"))

# List of messages
agent.run([
    Message(role="user", content="Hello"),
    Message(role="assistant", content="Hi!"),
    Message(role="user", content="Follow up"),
])

# Dict
agent.run({"role": "user", "content": "Hello"})

# Pydantic model (when using input schema)
agent.run(MyInputModel(field="value"))

Multimodal Support

from agno.media import Audio, Image, Video, File

agent.run("Describe this", images=[Image(url="https://...")])
agent.run("Transcribe this", audio=[Audio(filepath="audio.mp3")])
agent.run("Analyze this", videos=[Video(url="https://...")])
agent.run("Read this", files=[File(filepath="doc.pdf")])

Learning Reference

Overview

The LearningMachine provides persistent learning across sessions with 5 stores:

Imports

from agno.learn import (
    LearningMachine,
    LearningMode,
    UserProfileConfig,
    UserMemoryConfig,
    SessionContextConfig,
    EntityMemoryConfig,
    LearnedKnowledgeConfig,
)

Learning Modes

class LearningMode(Enum):
    ALWAYS = "always"    # Auto-extract after each response (invisible to agent)
    AGENTIC = "agentic"  # Agent decides when to learn via tool calls
    PROPOSE = "propose"  # Agent proposes, human confirms

Basic Setup

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.learn import LearningMachine, LearningMode, UserProfileConfig, UserMemoryConfig

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    db=db,
    learning=LearningMachine(
        user_profile=UserProfileConfig(mode=LearningMode.ALWAYS),
        user_memory=UserMemoryConfig(mode=LearningMode.AGENTIC),
    ),
)

Store Configurations

User Profile

Captures structured profile fields (name, preferred name, custom fields).

UserProfileConfig(
    mode=LearningMode.ALWAYS,          # Auto-extract after each response
    db=db,                             # Override default db
    model=model,                       # Override default model
    schema=CustomProfileSchema,        # Custom Pydantic schema for fields
    enable_update_profile=True,        # Allow profile updates
    instructions="Focus on preferences and work details",
)

User Memory

Stores unstructured observations about users.

UserMemoryConfig(
    mode=LearningMode.AGENTIC,        # Agent decides when to store
    enable_add_memory=True,
    enable_update_memory=True,
    enable_delete_memory=True,
    enable_clear_memories=False,       # Safety: don't allow bulk delete
    instructions="Remember important facts and preferences",
)

Session Context

Tracks current session state (goal, plan, progress).

SessionContextConfig(
    mode=LearningMode.ALWAYS,
    enable_planning=True,              # Track goals and plans
    enable_add_context=True,
    enable_update_context=True,
)

Entity Memory

Facts about third-party entities (people, companies, projects).

EntityMemoryConfig(
    mode=LearningMode.ALWAYS,
    namespace="global",                # Shared across users
    enable_create_entity=True,
    enable_add_fact=True,
    enable_add_event=True,
    enable_add_relationship=True,
)

Learned Knowledge

Reusable patterns and insights across users (requires knowledge base with vector DB).

LearnedKnowledgeConfig(
    mode=LearningMode.AGENTIC,
    knowledge=knowledge_base,          # Vector knowledge base
    namespace="global",
    agent_can_save=True,
    agent_can_search=True,
)

Full Example

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.learn import (
    LearningMachine, LearningMode,
    UserProfileConfig, UserMemoryConfig,
    SessionContextConfig, EntityMemoryConfig,
)
from agno.models.openai import OpenAIResponses

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    db=db,
    learning=LearningMachine(
        user_profile=UserProfileConfig(mode=LearningMode.ALWAYS),
        user_memory=UserMemoryConfig(mode=LearningMode.AGENTIC),
        session_context=SessionContextConfig(mode=LearningMode.ALWAYS, enable_planning=True),
        entity_memory=EntityMemoryConfig(mode=LearningMode.ALWAYS),
    ),
    markdown=True,
)

user_id = "alice@example.com"

# Session 1: Agent learns about user
agent.print_response(
    "Hi! I'm Alice Chen, call me Ali. I work at TechCorp as a data scientist.",
    user_id=user_id,
    session_id="session_1",
    stream=True,
)

# Access stored data
agent.learning_machine.user_profile_store.print(user_id=user_id)

# Session 2: Agent recalls everything
agent.print_response(
    "What do you remember about me?",
    user_id=user_id,
    session_id="session_2",
    stream=True,
)

Quick Enable (Defaults)

For simple cases, just pass learning=True:

agent = Agent(
    model=model,
    db=db,
    learning=True,  # Enables all stores with defaults
)

Accessing Stores

# Print stored data
agent.learning_machine.user_profile_store.print(user_id=user_id)
agent.learning_machine.user_memory_store.print(user_id=user_id)
agent.learning_machine.session_context_store.print(user_id=user_id, session_id=session_id)
agent.learning_machine.entity_memory_store.print(user_id=user_id)

MCP Integration Reference

Imports

from agno.tools.mcp import MCPTools, MultiMCPTools

Transport Types

MCPTools - Single Server

stdio Transport (default for commands)

import asyncio
from agno.agent import Agent
from agno.tools.mcp import MCPTools

async def run():
    async with MCPTools(command="uvx mcp-server-git") as tools:
        agent = Agent(tools=[tools])
        await agent.aprint_response("What's the project license?", stream=True)

asyncio.run(run())

Streamable HTTP Transport

async def run():
    async with MCPTools(
        transport="streamable-http",
        url="https://docs.agno.com/mcp",
    ) as tools:
        agent = Agent(tools=[tools], markdown=True)
        await agent.aprint_response("What is Agno?", stream=True)

asyncio.run(run())

Manual Connection Lifecycle

async def run():
    tools = MCPTools(command="uvx mcp-server-git")
    await tools.connect()

    try:
        agent = Agent(tools=[tools])
        await agent.aprint_response("query", stream=True)
    finally:
        await tools.close()

MCPTools Constructor

MCPTools(
    command="uvx mcp-server-git",      # stdio command (auto-detects stdio transport)
    url="http://localhost:8000/mcp",   # HTTP/SSE URL
    transport="streamable-http",       # "stdio", "sse", "streamable-http"
    env={"API_KEY": "..."},            # Environment variables for subprocess
    timeout_seconds=10,                # Read timeout
    include_tools=["tool1", "tool2"],  # Only include specific tools
    exclude_tools=["tool3"],           # Exclude specific tools
    tool_name_prefix="myserver",       # Prefix tool names (avoid collisions)
    refresh_connection=False,          # Refresh connection per agent run
    header_provider=lambda: {"Authorization": f"Bearer {get_token()}"},  # Dynamic headers
)

MultiMCPTools - Multiple Servers

Connect to multiple MCP servers simultaneously:

from agno.tools.mcp import MultiMCPTools

async def run():
    tools = MultiMCPTools(
        # stdio servers (commands)
        commands=[
            "npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt",
            "npx -y @modelcontextprotocol/server-brave-search",
        ],
        # HTTP servers (urls)
        urls=["http://localhost:8000/mcp"],
        urls_transports=["streamable-http"],
        # Shared config
        env={"BRAVE_API_KEY": os.getenv("BRAVE_API_KEY")},
        timeout_seconds=30,
    )
    await tools.connect()

    agent = Agent(tools=[tools], markdown=True)
    await agent.aprint_response("Find listings in Barcelona", stream=True)
    await tools.close()

MultiMCPTools Constructor

MultiMCPTools(
    commands=["cmd1", "cmd2"],         # List of stdio commands
    urls=["http://..."],               # List of HTTP/SSE URLs
    urls_transports=["streamable-http"],  # Transport per URL
    env={"KEY": "value"},              # Shared environment variables
    timeout_seconds=30,                # Read timeout
    include_tools=["tool1"],           # Filter tools
    exclude_tools=["tool2"],
    tool_name_prefix="prefix",
    refresh_connection=False,
)

Tool Filtering

# Only include specific tools
MCPTools(command="...", include_tools=["read_file", "write_file"])

# Exclude tools
MCPTools(command="...", exclude_tools=["delete_file"])

# Prefix tool names to avoid collisions with multiple servers
MCPTools(command="...", tool_name_prefix="git")
# Tools become: git_read_file, git_write_file, etc.

Dynamic Headers (Auth)

MCPTools(
    transport="streamable-http",
    url="https://api.example.com/mcp",
    header_provider=lambda: {
        "Authorization": f"Bearer {get_fresh_token()}"
    },
)

MCPToolbox (Toolbox Servers)

For MCP Toolbox for Databases and similar toolbox servers:

from agno.tools.mcp import MCPToolbox

toolbox = MCPToolbox(
    url="http://localhost:5000",
    toolsets=["my-toolset"],           # Filter by toolset
    transport="streamable-http",
)

Best Practices

1. Always close connections - Use async with or try/finally 2. Set reasonable timeouts - Default is 10s, increase for slow servers 3. Use tool_name_prefix with multiple servers to avoid name collisions 4. MCP is async-only - All MCP operations require async/await 5. Use refresh_connection=True if server state changes between runs

Model Providers Reference

Usage

Models can be specified as a class instance or string shorthand:

from agno.agent import Agent

# Class instance (full control)
from agno.models.openai import OpenAIChat
agent = Agent(model=OpenAIChat(id="gpt-4o"))

# String shorthand
agent = Agent(model="openai:gpt-4o")

Supported Providers

Tier 1 - Major Cloud Providers

Tier 2 - Inference Providers

Tier 3 - Local & Self-hosted

Tier 4 - Routing & Proxy

Common Model Parameters

from agno.models.openai import OpenAIChat

model = OpenAIChat(
    id="gpt-4o",                       # Model identifier
    api_key="sk-...",                   # API key (or set env var)
    temperature=0.7,                   # Sampling temperature
    max_tokens=4096,                   # Max output tokens
    top_p=1.0,                         # Nucleus sampling
    frequency_penalty=0.0,             # Frequency penalty
    presence_penalty=0.0,              # Presence penalty
    stop=["END"],                      # Stop sequences
)

OpenAI-Compatible Providers

Use OpenAILike for any OpenAI-compatible API:

from agno.models.openai import OpenAILike

model = OpenAILike(
    id="my-model",
    api_key="my-api-key",
    base_url="https://my-provider.com/v1",
)

Team Reference

Creating a Team

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

team = Team(
    # --- Required ---
    members=[agent1, agent2],          # List of Agent or Team instances

    # --- Identity ---
    name="My Team",
    model=Gemini(id="gemini-3-flash-preview"),  # Leader model
    role="Team leader role description",

    # --- Execution Mode ---
    mode="coordinate",                 # coordinate, route, broadcast, tasks
    respond_directly=False,            # Members respond directly to user
    max_iterations=10,                 # Max coordination loops

    # --- Instructions ---
    instructions=["Instruction 1"],
    description="Team description",

    # --- Member Coordination ---
    show_members_responses=True,       # Show individual member responses
    add_team_history_to_members=False, # Share team history with members
    share_member_interactions=False,   # Members see each other's responses

    # --- Session & Storage ---
    db=SqliteDb(db_file="agents.db"),
    session_id="team-session",
    user_id="user@example.com",
    add_history_to_context=True,
    num_history_runs=5,

    # --- Output ---
    output_schema=MyModel,             # Structured output
    markdown=True,
)

Team Modes

coordinate (default)

Supervisor pattern. Leader picks members, crafts tasks, synthesizes responses.

team = Team(members=[agent1, agent2], mode="coordinate")

route

Router pattern. Leader routes to a single specialist and returns their response directly.

team = Team(
    members=[finance_agent, legal_agent, tech_agent],
    mode="route",
)
# Leader picks the best-fit agent for each query

broadcast

Fan-out pattern. Leader sends the same task to all members simultaneously.

team = Team(
    members=[bull_agent, bear_agent],
    mode="broadcast",
)
# All members process the task, leader synthesizes

tasks

Autonomous task decomposition. Leader breaks goals into tasks, delegates to members, loops until complete.

team = Team(
    members=[researcher, analyst, writer],
    mode="tasks",
    max_iterations=10,
)
# Leader creates task list, assigns to members, tracks completion

Key Methods

# Synchronous
response = team.run("Your message")
team.print_response("Your message", stream=True)

# Asynchronous
response = await team.arun("Your message")
await team.aprint_response("Your message", stream=True)

Nested Teams

Teams can contain other teams as members:

research_team = Team(
    name="Research Team",
    members=[web_agent, arxiv_agent],
    mode="broadcast",
)

analysis_team = Team(
    name="Analysis Team",
    members=[data_agent, viz_agent],
    mode="coordinate",
)

main_team = Team(
    name="Main Team",
    members=[research_team, analysis_team],
    mode="coordinate",
)

Example: Investment Research Team

from agno.agent import Agent
from agno.team.team import Team
from agno.models.google import Gemini
from agno.tools.yfinance import YFinanceTools

bull = Agent(
    name="Bull Analyst",
    role="Make the investment case FOR a stock",
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
)

bear = Agent(
    name="Bear Analyst",
    role="Make the investment case AGAINST a stock",
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
)

team = Team(
    name="Investment Research",
    model=Gemini(id="gemini-3-flash-preview"),
    members=[bull, bear],
    mode="broadcast",
    show_members_responses=True,
    markdown=True,
)

team.print_response("Should I invest in NVIDIA?", stream=True)

Tools Reference

Creating Custom Tools

Using the @tool Decorator

from agno.tools.decorator import tool

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city.

    Args:
        city: City name to get weather for.
    """
    # Your implementation
    return f"Weather in {city}: 72F, sunny"

agent = Agent(tools=[get_weather])

Async Tools

@tool
async def fetch_data(url: str) -> str:
    """Fetch data from a URL."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()

Decorator Options

@tool(
    name="custom_name",                # Override function name
    description="Custom description",  # Override docstring
    show_result=True,                  # Show result to user
    stop_after_tool_call=True,         # Stop agent after this tool
    requires_confirmation=True,        # Ask user before executing
    cache_results=True,                # Cache results
    cache_ttl=3600,                    # Cache TTL in seconds
)
def my_tool(arg: str) -> str:
    """Docstring used as description if not overridden."""
    return "result"

Tool Hooks

@tool(
    pre_hook=lambda name, args: print(f"Calling {name}"),
    post_hook=lambda name, args, result: print(f"Result: {result}"),
)
def my_tool(arg: str) -> str:
    return "result"

Creating Toolkit Classes

For related tools, extend Toolkit:

from agno.tools.toolkit import Toolkit

class MyToolkit(Toolkit):
    def __init__(self, api_key: str):
        super().__init__(name="my_toolkit")
        self.api_key = api_key
        self.register(self.search)
        self.register(self.get_details)

    def search(self, query: str) -> str:
        """Search for items."""
        return f"Results for {query}"

    def get_details(self, item_id: str) -> str:
        """Get details for an item."""
        return f"Details for {item_id}"

agent = Agent(tools=[MyToolkit(api_key="...")])

Built-in Tools (120+)

Search & Web

Data & Databases

Content & Knowledge

APIs & Integrations

AI & Media

Finance

System & Files

MCP

Using Tools with Agents

from agno.agent import Agent
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools

# Multiple toolkits
agent = Agent(
    tools=[
        DuckDuckGoTools(),
        YFinanceTools(),
        get_weather,  # Custom @tool function
    ],
    tool_call_limit=20,
)

Workflow Reference

Imports

from agno.workflow import Workflow, Step, Steps, Parallel, Condition, Loop, Router

Creating a Workflow

workflow = Workflow(
    name="My Workflow",
    description="What this workflow does",
    steps=[step1, step2, step3],       # Sequential steps
    db=SqliteDb(db_file="agents.db"),  # Session persistence
    session_id="workflow-session",
    debug_mode=False,
)

Step Types

Step - Single Unit of Work

step = Step(
    name="Data Gathering",
    agent=my_agent,                    # Agent to execute
    description="Fetch market data",
    max_retries=3,                     # Retry on failure
    skip_on_failure=False,             # Skip instead of failing workflow
    add_workflow_history=True,         # Include prior step outputs
    num_history_runs=3,                # How many prior runs to include
)

A Step can use an agent, team, or custom executor:

# With agent
Step(agent=my_agent)

# With team
Step(team=my_team)

# With custom executor function
def my_executor(input: StepInput) -> StepOutput:
    # Custom logic
    return StepOutput(content="result")

Step(executor=my_executor)

Steps - Sequential Pipeline

pipeline = Steps(
    name="Processing Pipeline",
    steps=[step1, step2, step3],  # Execute in order
)

Parallel - Concurrent Execution

parallel = Parallel(
    step_a,
    step_b,
    step_c,
    name="Parallel Analysis",
)

# Or with list syntax
parallel = Parallel(
    "Parallel Analysis",   # Name as first string arg
    step_a, step_b, step_c,
)

Condition - Conditional Execution

# With callable
condition = Condition(
    evaluator=lambda input: "urgent" in input.input.lower(),
    steps=urgent_step,
    else_steps=normal_step,
    name="Priority Check",
)

# With CEL expression
condition = Condition(
    evaluator='input.contains("urgent")',
    steps=urgent_step,
    else_steps=normal_step,
)

CEL variables available: input, previous_step_content, previous_step_outputs, additional_data, session_state

Loop - Iterative Execution

loop = Loop(
    steps=[review_step, refine_step],
    max_iterations=3,
    end_condition=lambda outputs: "APPROVED" in outputs[-1].content,
    name="Refinement Loop",
)

# With CEL expression
loop = Loop(
    steps=[review_step],
    max_iterations=5,
    end_condition='last_step_content.contains("DONE")',
)

CEL variables: current_iteration, max_iterations, all_success, last_step_content, step_outputs

Router - Dynamic Step Selection

router = Router(
    selector=lambda input: simple_step if len(input.input) < 100 else complex_step,
    choices=[simple_step, complex_step],
    name="Complexity Router",
)

# With CEL expression
router = Router(
    selector='input.contains("simple") ? "simple_step" : "complex_step"',
    choices=[simple_step, complex_step],
)

Running Workflows

# Synchronous
response = workflow.run("Input message")
workflow.print_response("Input message", stream=True)

# Asynchronous
response = await workflow.arun("Input message")
await workflow.aprint_response("Input message", stream=True)

Example: Research Pipeline

from agno.agent import Agent
from agno.models.google import Gemini
from agno.tools.yfinance import YFinanceTools
from agno.workflow import Step, Workflow

data_agent = Agent(
    name="Data Gatherer",
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
    instructions=["Gather raw market data. Don't analyze, just organize."],
)

analyst = Agent(
    name="Analyst",
    model=Gemini(id="gemini-3-flash-preview"),
    instructions=["Analyze the data. Identify strengths, weaknesses, red flags."],
)

writer = Agent(
    name="Report Writer",
    model=Gemini(id="gemini-3-flash-preview"),
    instructions=["Write a concise investment brief. Lead with the bottom line."],
    markdown=True,
)

workflow = Workflow(
    name="Research Pipeline",
    steps=[
        Step(name="Gather", agent=data_agent),
        Step(name="Analyze", agent=analyst),
        Step(name="Report", agent=writer),
    ],
)

workflow.print_response("Analyze NVIDIA for investment", stream=True)

Example: Conditional Workflow

from agno.workflow import Workflow, Step, Condition

workflow = Workflow(
    steps=[
        Step(name="Classify", agent=classifier_agent),
        Condition(
            evaluator=lambda input: "technical" in input.previous_step_content.lower(),
            steps=Step(name="Technical", agent=tech_agent),
            else_steps=Step(name="General", agent=general_agent),
        ),
        Step(name="Finalize", agent=writer_agent),
    ],
)