or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

agent.mdagentos.mdeval.mdexceptions.mdguardrails.mdindex.mdknowledge.mdmedia.mdmemory.mdmodels.mdsessions.mdstorage.mdteam.mdtools.mdworkflow.md
tile.json

tessl/pypi-agno

A lightweight library for building Multi-Agent Systems

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/agno@2.3.x

To install, run

npx @tessl/cli install tessl/pypi-agno@2.3.0

index.mddocs/

Agno

Agno is a high-performance Python library for building Multi-Agent Systems with comprehensive features for production deployment. It provides a complete stack including a framework for building agents, AgentOS Runtime for deployment, and a Control Plane for monitoring.

Package Information

  • Package Name: agno
  • Language: Python
  • Installation: pip install agno

Core Imports

from agno.agent import Agent
from agno.team import Team
from agno.workflow import Workflow
from agno.models.openai import OpenAIChat
from agno.models.anthropic import Claude

Basic Usage

from agno.agent import Agent
from agno.models.openai import OpenAIChat

# Create a simple agent
agent = Agent(
    name="My Agent",
    model=OpenAIChat(id="gpt-4"),
    description="A helpful AI assistant",
    instructions=["Be concise", "Be helpful"]
)

# Run the agent
response = agent.run("Hello, how can you help me?")
print(response.content)

# Run with streaming
for chunk in agent.run("Tell me about Python", stream=True):
    if hasattr(chunk, 'content'):
        print(chunk.content, end="", flush=True)

Architecture

Agno's architecture consists of three main layers:

Core Abstractions:

  • Agent: Individual AI agents with model, tools, memory, and knowledge
  • Team: Multi-agent orchestration with shared context
  • Workflow: Step-based execution with conditional logic and parallelization

Integration Layer:

  • Models: 40+ LLM provider integrations
  • Tools: 100+ pre-built toolkits for web, data, AI services, and more
  • Storage: 15+ database backends for persistence
  • Vector DBs: 18+ vector database integrations for RAG

Production Features:

  • AgentOS: FastAPI runtime for production deployment
  • Memory: User memory and session management
  • Knowledge: Agentic RAG with multiple vector stores
  • Guardrails: Input/output validation and safety
  • Evaluation: Accuracy, performance, and reliability metrics

Capabilities

Agent Creation and Execution

Core agent functionality including creation, configuration, execution, and streaming. Agents support synchronous and asynchronous execution, conversation history, session management, and flexible input/output handling.

class Agent:
    def __init__(
        self,
        *,
        model: Optional[Union[Model, str]] = None,
        name: Optional[str] = None,
        description: Optional[str] = None,
        instructions: Optional[Union[str, List[str], Callable]] = None,
        tools: Optional[Sequence[Union[Toolkit, Callable, Function, Dict]]] = None,
        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,
        num_history_runs: Optional[int] = None,
        add_memories_to_context: Optional[bool] = None,
        add_knowledge_to_context: bool = False,
        output_schema: Optional[Type[BaseModel]] = None,
        stream: Optional[bool] = None,
        debug_mode: bool = False,
        # ... many more parameters
    ): ...

    def run(
        self,
        input: Union[str, List, Dict, Message, BaseModel, List[Message]],
        *,
        stream: Optional[bool] = None,
        session_id: Optional[str] = None,
        user_id: Optional[str] = None,
        images: Optional[Sequence[Image]] = None,
        audio: Optional[Sequence[Audio]] = None,
        videos: Optional[Sequence[Video]] = None,
        files: Optional[Sequence[File]] = None,
        output_schema: Optional[Type[BaseModel]] = None,
        **kwargs
    ) -> Union[RunOutput, Iterator[Union[RunOutputEvent, RunOutput]]]: ...

    async def arun(
        self,
        input: Union[str, List, Dict, Message, BaseModel, List[Message]],
        *,
        stream: Optional[bool] = None,
        session_id: Optional[str] = None,
        user_id: Optional[str] = None,
        **kwargs
    ) -> Union[RunOutput, AsyncIterator[RunOutputEvent]]: ...

Agent API

Multi-Agent Teams

Team orchestration for coordinating multiple agents with shared context, collaborative problem-solving, and role-based task distribution.

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,
        **kwargs
    ): ...

    def run(
        self,
        input: Union[str, List, Dict, Message, List[Message]],
        *,
        stream: Optional[bool] = None,
        **kwargs
    ) -> Union[TeamRunOutput, Iterator[Union[TeamRunOutputEvent, TeamRunOutput]]]: ...

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

Team API

Workflows

Workflow orchestration with steps, conditional logic, loops, and parallel execution for complex multi-step processes.

class Workflow:
    def __init__(
        self,
        *,
        name: Optional[str] = None,
        description: Optional[str] = None,
        steps: Optional[List[Step]] = None,
        db: Optional[Union[BaseDb, AsyncBaseDb]] = None,
        **kwargs
    ): ...

class Step:
    def __init__(
        self,
        agent: Agent,
        *,
        name: Optional[str] = None,
        condition: Optional[Condition] = None,
        **kwargs
    ): ...

class Loop:
    def __init__(
        self,
        steps: List[Step],
        *,
        max_iterations: int = 10,
        condition: Optional[Callable] = None,
        **kwargs
    ): ...

class Parallel:
    def __init__(
        self,
        steps: List[Step],
        **kwargs
    ): ...

Workflow API

Language Models

Support for 40+ LLM providers including OpenAI, Anthropic, Google, AWS Bedrock, Azure, and many others.

# OpenAI
class OpenAIChat:
    def __init__(
        self,
        id: str = "gpt-4",
        api_key: Optional[str] = None,
        **kwargs
    ): ...

# Anthropic
class Claude:
    def __init__(
        self,
        id: str = "claude-3-5-sonnet-20241022",
        api_key: Optional[str] = None,
        **kwargs
    ): ...

# Google
class Gemini:
    def __init__(
        self,
        id: str = "gemini-2.0-flash-exp",
        api_key: Optional[str] = None,
        **kwargs
    ): ...

Models API

Tools and Toolkits

100+ pre-built toolkits for web search, data analysis, AI services, collaboration tools, development, and more.

# Base classes
class Toolkit:
    def __init__(
        self,
        name: str = "toolkit",
        tools: List[Callable] = [],
        **kwargs
    ): ...

class Function:
    def __init__(
        self,
        name: str,
        description: Optional[str] = None,
        parameters: Dict[str, Any] = ...,
        entrypoint: Optional[Callable] = None,
        **kwargs
    ): ...

# Decorator for creating tools
def tool(func: Callable) -> Function: ...

# Example toolkits
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools
from agno.tools.github import GithubTools

Tools API

Knowledge and RAG

Agentic RAG with document processing, embedding, vector search, and knowledge retrieval across 18+ vector databases.

class Knowledge:
    def __init__(
        self,
        vector_db: VectorDb,
        *,
        embedder: Optional[Embedder] = None,
        reader: Optional[Reader] = None,
        num_documents: int = 5,
        **kwargs
    ): ...

# Vector databases
from agno.vectordb.pgvector import PgVector
from agno.vectordb.pineconedb import PineconeDb
from agno.vectordb.qdrant import Qdrant

# Embedders
from agno.knowledge.embedder import OpenAIEmbedder
from agno.knowledge.embedder import CohereEmbedder

# Readers
from agno.knowledge.reader import PDFReader, DocxReader, WebsiteReader

Knowledge API

Storage and Databases

Persistent storage for agent sessions, runs, and state across 15+ database backends.

# Base class
class BaseDb: ...

# SQL databases
from agno.db.postgres import PostgresDb
from agno.db.sqlite import SqliteDb

# NoSQL databases
from agno.db.mongo import MongoDb
from agno.db.redis import RedisDb

# Cloud databases
from agno.db.firestore import FirestoreDb
from agno.db.dynamo import DynamoDb

Storage API

Memory Management

User memory, session summaries, and memory optimization for contextual agent interactions.

class MemoryManager:
    def __init__(
        self,
        db: Union[BaseDb, AsyncBaseDb],
        **kwargs
    ): ...

class UserMemory:
    def __init__(
        self,
        memory: str,
        user_id: str,
        **kwargs
    ): ...

Memory API

AgentOS Runtime

Production runtime providing FastAPI server, SSE endpoints, and integrated control plane.

class AgentOS:
    def __init__(
        self,
        agents: Optional[List[Agent]] = None,
        teams: Optional[List[Team]] = None,
        workflows: Optional[List[Workflow]] = None,
        **kwargs
    ): ...

    def get_app(self) -> FastAPI: ...

    def serve(
        self,
        app: str,
        host: str = "0.0.0.0",
        port: int = 7777,
        reload: bool = False,
        **kwargs
    ) -> None: ...

AgentOS API

Guardrails and Safety

Input validation, output filtering, PII detection, and prompt injection protection.

class BaseGuardrail:
    def guard(
        self,
        input: Any,
        **kwargs
    ) -> Any: ...

class OpenAIModerationGuardrail(BaseGuardrail): ...
class PIIDetectionGuardrail(BaseGuardrail): ...
class PromptInjectionGuardrail(BaseGuardrail): ...

Guardrails API

Session Management

Session tracking, state management, and conversation history.

class AgentSession:
    session_id: str
    agent_id: Optional[str]
    user_id: Optional[str]
    session_data: Optional[Dict[str, Any]]
    runs: Optional[List[RunOutput]]
    created_at: Optional[int]
    updated_at: Optional[int]

Sessions API

Evaluation

Measure accuracy, performance, latency, and reliability of agents and workflows.

class AccuracyEval:
    def __init__(
        self,
        evaluations: List[AccuracyEvaluation],
        **kwargs
    ): ...

class PerformanceEval:
    def __init__(
        self,
        agent: Agent,
        num_runs: int = 10,
        **kwargs
    ): ...

class ReliabilityEval:
    def __init__(
        self,
        agent: Agent,
        num_runs: int = 10,
        **kwargs
    ): ...

Evaluation API

Media Types

Support for images, audio, video, and file inputs/outputs.

class Image:
    def __init__(
        self,
        url: Optional[str] = None,
        content: Optional[Union[str, bytes]] = None,
        **kwargs
    ): ...

class Audio:
    def __init__(
        self,
        url: Optional[str] = None,
        content: Optional[Union[str, bytes]] = None,
        **kwargs
    ): ...

class Video:
    def __init__(
        self,
        url: Optional[str] = None,
        **kwargs
    ): ...

class File:
    def __init__(
        self,
        url: Optional[str] = None,
        content: Optional[Union[str, bytes]] = None,
        **kwargs
    ): ...

Media API

Exceptions

Exception handling for agent runs, model errors, and validation failures.

class AgnoError(Exception): ...
class AgentRunException(Exception): ...
class RetryAgentRun(AgnoError): ...
class StopAgentRun(AgnoError): ...
class RunCancelledException(AgnoError): ...
class ModelAuthenticationError(AgnoError): ...
class ModelProviderError(AgnoError): ...
class ModelRateLimitError(AgnoError): ...
class InputCheckError(AgnoError): ...
class OutputCheckError(AgnoError): ...
class EvalError(AgnoError): ...

class CheckTrigger(str, Enum):
    input = "input"
    output = "output"

Exceptions API

Additional Features

Reasoning

Advanced reasoning capabilities for multi-step problem solving.

Hooks

Lifecycle hooks for pre/post processing of agent runs.

Compression

Context compression to manage token limits.

Culture

Shared long-term collective memory across agents.

Tracing

Integration with OpenTelemetry, Langfuse, and Arize for observability.

Filters

Query filters for vector databases and knowledge retrieval.

Types

from typing import Union, Optional, List, Dict, Any, Callable, Type
from pydantic import BaseModel

# Core message type
class Message(BaseModel):
    role: str
    content: Optional[Union[List[Any], str]]
    tool_calls: Optional[List[Dict[str, Any]]]
    ...

# Run output types
class RunOutput(BaseModel):
    run_id: str
    content: Optional[str]
    messages: List[Message]
    metrics: Metrics
    ...

class RunOutputEvent(BaseModel):
    event: RunEvent
    content: Optional[str]
    ...

# Run events enum
class RunEvent(str, Enum):
    run_started = "run_started"
    run_content = "run_content"
    run_completed = "run_completed"
    run_error = "run_error"
    tool_call_started = "tool_call_started"
    tool_call_completed = "tool_call_completed"
    ...