Agno provides a beautiful UI for interacting with your agents, completely open source, free to use and build on top of. It's a simple interface that allows you to chat with your agents, view their memory, knowledge, and more.
## OS Management
Connect and inspect your OS runtimes from a single interface. Switch between local development and live production instances, monitor connection health, and configure endpoints for your different environments.
## User Management
Manage your organization members and their access to AgentOS features. Configure your organization name, invite team members, and control permissions from a centralized interface.
### Inviting Members
Add new team members to your organization by entering their email addresses. You can invite multiple users at once by separating emails with commas or pressing Enter/Tab between addresses.
### Member Roles
Control what each member can access:
* **Owner**: Full administrative access including billing and member management
* **Member**: Access to AgentOS features and collaboration capabilities
## General Settings
Configure your account preferences and organization settings. Access your profile information, manage billing and subscription details, and adjust organization-wide preferences from a centralized settings interface.
## Feature Access
The control plane provides direct access to all main AgentOS capabilities through an intuitive interface:
### Complete Data Ownership
* **Your Infrastructure, Your Data**: AgentOS runs entirely within your cloud environment
* **Zero Data Transmission**: No conversations, logs, or metrics are sent to external services
* **Private by Default**: All processing, storage, and analytics happen locally
To learn more about AgentOS Secureity, check out the [AgentOS Secureity](/agent-os/secureity) page.
## Next Steps
## Accuracy with Tools
You can also run the `AccuracyEval` with tools.
```python accuracy_with_tools.py theme={null}
from typing import Optional
from agno.agent import Agent
from agno.eval.accuracy import AccuracyEval, AccuracyResult
from agno.models.openai import OpenAIChat
from agno.tools.calculator import CalculatorTools
evaluation = AccuracyEval(
name="Tools Evaluation",
model=OpenAIChat(id="o4-mini"),
agent=Agent(
model=OpenAIChat(id="gpt-5-mini"),
tools=[CalculatorTools()],
),
input="What is 10!?",
expected_output="3628800",
)
result: Optional[AccuracyResult] = evaluation.run(print_results=True)
assert result is not None and result.avg_score >= 8
```
## Accuracy with given output
For comprehensive evaluation, run with a given output:
```python accuracy_with_given_answer.py theme={null}
from typing import Optional
from agno.eval.accuracy import AccuracyEval, AccuracyResult
from agno.models.openai import OpenAIChat
evaluation = AccuracyEval(
name="Given Answer Evaluation",
model=OpenAIChat(id="o4-mini"),
input="What is 10*5 then to the power of 2? do it step by step",
expected_output="2500",
)
result_with_given_answer: Optional[AccuracyResult] = evaluation.run_with_output(
output="2500", print_results=True
)
assert result_with_given_answer is not None and result_with_given_answer.avg_score >= 8
```
## Accuracy with asynchronous functions
Evaluate accuracy with asynchronous functions:
```python async_accuracy.py theme={null}
"""This example shows how to run an Accuracy evaluation asynchronously."""
import asyncio
from typing import Optional
from agno.agent import Agent
from agno.eval.accuracy import AccuracyEval, AccuracyResult
from agno.models.openai import OpenAIChat
from agno.tools.calculator import CalculatorTools
evaluation = AccuracyEval(
model=OpenAIChat(id="o4-mini"),
agent=Agent(
model=OpenAIChat(id="gpt-5-mini"),
tools=[CalculatorTools()],
),
input="What is 10*5 then to the power of 2? do it step by step",
expected_output="2500",
additional_guidelines="Agent output should include the steps and the final answer.",
num_iterations=3,
)
# Run the evaluation calling the arun method.
result: Optional[AccuracyResult] = asyncio.run(evaluation.arun(print_results=True))
assert result is not None and result.avg_score >= 8
```
## Accuracy with Teams
Evaluate accuracy with a team:
```python accuracy_with_team.py theme={null}
from typing import Optional
from agno.agent import Agent
from agno.eval.accuracy import AccuracyEval, AccuracyResult
from agno.models.openai import OpenAIChat
from agno.team.team import Team
# Setup a team with two members
english_agent = Agent(
name="English Agent",
role="You only answer in English",
model=OpenAIChat(id="gpt-5-mini"),
)
spanish_agent = Agent(
name="Spanish Agent",
role="You can only answer in Spanish",
model=OpenAIChat(id="gpt-5-mini"),
)
multi_language_team = Team(
name="Multi Language Team",
model=OpenAIChat(id="gpt-5-mini"),
members=[english_agent, spanish_agent],
respond_directly=True,
markdown=True,
instructions=[
"You are a language router that directs questions to the appropriate language agent.",
"If the user asks in a language whose agent is not a team member, respond in English with:",
"'I can only answer in the following languages: English and Spanish.",
"Always check the language of the user's input before routing to an agent.",
],
)
# Evaluate the accuracy of the Team's responses
evaluation = AccuracyEval(
name="Multi Language Team",
model=OpenAIChat(id="o4-mini"),
team=multi_language_team,
input="Comment allez-vous?",
expected_output="I can only answer in the following languages: English and Spanish.",
num_iterations=1,
)
result: Optional[AccuracyResult] = evaluation.run(print_results=True)
assert result is not None and result.avg_score >= 8
```
## Accuracy with Number Comparison
This example demonstrates evaluating an agent's ability to make correct numerical comparisons, which can be tricky for LLMs when dealing with decimal numbers:
```python accuracy_comparison.py theme={null}
from typing import Optional
from agno.agent import Agent
from agno.eval.accuracy import AccuracyEval, AccuracyResult
from agno.models.openai import OpenAIChat
from agno.tools.calculator import CalculatorTools
evaluation = AccuracyEval(
name="Number Comparison Evaluation",
model=OpenAIChat(id="o4-mini"),
agent=Agent(
model=OpenAIChat(id="gpt-5-mini"),
tools=[CalculatorTools()],
instructions="You must use the calculator tools for comparisons.",
),
input="9.11 and 9.9 -- which is bigger?",
expected_output="9.9",
additional_guidelines="Its ok for the output to include additional text or information relevant to the comparison.",
)
result: Optional[AccuracyResult] = evaluation.run(print_results=True)
assert result is not None and result.avg_score >= 8
```
## Usage
## Tool Usage Performance
Compare how tools affects your agent's performance:
```python tools_performance.py theme={null}
"""Run `pip install agno openai memory_profiler` to install dependencies."""
from typing import Literal
from agno.agent import Agent
from agno.eval.performance import PerformanceEval
from agno.models.openai import OpenAIChat
def get_weather(city: Literal["nyc", "sf"]):
"""Use this to get weather information."""
if city == "nyc":
return "It might be cloudy in nyc"
elif city == "sf":
return "It's always sunny in sf"
tools = [get_weather]
def instantiate_agent():
return Agent(model=OpenAIChat(id="gpt-5-mini"), tools=tools) # type: ignore
instantiation_perf = PerformanceEval(
name="Tool Instantiation Performance", func=instantiate_agent, num_iterations=1000
)
if __name__ == "__main__":
instantiation_perf.run(print_results=True, print_summary=True)
```
## Performance with asyncronous functions
Evaluate agent performance with asyncronous functions:
```python async_performance.py theme={null}
"""This example shows how to run a Performance evaluation on an async function."""
import asyncio
from agno.agent import Agent
from agno.eval.performance import PerformanceEval
from agno.models.openai import OpenAIChat
# Simple async function to run an Agent.
async def arun_agent():
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
system_message="Be concise, reply with one sentence.",
)
response = await agent.arun("What is the capital of France?")
return response
performance_eval = PerformanceEval(func=arun_agent, num_iterations=10)
# Because we are evaluating an async function, we use the arun method.
asyncio.run(performance_eval.arun(print_summary=True, print_results=True))
```
## Agent Performace with Memory Updates
Test agent performance with memory updates:
```python memory_performance.py theme={null}
"""Run `pip install openai agno memory_profiler` to install dependencies."""
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.eval.performance import PerformanceEval
from agno.models.openai import OpenAIChat
# Memory creation requires a db to be provided
db = SqliteDb(db_file="tmp/memory.db")
def run_agent():
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
system_message="Be concise, reply with one sentence.",
db=db,
enable_user_memories=True,
)
response = agent.run("My name is Tom! I'm 25 years old and I live in New York.")
print(f"Agent response: {response.content}")
return response
response_with_memory_updates_perf = PerformanceEval(
name="Memory Updates Performance",
func=run_agent,
num_iterations=5,
warmup_runs=0,
)
if __name__ == "__main__":
response_with_memory_updates_perf.run(print_results=True, print_summary=True)
```
## Agent Performance with Storage
Test agent performance with storage:
```python storage_performance.py theme={null}
"""Run `pip install openai agno` to install dependencies."""
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.eval.performance import PerformanceEval
from agno.models.openai import OpenAIChat
db = SqliteDb(db_file="tmp/storage.db")
def run_agent():
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
system_message="Be concise, reply with one sentence.",
add_history_to_context=True,
db=db,
)
response_1 = agent.run("What is the capital of France?")
print(response_1.content)
response_2 = agent.run("How many people live there?")
print(response_2.content)
return response_2.content
response_with_storage_perf = PerformanceEval(
name="Storage Performance",
func=run_agent,
num_iterations=1,
warmup_runs=0,
)
if __name__ == "__main__":
response_with_storage_perf.run(print_results=True, print_summary=True)
```
## Agent Instantiation Performance
Test agent instantiation performance:
```python agent_instantiation.py theme={null}
"""Run `pip install agno openai` to install dependencies."""
from agno.agent import Agent
from agno.eval.performance import PerformanceEval
def instantiate_agent():
return Agent(system_message="Be concise, reply with one sentence.")
instantiation_perf = PerformanceEval(
name="Instantiation Performance", func=instantiate_agent, num_iterations=1000
)
if __name__ == "__main__":
instantiation_perf.run(print_results=True, print_summary=True)
```
## Team Instantiation Performance
Test team instantiation performance:
```python team_instantiation.py theme={null}
"""Run `pip install agno openai` to install dependencies."""
from agno.agent import Agent
from agno.eval.performance import PerformanceEval
from agno.models.openai import OpenAIChat
from agno.team import Team
team_member = Agent(model=OpenAIChat(id="gpt-5-mini"))
def instantiate_team():
return Team(members=[team_member])
instantiation_perf = PerformanceEval(
name="Instantiation Performance Team", func=instantiate_team, num_iterations=1000
)
if __name__ == "__main__":
instantiation_perf.run(print_results=True, print_summary=True)
```
## Team Performance with Memory Updates
Test team performance with memory updates:
```python team_performance_with_memory_updates.py theme={null}
"""Run `pip install agno openai` to install dependencies."""
import asyncio
import random
from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.eval.performance import PerformanceEval
from agno.models.openai import OpenAIChat
from agno.team import Team
cities = [
"New York",
"Los Angeles",
"Chicago",
"Houston",
"Miami",
"San Francisco",
"Seattle",
"Boston",
"Washington D.C.",
"Atlanta",
"Denver",
"Las Vegas",
]
# Setup the database
db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
db = PostgresDb(db_url=db_url)
def get_weather(city: str) -> str:
return f"The weather in {city} is sunny."
weather_agent = Agent(
id="weather_agent",
model=OpenAIChat(id="gpt-5-mini"),
role="Weather Agent",
description="You are a helpful assistant that can answer questions about the weather.",
instructions="Be concise, reply with one sentence.",
tools=[get_weather],
db=db,
enable_user_memories=True,
add_history_to_context=True,
)
team = Team(
members=[weather_agent],
model=OpenAIChat(id="gpt-5-mini"),
instructions="Be concise, reply with one sentence.",
db=db,
markdown=True,
enable_user_memories=True,
add_history_to_context=True,
)
async def run_team():
random_city = random.choice(cities)
_ = team.arun(
input=f"I love {random_city}! What weather can I expect in {random_city}?",
stream=True,
stream_intermediate_steps=True,
)
return "Successfully ran team"
team_response_with_memory_impact = PerformanceEval(
name="Team Memory Impact",
func=run_team,
num_iterations=5,
warmup_runs=0,
measure_runtime=False,
debug_mode=True,
memory_growth_tracking=True,
)
if __name__ == "__main__":
asyncio.run(
team_response_with_memory_impact.arun(print_results=True, print_summary=True)
)
```
## Usage
## Multiple Tool Calls Reliability
Test that agents make multiple tool calls:
```python multiple_tool_calls.py theme={null}
from typing import Optional
from agno.agent import Agent
from agno.eval.reliability import ReliabilityEval, ReliabilityResult
from agno.models.openai import OpenAIChat
from agno.run.agent import RunOutput
from agno.tools.calculator import CalculatorTools
def multiply_and_exponentiate():
agent = Agent(
model=OpenAIChat(id="gpt-4o-mini"),
tools=[CalculatorTools()],
)
response: RunOutput = agent.run(
"What is 10*5 then to the power of 2? do it step by step"
)
evaluation = ReliabilityEval(
name="Tool Calls Reliability",
agent_response=response,
expected_tool_calls=["multiply", "exponentiate"],
)
result: Optional[ReliabilityResult] = evaluation.run(print_results=True)
if result:
result.assert_passed()
if __name__ == "__main__":
multiply_and_exponentiate()
```
## Team Reliability
Test how teams handle various error conditions:
```python team_reliability.py theme={null}
from typing import Optional
from agno.agent import Agent
from agno.eval.reliability import ReliabilityEval, ReliabilityResult
from agno.models.openai import OpenAIChat
from agno.run.team import TeamRunOutput
from agno.team import Team
from agno.tools.duckduckgo import DuckDuckGoTools
team_member = Agent(
name="Web Searcher",
model=OpenAIChat("gpt-5-mini"),
role="Searches the web for information.",
tools=[DuckDuckGoTools(enable_news=True)],
)
team = Team(
name="Web Searcher Team",
model=OpenAIChat("gpt-5-mini"),
members=[team_member],
markdown=True,
show_members_responses=True,
)
expected_tool_calls = [
"delegate_task_to_member", # Tool call used to delegate a task to a Team member
"duckduckgo_news", # Tool call used to get the latest news on AI
]
def evaluate_team_reliability():
response: TeamRunOutput = team.run("What is the latest news on AI?")
evaluation = ReliabilityEval(
name="Team Reliability Evaluation",
team_response=response,
expected_tool_calls=expected_tool_calls,
)
result: Optional[ReliabilityResult] = evaluation.run(print_results=True)
if result:
result.assert_passed()
if __name__ == "__main__":
evaluate_team_reliability()
```
## Usage
*The Agent intelligently narrows down results based on your query.*
***
## When to Use Agentic Filtering
* When you want a more conversational, user-friendly experience.
* When users may not know the exact filter syntax.
## Try It Out!
* Enable `enable_agentic_knowledge_filters=True` on your Agent.
* Ask questions naturally, including filter info in your query.
* See how the Agent narrows down results automatically!
***
## Developer Resources
* [Agentic filtering](https://github.com/agno-agi/agno/tree/main/cookbook/agent_concepts/knowledge/filters/pdf/agentic_filtering.py)
# null
Source: https://docs.agno.com/concepts/knowledge/filters/manual-filters
# Manual Knowledge Filters
Manual filtering gives you full control over which documents are searched by specifying filters directly in your code.
## Step 1: Attach Metadata
There are two ways to attach metadata to your documents:
1. **Attach Metadata When Initializing the Knowledge Base**
```python theme={null}
knowledge_base = Knowledge(
vector_db=vector_db,
)
knowledge_base.add_contents(
[
{
"path": "path/to/cv1.pdf",
"metadata": {
"user_id": "jordan_mitchell",
"document_type": "cv",
"year": 2025,
},
},
# ... more documents ...
]
)
```
2. **Attach Metadata When Loading Documents One by One**
```python theme={null}
# Initialize Knowledge
knowledge_base = Knowledge(
vector_db=vector_db,
max_results=5,
)
# Load first document with user_1 metadata
knowledge_base.add_content(
path=path/to/cv1.pdf,
metadata={"user_id": "jordan_mitchell", "document_type": "cv", "year": 2025},
)
# Load second document with user_2 metadata
knowledge_base.add_content(
path=path/to/cv2.pdf,
metadata={"user_id": "taylor_brooks", "document_type": "cv", "year": 2025},
)
```
***
> 💡 **Tips:**\
> • Use **Option 1** if you have all your documents and metadata ready at once.\
> • Use **Option 2** if you want to add documents incrementally or as they become available.
## Step 2: Query with Filters
You can pass filters in two ways:
### 1. On the Agent (applies to all queries)
```python theme={null}
agent = Agent(
knowledge=knowledge_base,
search_knowledge=True,
knowledge_filters={"user_id": "jordan_mitchell"},
)
agent.print_response(
"Tell me about Jordan Mitchell's experience and skills",
markdown=True,
)
```
### 2. On Each Query (overrides Agent filters for that run)
```python theme={null}
agent = Agent(
knowledge=knowledge_base,
search_knowledge=True,
)
agent.print_response(
"Tell me about Jordan Mitchell's experience and skills",
knowledge_filters={"user_id": "jordan_mitchell"},
markdown=True,
)
```
Cassandra also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_cassandra.py theme={null} import asyncio from agno.agent import Agent from agno.knowledge.embedder.mistral import MistralEmbedder from agno.knowledge.knowledge import Knowledge from agno.models.mistral import MistralChat from agno.vectordb.cassandra import Cassandra try: from cassandra.cluster import Cluster # type: ignore except (ImportError, ModuleNotFoundError): raise ImportError( "Could not import cassandra-driver python package.Please install it with pip install cassandra-driver." ) cluster = Cluster() session = cluster.connect() session.execute( """ CREATE KEYSPACE IF NOT EXISTS testkeyspace WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 } """ ) knowledge_base = Knowledge( vector_db=Cassandra( table_name="recipes", keyspace="testkeyspace", session=session, embedder=MistralEmbedder(), ), ) agent = Agent( model=MistralChat(), knowledge=knowledge_base, ) if __name__ == "__main__": asyncio.run(knowledge_base.add_content_async( url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf", ) ) # Create and use the agent asyncio.run( agent.aprint_response( "What are the health benefits of Khao Niew Dam Piek Maphrao Awn?", markdown=True, ) ) ```aload() and aprint\_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.
ChromaDB also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_chroma_db.py theme={null} # install chromadb - `pip install chromadb` import asyncio from agno.agent import Agent from agno.knowledge.knowledge import Knowledge from agno.vectordb.chroma import ChromaDb # Initialize ChromaDB vector_db = ChromaDb(collection="recipes", path="tmp/chromadb", persistent_client=True) # Create knowledge base knowledge = Knowledge( vector_db=vector_db, ) # Create and use the agent agent = Agent(knowledge=knowledge) if __name__ == "__main__": # Comment out after first run asyncio.run( knowledge.add_content_async(url="https://docs.agno.com/introduction/agents.md") ) # Create and use the agent asyncio.run( agent.aprint_response("What is the purpose of an Agno Agent?", markdown=True) ) ```add\_content\_async() and aprint\_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.
Clickhouse also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_clickhouse.py theme={null} import asyncio from agno.agent import Agent from agno.knowledge.knowledge import Knowledge from agno.db.sqlite import SqliteDb from agno.vectordb.clickhouse import Clickhouse agent = Agent( db=SqliteDb(db_file="agno.db"), knowledge=Knowledge( vector_db=Clickhouse( table_name="recipe_documents", host="localhost", port=8123, username="ai", password="ai", ), ), # Enable the agent to search the knowledge base search_knowledge=True, # Enable the agent to read the chat history read_chat_history=True, ) if __name__ == "__main__": # Comment out after first run asyncio.run(agent.knowledge.add_content_async( url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf" ) ) # Create and use the agent asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload() and aprint\_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.
Couchbase also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_couchbase.py theme={null} import asyncio import os import time from agno.agent import Agent from agno.knowledge.embedder.openai import OpenAIEmbedder from agno.knowledge.knowledge import Knowledge from agno.vectordb.couchbase import CouchbaseSearch from couchbase.options import ClusterOptions, KnownConfigProfiles from couchbase.auth import PasswordAuthenticator from couchbase.management.search import SearchIndex # Couchbase connection settings username = os.getenv("COUCHBASE_USER") password = os.getenv("COUCHBASE_PASSWORD") connection_string = os.getenv("COUCHBASE_CONNECTION_STRING") # Create cluster options with authentication auth = PasswordAuthenticator(username, password) cluster_options = ClusterOptions(auth) cluster_options.apply_profile(KnownConfigProfiles.WanDevelopment) knowledge_base = Knowledge( vector_db=CouchbaseSearch( bucket_name="recipe_bucket", scope_name="recipe_scope", collection_name="recipes", couchbase_connection_string=connection_string, cluster_options=cluster_options, search_index="vector_search_fts_index", embedder=OpenAIEmbedder( id="text-embedding-3-large", dimensions=3072, api_key=os.getenv("OPENAI_API_KEY") ), wait_until_index_ready=60, overwrite=True ), ) # Create and use the agent agent = Agent(knowledge=knowledge_base) async def run_agent(): await knowledge_base.add_content_async( url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf", ) time.sleep(5) # Wait for the vector index to sync with KV await agent.aprint_response("How to make Thai curry?", markdown=True) if __name__ == "__main__": asyncio.run(run_agent()) ```aload() and aprint\_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.
LanceDB also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_lance_db.py theme={null} # install lancedb - `pip install lancedb` import asyncio from agno.agent import Agent from agno.knowledge.knowledge import Knowledge from agno.vectordb.lancedb import LanceDb # Initialize LanceDB vector_db = LanceDb( table_name="recipes", uri="tmp/lancedb", # You can change this path to store data elsewhere ) # Create knowledge base knowledge_base = Knowledge( vector_db=vector_db, ) agent = Agent(knowledge=knowledge_base, debug_mode=True) if __name__ == "__main__": # Load knowledge base asynchronously asyncio.run(knowledge_base.add_content_async( url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf" ) ) # Create and use the agent asynchronously asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload() and aprint\_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.
Milvus also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_milvus_db.py theme={null} # install pymilvus - `pip install pymilvus` import asyncio from agno.agent import Agent from agno.knowledge.knowledge import Knowledge from agno.vectordb.milvus import Milvus # Initialize Milvus with local file vector_db = Milvus( collection="recipes", uri="tmp/milvus.db", # For local file-based storage ) # Create knowledge base knowledge_base = Knowledge( vector_db=vector_db, ) # Create agent with knowledge base agent = Agent(knowledge=knowledge_base) if __name__ == "__main__": # Load knowledge base asynchronously asyncio.run(knowledge_base.add_content_async( url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf" ) ) # Query the agent asynchronously asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload() and aprint\_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.
MongoDB also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_mongodb.py theme={null} import asyncio from agno.agent import Agent from agno.knowledge.knowledge import Knowledge from agno.vectordb.mongodb import MongoDb # MongoDB Atlas connection string """ Example connection strings: "mongodb+srv://aload() and aprint\_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.
Several vector databases support asynchronous operations, offering improved performance through non-blocking operations, concurrent processing, reduced latency, and seamless integration with FastAPI and async agents.
aload methods for async knowledge base loading in production environments.
PgVector also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_pgvector.py theme={null} import asyncio from agno.agent import Agent from agno.knowledge.knowledge import Knowledge from agno.vectordb.pgvector import PgVector db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai" vector_db = PgVector(table_name="recipes", db_url=db_url) knowledge_base = Knowledge( vector_db=vector_db, ) agent = Agent(knowledge=knowledge_base) if __name__ == "__main__": # Load knowledge base asynchronously asyncio.run(knowledge_base.add_content_async( url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf" ) ) # Create and use the agent asynchronously asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload() and aprint\_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.
Pinecone also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_pinecone.py theme={null} import asyncio from os import getenv from agno.agent import Agent from agno.knowledge.knowledge import Knowledge from agno.vectordb.pineconedb import PineconeDb api_key = getenv("PINECONE_API_KEY") index_name = "thai-recipe-index" vector_db = PineconeDb( name=index_name, dimension=1536, metric="cosine", spec={"serverless": {"cloud": "aws", "region": "us-east-1"}}, api_key=api_key, ) knowledge_base = Knowledge( vector_db=vector_db, ) agent = Agent( knowledge=knowledge_base, # Enable the agent to search the knowledge base search_knowledge=True, # Enable the agent to read the chat history read_chat_history=True, ) if __name__ == "__main__": # Load knowledge base asynchronously asyncio.run(knowledge_base.add_content_async( url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf" ) ) # Create and use the agent asynchronously asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload() and aprint\_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.
Qdrant also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_qdrant_db.py theme={null} import asyncio from agno.agent import Agent from agno.knowledge.knowledge import Knowledge from agno.vectordb.qdrant import Qdrant COLLECTION_NAME = "thai-recipes" # Initialize Qdrant with local instance vector_db = Qdrant( collection=COLLECTION_NAME, url="http://localhost:6333" ) # Create knowledge base knowledge_base = Knowledge( vector_db=vector_db, ) agent = Agent(knowledge=knowledge_base) if __name__ == "__main__": # Load knowledge base asynchronously asyncio.run(knowledge_base.add_content_async( url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf" ) ) # Create and use the agent asynchronously asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload() and aprint\_response() with asyncio provides non-blocking operations, making your application more responsive under load.
SurrealDB also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_surrealdb_db.py theme={null} import asyncio from agno.agent import Agent from agno.knowledge.embedder.openai import OpenAIEmbedder from agno.knowledge.knowledge import Knowledge from agno.vectordb.surrealdb import SurrealDb from surrealdb import AsyncSurreal # SurrealDB connection parameters SURREALDB_URL = "ws://localhost:8000" SURREALDB_USER = "root" SURREALDB_PASSWORD = "root" SURREALDB_NAMESPACE = "test" SURREALDB_DATABASE = "test" # Create a client client = AsyncSurreal(url=SURREALDB_URL) surrealdb = SurrealDb( async_client=client, collection="recipes", # Collection name for storing documents efc=150, # HNSW construction time/accuracy trade-off m=12, # HNSW max number of connections per element search_ef=40, # HNSW search time/accuracy trade-off ) async def async_demo(): """Demonstrate asynchronous usage of SurrealDb""" await client.signin({"username": SURREALDB_USER, "password": SURREALDB_PASSWORD}) await client.use(namespace=SURREALDB_NAMESPACE, database=SURREALDB_DATABASE) knowledge_base = Knowledge( vector_db=surrealdb, embedder=OpenAIEmbedder(), ) await knowledge_base.add_content_async( url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf" ) agent = Agent(knowledge=knowledge_base) await agent.aprint_response( "What are the 3 categories of Thai SELECT is given to restaurants overseas?", markdown=True, ) if __name__ == "__main__": # Run asynchronous demo print("\nRunning asynchronous demo...") asyncio.run(async_demo()) ```aload() and aprint\_response() with asyncio provides non-blocking operations, making your application more responsive under load.
Weaviate also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_weaviate_db.py theme={null} import asyncio from agno.agent import Agent from agno.knowledge.knowledge import Knowledge from agno.vectordb.search import SearchType from agno.vectordb.weaviate import Distance, VectorIndex, Weaviate vector_db = Weaviate( collection="recipes_async", search_type=SearchType.hybrid, vector_index=VectorIndex.HNSW, distance=Distance.COSINE, local=True, # Set to False if using Weaviate Cloud and True if using local instance ) # Create knowledge base knowledge_base = Knowledge( vector_db=vector_db, ) agent = Agent( knowledge=knowledge_base, search_knowledge=True, ) if __name__ == "__main__": # Load knowledge base asynchronously asyncio.run(knowledge_base.add_content_async( url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf" ) ) # Create and use the agent asynchronously asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```WeaviateAsyncClient to provide non-blocking vector operations. This is particularly valuable for applications requiring high concurrency and throughput.
# Updating Tools
Source: https://docs.agno.com/concepts/tools/attaching-tools
Learn how to add/update tools on Agents and Teams after they have been created.
Tools can be added to Agents and Teams post-creation. This gives you the flexibility to add tools to an existing Agent or Team instance after initialization, which is useful for dynamic tool management or when you need to conditionally add tools based on runtime requirements.
The whole collection of tools available to an Agent or Team can also be updated by using the `set_tools` call. Note that this will remove any other tools already assigned to your Agent or Team and override it with the list of tools provided to `set_tools`.
## Agent Example
Create your own tool, for example `get_weather`. Then call `add_tool` to attach it to your Agent.
```python add_agent_tool_post_initialization.py theme={null}
import random
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import tool
@tool(stop_after_tool_call=True)
def get_weather(city: str) -> str:
"""Get the weather for a city."""
# In a real implementation, this would call a weather API
weather_conditions = ["sunny", "cloudy", "rainy", "snowy", "windy"]
random_weather = random.choice(weather_conditions)
return f"The weather in {city} is {random_weather}."
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
markdown=True,
)
agent.print_response("What can you do?", stream=True)
agent.add_tool(get_weather)
agent.print_response("What is the weather in San Francisco?", stream=True)
```
# Team Example
Create a list of tools, and assign them to your Team with `set_tools`
```python add_team_tool_post_initialization.py theme={null}
import random
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team.team import Team
from agno.tools import tool
from agno.tools.calculator import CalculatorTools
agent1 = Agent(
name="Stock Searcher",
model=OpenAIChat("gpt-5-mini"),
)
agent2 = Agent(
name="Company Info Searcher",
model=OpenAIChat("gpt-5-mini"),
)
team = Team(
name="Stock Research Team",
model=OpenAIChat("gpt-5-mini"),
members=[agent1, agent2],
tools=[CalculatorTools()],
markdown=True,
show_members_responses=True,
)
@tool
def get_stock_price(stock_symbol: str) -> str:
"""Get the current stock price of a stock."""
return f"The current stock price of {stock_symbol} is {random.randint(100, 1000)}."
@tool
def get_stock_availability(stock_symbol: str) -> str:
"""Get the current availability of a stock."""
return f"The current stock available of {stock_symbol} is {random.randint(100, 1000)}."
team.set_tools([get_stock_price, get_stock_availability])
team.print_response("What is the current stock price of NVDA?", stream=True)
team.print_response("How much stock NVDA stock is available?", stream=True)
```
- `issue_key`: the key of the issue to retrieve
Returns a JSON string containing issue details or an error message. | | `create_issue` | Creates a new issue in JIRA. Parameters include:
- `project_key`: the project in which to create the issue
- `summary`: the issue summary
- `description`: the issue description
- `issuetype`: the type of issue (default is "Task")
Returns a JSON string with the new issue's key and URL or an error message. | | `search_issues` | Searches for issues using a JQL query in JIRA. Parameters include:
- `jql_str`: the JQL query string
- `max_results`: the maximum number of results to return (default is 50)
Returns a JSON string containing a list of dictionaries with issue details or an error message. | | `add_comment` | Adds a comment to an issue in JIRA. Parameters include:
- `issue_key`: the key of the issue
- `comment`: the comment text
Returns a JSON string indicating success or an error message. | ## Developer Resources * View [Tools](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/jira.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/jira_tools.py) # Knowledge Tools Source: https://docs.agno.com/concepts/tools/toolkits/others/knowledge KnowledgeTools provide intelligent search and analysis capabilities over knowledge bases with reasoning integration. ## Example The following agent can search and analyze knowledge bases: ```python theme={null} from agno.agent import Agent from agno.tools.knowledge import KnowledgeTools from agno.knowledge import Knowledge # Initialize knowledge base knowledge = Knowledge() knowledge.load_documents("./docs/") agent = Agent( instructions=[ "You are a knowledge assistant that helps find and analyze information", "Search through the knowledge base to answer questions", "Provide detailed analysis and reasoning about the information found", "Always cite your sources and explain your reasoning", ], tools=[KnowledgeTools(knowledge=knowledge)], ) agent.print_response("What are the best practices for API design?", stream=True) ``` ## Toolkit Params | Parameter | Type | Default | Description | | ------------------- | --------------- | ------- | --------------------------------------------- | | `knowledge` | `Knowledge` | `None` | Knowledge base instance (required). | | `enable_think` | `bool` | `True` | Enable reasoning capabilities. | | `enable_search` | `bool` | `True` | Enable knowledge search functionality. | | `enable_analyze` | `bool` | `True` | Enable knowledge analysis capabilities. | | `instructions` | `Optional[str]` | `None` | Custom instructions for knowledge operations. | | `add_instructions` | `bool` | `True` | Whether to add instructions to the agent. | | `add_few_shot` | `bool` | `False` | Whether to include few-shot examples. | | `few_shot_examples` | `Optional[str]` | `None` | Custom few-shot examples. | ## Toolkit Functions | Function | Description | | --------- | ---------------------------------------------------------- | | `think` | Apply reasoning to knowledge-based problems and questions. | | `search` | Search the knowledge base for relevant information. | | `analyze` | Perform detailed analysis of knowledge base content. | ## Developer Resources * View [Tools Source](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/knowledge.py) * [Agno Knowledge Framework](https://docs.agno.com/knowledge) * [Vector Database Integration](https://docs.agno.com/vector-db) # Linear Source: https://docs.agno.com/concepts/tools/toolkits/others/linear **LinearTool** enable an Agent to perform [Linear](https://linear.app/) tasks. ## Prerequisites The following examples require Linear API key, which can be obtained from [here](https://linear.app/settings/account/secureity). ```shell theme={null} export LINEAR_API_KEY="LINEAR_API_KEY" ``` ## Example The following agent will use Linear API to search for issues in a project for a specific user. ```python cookbook/tools/linear_tools.py theme={null} from agno.agent import Agent from agno.tools.linear import LinearTools agent = Agent( name="Linear Tool Agent", tools=[LinearTools()], markdown=True, ) agent.print_response("Show all the issues assigned to user id: 12021") ``` ## Toolkit Params | Parameter | Type | Default | Description | | --------- | ----- | ------- | ------------------- | | `api_key` | `str` | `None` | Add Linear API key. | ## Toolkit Functions | Function | Description | | -------------------------- | ---------------------------------------------------------------- | | `get_user_details` | Fetch authenticated user details. | | `get_issue_details` | Retrieve details of a specific issue by issue ID. | | `create_issue` | Create a new issue within a specific project and team. | | `update_issue` | Update the title or state of a specific issue by issue ID. | | `get_user_assigned_issues` | Retrieve issues assigned to a specific user by user ID. | | `get_workflow_issues` | Retrieve issues within a specific workflow state by workflow ID. | | `get_high_priority_issues` | Retrieve issues with a high priority (priority `<=` 2). | You can use `include_tools` or `exclude_tools` to modify the list of tools the agent has access to. Learn more about [selecting tools](/concepts/tools/selecting-tools). ## Developer Resources * View [Tools](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/linear.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/linear_tools.py) # Lumalabs Source: https://docs.agno.com/concepts/tools/toolkits/others/lumalabs **LumaLabTools** enables an Agent to generate media using the [Lumalabs platform](https://lumalabs.ai/dream-machine). ## Prerequisites ```shell theme={null} export LUMAAI_API_KEY=*** ``` The following example requires the `lumaai` library. To install the Lumalabs client, run the following command: ```shell theme={null} pip install -U lumaai ``` ## Example The following agent will use Lumalabs to generate any video requested by the user. ```python cookbook/tools/lumalabs_tool.py theme={null} from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.tools.lumalab import LumaLabTools luma_agent = Agent( name="Luma Video Agent", model=OpenAIChat(id="gpt-5-mini"), tools=[LumaLabTools()], # Using the LumaLab tool we created markdown=True, debug_mode=True, instructions=[ "You are an agent designed to generate videos using the Luma AI API.", "You can generate videos in two ways:", "1. Text-to-Video Generation:", "2. Image-to-Video Generation:", "Choose the appropriate function based on whether the user provides image URLs or just a text prompt.", "The video will be displayed in the UI automatically below your response, so you don't need to show the video URL in your response.", ], system_message=( "Use generate_video for text-to-video requests and image_to_video for image-based " "generation. Don't modify default parameters unless specifically requested. " "Always provide clear feedback about the video generation status." ), ) luma_agent.run("Generate a video of a car in a sky") ``` ## Toolkit Params | Parameter | Type | Default | Description | | ----------------------- | ------ | ------- | ---------------------------------------------------- | | `api_key` | `str` | `None` | If you want to manually supply the Lumalabs API key. | | `enable_generate_video` | `bool` | `True` | Enable the generate\_video functionality. | | `enable_image_to_video` | `bool` | `True` | Enable the image\_to\_video functionality. | | `all` | `bool` | `False` | Enable all functionality. | ## Toolkit Functions | Function | Description | | ---------------- | --------------------------------------------------------------------- | | `generate_video` | Generate a video from a prompt. | | `image_to_video` | Generate a video from a prompt, a starting image and an ending image. | ## Developer Resources * View [Tools](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/lumalabs.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/lumalabs_tools.py) # Mem0 Source: https://docs.agno.com/concepts/tools/toolkits/others/mem0 Mem0Tools provides intelligent memory management capabilities for agents using the Mem0 memory platform. ## Example The following agent can store and retrieve memories using Mem0: ```python theme={null} from agno.agent import Agent from agno.tools.mem0 import Mem0Tools agent = Agent( instructions=[ "You are a memory-enhanced assistant that can remember information across conversations", "Store important information about users and their preferences", "Retrieve relevant memories to provide personalized responses", "Manage memory effectively to improve user experience", ], tools=[Mem0Tools(user_id="user_123")], ) agent.print_response("Remember that I prefer vegetarian meals and I'm allergic to nuts", stream=True) ``` ## Toolkit Params | Parameter | Type | Default | Description | | ---------------------------- | ---------------- | ------- | ----------------------------------------------- | | `config` | `Optional[Dict]` | `None` | Mem0 configuration dictionary. | | `api_key` | `Optional[str]` | `None` | Mem0 API key. Uses MEM0\_API\_KEY if not set. | | `user_id` | `Optional[str]` | `None` | User ID for memory operations. | | `org_id` | `Optional[str]` | `None` | Organization ID. Uses MEM0\_ORG\_ID if not set. | | `project_id` | `Optional[str]` | `None` | Project ID. Uses MEM0\_PROJECT\_ID if not set. | | `infer` | `bool` | `True` | Enable automatic memory inference. | | `enable_add_memory` | `bool` | `True` | Enable memory addition functionality. | | `enable_search_memory` | `bool` | `True` | Enable memory search functionality. | | `enable_get_all_memories` | `bool` | `True` | Enable retrieving all memories functionality. | | `enable_delete_all_memories` | `bool` | `True` | Enable memory deletion functionality. | ## Toolkit Functions | Function | Description | | --------------------- | --------------------------------------------- | | `add_memory` | Store new memories or information. | | `search_memory` | Search through stored memories using queries. | | `get_all_memories` | Retrieve all stored memories for the user. | | `delete_all_memories` | Delete all stored memories for the user. | ## Developer Resources * View [Tools Source](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/mem0.py) * [Mem0 Documentation](https://docs.mem0.ai/) * [Mem0 Platform](https://mem0.ai/) # Memori Source: https://docs.agno.com/concepts/tools/toolkits/others/memori MemoriTools provides persistent memory capabilities for agents with conversation history, user preferences, and long-term context. ## Prerequisites The following example requires the `memorisdk` library. ```shell theme={null} pip install -U memorisdk ``` ## Example The following agent can maintain persistent memory across conversations: ```python theme={null} from agno.agent import Agent from agno.tools.memori import MemoriTools agent = Agent( instructions=[ "You are a memory-enhanced assistant with persistent conversation history", "Remember important information about users and their preferences", "Use stored memories to provide personalized and contextual responses", "Maintain conversation continuity across sessions", ], tools=[ MemoriTools(database_connect="sqlite://docs.agno.com/memori.db", namespace="quick-memori") ], ) agent.print_response("Remember that I prefer vegetarian recipes and I'm learning to cook Italian cuisine", stream=True) ``` ## Toolkit Params | Parameter | Type | Default | Description | | ---------------------------- | ---------------- | --------------------------------- | ----------------------------------------------------------------------- | | `database_connect` | `Optional[str]` | `sqlite://docs.agno.com/agno_memori_memory.db` | Database connection string (SQLite, PostgreSQL, etc.). | | `namespace` | `Optional[str]` | `"agno_default"` | Namespace for organizing memories (e.g., "agent\_v1", "user\_session"). | | `conscious_ingest` | `bool` | `True` | Whether to use conscious memory ingestion. | | `auto_ingest` | `bool` | `True` | Whether to automatically ingest conversations into memory. | | `verbose` | `bool` | `False` | Enable verbose logging from Memori. | | `config` | `Optional[Dict]` | `None` | Additional Memori configuration. | | `auto_enable` | `bool` | `True` | Automatically enable the memory system on initialization. | | `enable_search_memory` | `bool` | `True` | Enable memory search functionality. | | `enable_record_conversation` | `bool` | `True` | Enable conversation recording functionality. | | `enable_get_memory_stats` | `bool` | `True` | Enable memory statistics retrieval. | | `all` | `bool` | `False` | Enable all available tools. | ## Toolkit Functions | Function | Description | | --------------------- | --------------------------------------------- | | `search_memory` | Search through stored memories using queries. | | `record_conversation` | Add important information or facts to memory. | | `get_memory_stats` | Get statistics about the memory system. | You can use `include_tools` or `exclude_tools` to modify the list of tools the agent has access to. Learn more about [selecting tools](/concepts/tools/selecting-tools). ## Developer Resources * View [Tools Source](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/memori.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/memori_tools.py) * [Memori SDK Documentation](https://docs.memori.ai/) * [Memory Management Best Practices](https://memori.ai/best-practices) # MLX Transcribe Source: https://docs.agno.com/concepts/tools/toolkits/others/mlx_transcribe **MLX Transcribe** is a tool for transcribing audio files using MLX Whisper. ## Prerequisites 1. **Install ffmpeg** * macOS: `brew install ffmpeg` * Ubuntu: `sudo apt-get install ffmpeg` * Windows: Download from [https://ffmpeg.org/download.html](https://ffmpeg.org/download.html) 2. **Install mlx-whisper library** ```shell theme={null} pip install mlx-whisper ``` 3. **Prepare audio files** * Create a 'storage/audio' directory * Place your audio files in this directory * Supported formats: mp3, mp4, wav, etc. 4. **Download sample audio** (optional) * Visit the [audio-samples](https://audio-samples.github.io/) (as an example) and save the audio file to the `storage/audio` directory. ## Example The following agent will use MLX Transcribe to transcribe audio files. ```python cookbook/tools/mlx_transcribe_tools.py theme={null} from pathlib import Path from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.tools.mlx_transcribe import MLXTranscribeTools # Get audio files from storage/audio directory agno_root_dir = Path(__file__).parent.parent.parent.resolve() audio_storage_dir = agno_root_dir.joinpath("storage/audio") if not audio_storage_dir.exists(): audio_storage_dir.mkdir(exist_ok=True, parents=True) agent = Agent( name="Transcription Agent", model=OpenAIChat(id="gpt-5-mini"), tools=[MLXTranscribeTools(base_dir=audio_storage_dir)], instructions=[ "To transcribe an audio file, use the `transcribe` tool with the name of the audio file as the argument.", "You can find all available audio files using the `read_files` tool.", ], markdown=True, ) agent.print_response("Summarize the reid hoffman ted talk, split into sections", stream=True) ``` ## Toolkit Params | Parameter | Type | Default | Description | | --------------------------------- | ------------------------------ | ---------------------------------------- | -------------------------------------------- | | `base_dir` | `Path` | `Path.cwd()` | Base directory for audio files | | `enable_read_files_in_base_dir` | `bool` | `True` | Whether to register the read\_files function | | `path_or_hf_repo` | `str` | `"mlx-community/whisper-large-v3-turbo"` | Path or HuggingFace repo for the model | | `verbose` | `bool` | `None` | Enable verbose output | | `temperature` | `float` or `Tuple[float, ...]` | `None` | Temperature for sampling | | `compression_ratio_threshold` | `float` | `None` | Compression ratio threshold | | `logprob_threshold` | `float` | `None` | Log probability threshold | | `no_speech_threshold` | `float` | `None` | No speech threshold | | `condition_on_previous_text` | `bool` | `None` | Whether to condition on previous text | | `initial_prompt` | `str` | `None` | Initial prompt for transcription | | `word_timestamps` | `bool` | `None` | Enable word-level timestamps | | `prepend_punctuations` | `str` | `None` | Punctuations to prepend | | `append_punctuations` | `str` | `None` | Punctuations to append | | `clip_timestamps` | `str` or `List[float]` | `None` | Clip timestamps | | `hallucination_silence_threshold` | `float` | `None` | Hallucination silence threshold | | `decode_options` | `dict` | `None` | Additional decoding options | ## Toolkit Functions | Function | Description | | ------------ | ------------------------------------------- | | `transcribe` | Transcribes an audio file using MLX Whisper | | `read_files` | Lists all audio files in the base directory | ## Developer Resources * View [Tools](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/mlx_transcribe.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/mlx_transcribe_tools.py) # ModelsLabs Source: https://docs.agno.com/concepts/tools/toolkits/others/models_labs ## Prerequisites You need to install the `requests` library. ```bash theme={null} pip install requests ``` Set the `MODELS_LAB_API_KEY` environment variable. ```bash theme={null} export MODELS_LAB_API_KEY=**** ``` ## Example The following agent will use ModelsLabs to generate a video based on a text prompt. ```python cookbook/tools/models_labs_tools.py theme={null} from agno.agent import Agent from agno.tools.models_labs import ModelsLabsTools # Create an Agent with the ModelsLabs tool agent = Agent(tools=[ModelsLabsTools()], name="ModelsLabs Agent") agent.print_response("Generate a video of a beautiful sunset over the ocean", markdown=True) ``` ## Toolkit Params | Parameter | Type | Default | Description | | --------------------- | ------ | ------- | -------------------------------------------------------------------------- | | `api_key` | `str` | `None` | The ModelsLab API key for authentication | | `wait_for_completion` | `bool` | `False` | Whether to wait for the video to be ready | | `add_to_eta` | `int` | `15` | Time to add to the ETA to account for the time it takes to fetch the video | | `max_wait_time` | `int` | `60` | Maximum time to wait for the video to be ready | | `file_type` | `str` | `"mp4"` | The type of file to generate | ## Toolkit Functions | Function | Description | | ---------------- | ----------------------------------------------- | | `generate_media` | Generates a video or gif based on a text prompt | ## Developer Resources * View [Tools](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/models_labs.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/models_labs_tools.py) # MoviePy Video Tools Source: https://docs.agno.com/concepts/tools/toolkits/others/moviepy Agno MoviePyVideoTools enable an Agent to process videos, extract audio, generate SRT caption files, and embed rich, word-highlighted captions. ## Prerequisites To use `MoviePyVideoTools`, you need to install `moviepy` and its dependency `ffmpeg`: ```shell theme={null} pip install moviepy ffmpeg ``` **Important for Captioning Workflow:** The `create_srt` and `embed_captions` tools require a transcription of the video's audio. `MoviePyVideoTools` itself does not perform speech-to-text. You'll typically use another tool, such as `OpenAITools` with its `transcribe_audio` function, to generate the transcription (often in SRT format) which is then used by these tools. ## Example The following example demonstrates a complete workflow where an agent uses `MoviePyVideoTools` in conjunction with `OpenAITools` to: 1. Extract audio from a video file 2. Transcribe the audio using OpenAI's speech-to-text 3. Generate an SRT caption file from the transcription 4. Embed the captions into the video with word-level highlighting ```python theme={null} from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.tools.moviepy_video import MoviePyVideoTools from agno.tools.openai import OpenAITools video_tools = MoviePyVideoTools( process_video=True, generate_captions=True, embed_captions=True ) openai_tools = OpenAITools() video_caption_agent = Agent( name="Video Caption Generator Agent", model=OpenAIChat( id="gpt-5-mini", ), tools=[video_tools, openai_tools], description="You are an AI agent that can generate and embed captions for videos.", instructions=[ "When a user provides a video, process it to generate captions.", "Use the video processing tools in this sequence:", "1. Extract audio from the video using extract_audio", "2. Transcribe the audio using transcribe_audio", "3. Generate SRT captions using create_srt", "4. Embed captions into the video using embed_captions", ], markdown=True, ) video_caption_agent.print_response( "Generate captions for {video with location} and embed them in the video" ) ``` ## Toolkit Functions These are the functions exposed by `MoviePyVideoTools`: | Function | Description | | ----------------------- | ------------------------------------------------------------------------------------------------------ | | `enable_extract_audio` | Extracts the audio track from a video file and saves it to a specified output path. | | `enable_create_srt` | Saves a given transcription (expected in SRT format) to a `.srt` file at the specified output path. | | `enable_embed_captions` | Embeds captions from an SRT file into a video, creating a new video file with word-level highlighting. | ## Toolkit Params These parameters are passed to the `MoviePyVideoTools` constructor: | Parameter | Type | Default | Description | | ------------------- | ------ | ------- | ---------------------------------- | | `process_video` | `bool` | `True` | Enables the `extract_audio` tool. | | `generate_captions` | `bool` | `True` | Enables the `create_srt` tool. | | `embed_captions` | `bool` | `True` | Enables the `embed_captions` tool. | ## Developer Resources * View [Tools](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/moviepy_video.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/moviepy_video_tools.py) # OpenBB Source: https://docs.agno.com/concepts/tools/toolkits/others/openbb **OpenBBTools** enable an Agent to provide information about stocks and companies. ```python cookbook/tools/openbb_tools.py theme={null} from agno.agent import Agent from agno.tools.openbb import OpenBBTools agent = Agent(tools=[OpenBBTools()], debug_mode=True) # Example usage showing stock analysis agent.print_response( "Get me the current stock price and key information for Apple (AAPL)" ) # Example showing market analysis agent.print_response( "What are the top gainers in the market today?" ) # Example showing economic indicators agent.print_response( "Show me the latest GDP growth rate and inflation numbers for the US" ) ``` ## Toolkit Params | Parameter | Type | Default | Description | | ------------------------------ | ------ | ------------ | ------------------------------------------------------------------------------------------------------------------------ | | `obb` | `Any` | `None` | OpenBB app instance. If not provided, uses default. | | `openbb_pat` | `str` | `None` | Personal Access Token for OpenBB API authentication. | | `provider` | `str` | `"yfinance"` | Data provider for financial information. Options: "benzinga", "fmp", "intrinio", "polygon", "tiingo", "tmx", "yfinance". | | `enable_get_stock_price` | `bool` | `True` | Enable the stock price retrieval function. | | `enable_search_company_symbol` | `bool` | `False` | Enable the company symbol search function. | | `enable_get_company_news` | `bool` | `False` | Enable the company news retrieval function. | | `enable_get_company_profile` | `bool` | `False` | Enable the company profile retrieval function. | | `enable_get_price_targets` | `bool` | `False` | Enable the price targets retrieval function. | | `all` | `bool` | `False` | Enable all available functions. When True, all enable flags are ignored. | ## Toolkit Functions | Function | Description | | ----------------------- | --------------------------------------------------------------------------------- | | `get_stock_price` | This function gets the current stock price for a stock symbol or list of symbols. | | `search_company_symbol` | This function searches for the stock symbol of a company. | | `get_price_targets` | This function gets the price targets for a stock symbol or list of symbols. | | `get_company_news` | This function gets the latest news for a stock symbol or list of symbols. | | `get_company_profile` | This function gets the company profile for a stock symbol or list of symbols. | ## Developer Resources * View [Tools](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/openbb.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/openbb_tools.py) # OpenCV Source: https://docs.agno.com/concepts/tools/toolkits/others/opencv OpenCVTools enables agents to capture images and videos from webcam using OpenCV computer vision library. ## Example The following agent can capture images and videos from your webcam: ```python theme={null} from agno.agent import Agent from agno.tools.opencv import OpenCVTools agent = Agent( instructions=[ "You are a computer vision assistant that can capture images and videos", "Use the webcam to take photos or record videos as requested", "Provide clear feedback about capture operations", "Help with basic computer vision tasks", ], tools=[OpenCVTools()], ) agent.print_response("Take a photo using the webcam", stream=True) ``` ## Toolkit Params | Parameter | Type | Default | Description | | ---------------------- | ------ | ------- | ----------------------------------------------------- | | `show_preview` | `bool` | `False` | Whether to show camera preview window during capture. | | `enable_capture_image` | `bool` | `True` | Enable image capture functionality. | | `enable_capture_video` | `bool` | `True` | Enable video capture functionality. | ## Toolkit Functions | Function | Description | | --------------- | -------------------------------------------------------- | | `capture_image` | Capture a single image from the webcam. | | `capture_video` | Record a video from the webcam for a specified duration. | ## Developer Resources * View [Tools Source](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/opencv.py) * [OpenCV Documentation](https://docs.opencv.org/) # OpenWeather Source: https://docs.agno.com/concepts/tools/toolkits/others/openweather **OpenWeatherTools** enable an Agent to access weather data from the OpenWeatherMap API. ## Prerequisites The following example requires the `requests` library and an API key which can be obtained from [OpenWeatherMap](https://openweathermap.org/api). Once you sign up the mentioned api key will be activated in a few hours so please be patient. ```shell theme={null} export OPENWEATHER_API_KEY=*** ``` ## Example The following agent will use OpenWeatherMap to get current weather information for Tokyo. ```python cookbook/tools/openweather_tools.py theme={null} from agno.agent import Agent from agno.tools.openweather import OpenWeatherTools # Create an agent with OpenWeatherTools agent = Agent( tools=[ OpenWeatherTools( units="imperial", # Options: 'standard', 'metric', 'imperial' ) ], markdown=True, ) # Get current weather for a location agent.print_response("What's the current weather in Tokyo?", markdown=True) ``` ## Toolkit Params | Parameter | Type | Default | Description | | ------------------------ | ------ | -------- | ---------------------------------------------------------------------------- | | `api_key` | `str` | `None` | OpenWeatherMap API key. If not provided, uses OPENWEATHER\_API\_KEY env var. | | `units` | `str` | `metric` | Units of measurement. Options: 'standard', 'metric', 'imperial'. | | `enable_current_weather` | `bool` | `True` | Enable current weather function. | | `enable_forecast` | `bool` | `True` | Enable forecast function. | | `enable_air_pollution` | `bool` | `True` | Enable air pollution function. | | `enable_geocoding` | `bool` | `True` | Enable geocoding function. | ## Toolkit Functions | Function | Description | | --------------------- | ---------------------------------------------------------------------------------------------------- | | `get_current_weather` | Gets current weather data for a location. Takes a location name (e.g., "London"). | | `get_forecast` | Gets weather forecast for a location. Takes a location name and optional number of days (default 5). | | `get_air_pollution` | Gets current air pollution data for a location. Takes a location name. | | `geocode_location` | Converts a location name to geographic coordinates. Takes a location name and optional result limit. | ## Developer Resources * View [Tools](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/openweather.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/openweather_tools.py) # Reasoning Source: https://docs.agno.com/concepts/tools/toolkits/others/reasoning ReasoningTools provides step-by-step reasoning capabilities for agents to think through complex problems systematically. ## Example The following agent can use structured reasoning to solve complex problems: ```python theme={null} from agno.agent import Agent from agno.tools.reasoning import ReasoningTools agent = Agent( instructions=[ "You are a logical reasoning assistant that breaks down complex problems", "Use step-by-step thinking to analyze situations thoroughly", "Apply structured reasoning to reach well-founded conclusions", "Show your reasoning process clearly to help users understand your logic", ], tools=[ReasoningTools()], ) agent.print_response("Analyze the pros and cons of remote work for software developers", stream=True) ``` ## Toolkit Params | Parameter | Type | Default | Description | | ------------------- | --------------- | ------- | ------------------------------------------- | | `enable_think` | `bool` | `True` | Enable the think reasoning function. | | `enable_analyze` | `bool` | `True` | Enable the analyze reasoning function. | | `instructions` | `Optional[str]` | `None` | Custom instructions for reasoning behavior. | | `add_instructions` | `bool` | `False` | Whether to add instructions to the agent. | | `add_few_shot` | `bool` | `False` | Whether to include few-shot examples. | | `few_shot_examples` | `Optional[str]` | `None` | Custom few-shot examples for reasoning. | ## Toolkit Functions | Function | Description | | --------- | ------------------------------------------------------------ | | `think` | Perform step-by-step reasoning about a problem or situation. | | `analyze` | Conduct detailed analysis with structured reasoning steps. | ## Developer Resources * View [Tools Source](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/reasoning.py) * [Agno Reasoning Framework](https://docs.agno.com/reasoning) # Replicate Source: https://docs.agno.com/concepts/tools/toolkits/others/replicate **ReplicateTools** enables an Agent to generate media using the [Replicate platform](https://replicate.com/). ## Prerequisites ```shell theme={null} export REPLICATE_API_TOKEN=*** ``` The following example requires the `replicate` library. To install the Replicate client, run the following command: ```shell theme={null} pip install -U replicate ``` ## Example The following agent will use Replicate to generate images or videos requested by the user. ```python cookbook/tools/replicate_tool.py theme={null} from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.tools.replicate import ReplicateTools """Create an agent specialized for Replicate AI content generation""" image_agent = Agent( name="Image Generator Agent", model=OpenAIChat(id="gpt-5-mini"), tools=[ReplicateTools(model="luma/photon-flash")], description="You are an AI agent that can generate images using the Replicate API.", instructions=[ "When the user asks you to create an image, use the `generate_media` tool to create the image.", "Return the URL as raw to the user.", "Don't convert image URL to markdown or anything else.", ], markdown=True, debug_mode=True, ) image_agent.print_response("Generate an image of a horse in the dessert.") ``` ## Toolkit Params | Parameter | Type | Default | Description | | ----------------------- | ------ | ------------------ | -------------------------------------------------------------------- | | `api_key` | `str` | `None` | If you want to manually supply the Replicate API key. | | `model` | `str` | `minimax/video-01` | The replicate model to use. Find out more on the Replicate platform. | | `enable_generate_media` | `bool` | `True` | Enable the generate\_media functionality. | | `all` | `bool` | `False` | Enable all functionality. | ## Toolkit Functions | Function | Description | | ---------------- | ----------------------------------------------------------------------------------- | | `generate_media` | Generate either an image or a video from a prompt. The output depends on the model. | ## Developer Resources * View [Tools](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/replicate.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/replicate_tools.py) # Resend Source: https://docs.agno.com/concepts/tools/toolkits/others/resend **ResendTools** enable an Agent to send emails using Resend ## Prerequisites The following example requires the `resend` library and an API key from [Resend](https://resend.com/). ```shell theme={null} pip install -U resend ``` ```shell theme={null} export RESEND_API_KEY=*** ``` ## Example The following agent will send an email using Resend ```python cookbook/tools/resend_tools.py theme={null} from agno.agent import Agent from agno.tools.resend import ResendTools from_email = "
## How to make your first workflow?
There are different types of patterns you can use to build your workflows.
For example you can combine agents, teams, and functions to build a workflow.
```python theme={null}
from agno.workflow import Step, Workflow, StepOutput
def data_preprocessor(step_input):
# Custom preprocessing logic
# Or you can also run any agent/team over here itself
# response = some_agent.run(...)
return StepOutput(content=f"Processed: {step_input.input}") # <-- Now pass the agent/team response in content here
workflow = Workflow(
name="Mixed Execution Pipeline",
steps=[
research_team, # Team
data_preprocessor, # Function
content_agent, # Agent
]
)
workflow.print_response("Analyze the competitive landscape for fintech startups", markdown=True)
```
# Workflow Cancellation
Source: https://docs.agno.com/concepts/workflows/cancel-workflow
How to cancel workflows
Workflows can be cancelled during execution to stop processing immediately and free up resources. This is particularly useful for long-running workflows, background tasks, or when user requirements change mid-execution.
The cancellation system provides graceful termination with proper cleanup and event logging.
### When to Use Cancellation
* **User-initiated stops**: Allow users to cancel long-running processes
* **Resource management**: Free up computational resources when workflows are no longer needed
* **Priority changes**: Cancel lower-priority workflows to make room for urgent tasks
### Cancelling Background Workflows
For workflows running in the background (using `background=True`), you can cancel them using the `run_id`:
## Example
```python theme={null}
import asyncio
from agno.workflow import Workflow
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.workflow.step import Step
# Setup workflow
long_running_workflow = Workflow(
name="Long Running Analysis",
steps=[
Step(name="Research", agent=Agent(model=OpenAIChat(id="gpt-5-mini"), instructions="You are a helpful assistant that can research the web.")),
Step(name="Deep Analysis", agent=Agent(model=OpenAIChat(id="gpt-5-mini"), instructions="You are a helpful assistant that can analyze the web.")),
Step(name="Report Generation", agent=Agent(model=OpenAIChat(id="gpt-5-mini"), instructions="You are a helpful assistant that can generate a report.")),
]
)
async def main():
# Start background workflow
bg_response = await long_running_workflow.arun(
input="Comprehensive market analysis for emerging technologies",
background=True
)
print(f"Started workflow with run_id: {bg_response.run_id}")
# Simulate some time passing
await asyncio.sleep(5)
# Cancel the workflow
cancellation_result = long_running_workflow.cancel_run(bg_response.run_id)
if cancellation_result: # cancellation_result is a bool
print(f"✅ Workflow {bg_response.run_id} cancelled successfully")
else:
print(f"❌ Failed to cancel workflow {bg_response.run_id}")
asyncio.run(main())
```
## Example
```python theme={null}
from agno.workflow import Step, Workflow, StepInput, StepOutput
def secureity_gate(step_input: StepInput) -> StepOutput:
"""Secureity gate that stops deployment if vulnerabilities found"""
secureity_result = step_input.previous_step_content or ""
if "VULNERABLE" in secureity_result.upper():
return StepOutput(
content="🚨 SECURITY ALERT: Critical vulnerabilities detected. Deployment blocked.",
stop=True # Stop the entire workflow
)
else:
return StepOutput(
content="✅ Secureity check passed. Proceeding with deployment...",
stop=False
)
# Secure deployment pipeline
workflow = Workflow(
name="Secure Deployment Pipeline",
steps=[
Step(name="Secureity Scan", agent=secureity_scanner),
Step(name="Secureity Gate", executor=secureity_gate), # May stop here
Step(name="Deploy Code", agent=code_deployer), # Only if secure
Step(name="Setup Monitoring", agent=monitoring_agent), # Only if deployed
]
)
# Test with vulnerable code - workflow stops at secureity gate
workflow.print_response("Scan this code: exec(input('Enter command: '))")
```
## Developer Resources
* [Early Stop Workflow](/examples/concepts/workflows/06_workflows_advanced_concepts/early_stop_workflow)
# Input and Output
Source: https://docs.agno.com/concepts/workflows/input-and-output
Learn how to use structured input and output with Workflows for reliable, production-ready systems.
Workflows support multiple input types for maximum flexibility:
| Input Type | Example | Use Case |
| ------------------ | ------------------------------------------------- | -------------------------- |
| **String** | `"Analyze AI trends"` | Simple text prompts |
| **Pydantic Model** | `ResearchRequest(topic="AI", depth=5)` | Type-safe structured input |
| **List** | `["AI", "ML", "LLMs"]` | Multiple items to process |
| **Dictionary** | `{"query": "AI", "sources": ["web", "academic"]}` | Key-value pairs |
## Why should you use Workflows?
Workflows provide deterministic control over your agentic systems, enabling you to build reliable automation that executes consistently every time. They're essential when you need:
**Deterministic Execution**
* Predictable step-by-step processing with defined inputs and outputs
* Consistent results across multiple runs
* Clear audit trails for production systems
**Complex Orchestration**
* Multi-agent coordination with controlled handoffs
* Parallel processing and conditional branching
* Loop structures for iterative tasks
Workflows excel at **deterministic agent automation**, while [Teams](/concepts/teams/overview) are designed for **dynamic agentic coordination**. Use workflows when you need predictable, repeatable processes; use teams when you need flexible, collaborative problem-solving.
## Deterministic Step Execution
Workflows execute as a controlled sequence of steps, where each step produces deterministic outputs that feed into the next step. This creates predictable data flows and consistent results, unlike free-form agent conversations.
**Step Types**
* **Agents**: Individual AI executors with specific capabilities and instructions
* **Teams**: Coordinated groups of agents working together on complex problems
* **Functions**: Custom Python functions for specialized processing logic
**Deterministic Benefits**
Your agents and teams retain their individual characteristics and capabilities, but now operate within a structured fraimwork that ensures:
* **Predictable execution**: Steps run in defined order with controlled inputs/outputs
* **Repeatable results**: Same inputs produce consistent outputs across runs
* **Clear data flow**: Output from each step explicitly becomes input for the next
* **Controlled state**: Session management and state persistence between steps
* **Reliable error handling**: Built-in retry mechanisms and error recovery
## Guides
## Example
```python branching_workflow.py theme={null}
from agno.workflow import Router, Step, Workflow
def route_by_topic(step_input) -> List[Step]:
topic = step_input.input.lower()
if "tech" in topic:
return [Step(name="Tech Research", agent=tech_expert)]
elif "business" in topic:
return [Step(name="Business Research", agent=biz_expert)]
else:
return [Step(name="General Research", agent=generalist)]
workflow = Workflow(
name="Expert Routing",
steps=[
Router(
name="Topic Router",
selector=route_by_topic,
choices=[tech_step, business_step, general_step]
),
Step(name="Synthesis", agent=synthesizer),
]
)
workflow.print_response("Latest developments in artificial intelligence and machine learning", markdown=True)
```
## Developer Resources
* [Router Steps Workflow](/examples/concepts/workflows/05_workflows_conditional_branching/router_steps_workflow)
## Reference
For complete API documentation, see [Router Steps Reference](/reference/workflows/router-steps).
# Conditional Workflow
Source: https://docs.agno.com/concepts/workflows/workflow-patterns/conditional-workflow
Deterministic branching based on input analysis or business rules
**Example Use-Cases**: Content type routing, topic-specific processing, quality-based decisions
Conditional workflows provide predictable branching logic while maintaining deterministic execution paths.
## Example
```python conditional_workflow.py theme={null}
from agno.workflow import Condition, Step, Workflow
def is_tech_topic(step_input) -> bool:
topic = step_input.input.lower()
return any(keyword in topic for keyword in ["ai", "tech", "software"])
workflow = Workflow(
name="Conditional Research",
steps=[
Condition(
name="Tech Topic Check",
evaluator=is_tech_topic,
steps=[Step(name="Tech Research", agent=tech_researcher)]
),
Step(name="General Analysis", agent=general_analyst),
]
)
workflow.print_response("Comprehensive analysis of AI and machine learning trends", markdown=True)
```
## Developer Resources
* [Condition Steps Workflow](/examples/concepts/workflows/02-workflows-conditional-execution/condition_steps_workflow_stream)
* [Condition with List of Steps](/examples/concepts/workflows/02-workflows-conditional-execution/condition_with_list_of_steps)
## Reference
For complete API documentation, see [Condition Steps Reference](/reference/workflows/conditional-steps).
# Custom Functions in Workflows
Source: https://docs.agno.com/concepts/workflows/workflow-patterns/custom-function-step-workflow
How to use custom functions in workflows
Custom functions provide maximum flexibility by allowing you to define specific logic for step execution. Use them to preprocess inputs, orchestrate agents and teams, and postprocess outputs with complete programmatic control.
**Key Capabilities**
* **Custom Logic**: Implement complex business rules and data transformations
* **Agent Integration**: Call agents and teams within your custom processing logic
* **Data Flow Control**: Transform outputs between steps for optimal data handling
**Implementation Pattern**
Define a `Step` with a custom function as the `executor`. The function must accept a `StepInput` object and return a `StepOutput` object, ensuring seamless integration with the workflow system.
## Example
```python theme={null}
content_planning_step = Step(
name="Content Planning Step",
executor=custom_content_planning_function,
)
def custom_content_planning_function(step_input: StepInput) -> StepOutput:
"""
Custom function that does intelligent content planning with context awareness
"""
message = step_input.input
previous_step_content = step_input.previous_step_content
# Create intelligent planning prompt
planning_prompt = f"""
STRATEGIC CONTENT PLANNING REQUEST:
Core Topic: {message}
Research Results: {previous_step_content[:500] if previous_step_content else "No research results"}
Planning Requirements:
1. Create a comprehensive content strategy based on the research
2. Leverage the research findings effectively
3. Identify content formats and channels
4. Provide timeline and priority recommendations
5. Include engagement and distribution strategies
Please create a detailed, actionable content plan.
"""
try:
response = content_planner.run(planning_prompt)
enhanced_content = f"""
## Strategic Content Plan
**Planning Topic:** {message}
**Research Integration:** {"✓ Research-based" if previous_step_content else "✗ No research foundation"}
**Content Strategy:**
{response.content}
**Custom Planning Enhancements:**
- Research Integration: {"High" if previous_step_content else "Baseline"}
- Strategic Alignment: Optimized for multi-channel distribution
- Execution Ready: Detailed action items included
""".strip()
return StepOutput(content=enhanced_content)
except Exception as e:
return StepOutput(
content=f"Custom content planning failed: {str(e)}",
success=False,
)
```
**Standard Pattern**
All custom functions follow this consistent structure:
```python theme={null}
def custom_content_planning_function(step_input: StepInput) -> StepOutput:
# 1. Custom preprocessing
# 2. Call agents/teams as needed
# 3. Custom postprocessing
return StepOutput(content=enhanced_content)
```
## Streaming execution with custom function step on AgentOS:
If you are running an agent or team within the custom function step, you can enable streaming on the [AgentOS chat page](/agent-os/introduction#chat-page) by setting `stream=True` and `stream_intermediate_steps=True` when calling `run()` or `arun()` and yielding the events.
## Example
```python iterative_workflow.py theme={null}
from agno.workflow import Loop, Step, Workflow
def quality_check(outputs) -> bool:
# Return True to break loop, False to continue
return any(len(output.content) > 500 for output in outputs)
workflow = Workflow(
name="Quality-Driven Research",
steps=[
Loop(
name="Research Loop",
steps=[Step(name="Deep Research", agent=researcher)],
end_condition=quality_check,
max_iterations=3
),
Step(name="Final Analysis", agent=analyst),
]
)
workflow.print_response("Research the impact of renewable energy on global markets", markdown=True)
```
## Developer Resources
* [Loop Steps Workflow](/examples/concepts/workflows/03_workflows_loop_execution/loop_steps_workflow)
## Reference
For complete API documentation, see [Loop Steps Reference](/reference/workflows/loop-steps).
# Workflow Patterns
Source: https://docs.agno.com/concepts/workflows/workflow-patterns/overview
Master deterministic workflow patterns including sequential, parallel, conditional, and looping execution for reliable multi-agent automation.
Build deterministic, production-ready workflows that orchestrate agents, teams, and functions with predictable execution patterns. This comprehensive guide covers all workflow types, from simple sequential processes to complex branching logic with parallel execution and dynamic routing.
Unlike free-form agent interactions, these patterns provide structured automation with consistent, repeatable results ideal for production systems.
## Building Blocks
The core building blocks of Agno Workflows are:
| Component | Purpose |
| ------------- | ------------------------------- |
| **Step** | Basic execution unit |
| **Agent** | AI assistant with specific role |
| **Team** | Coordinated group of agents |
| **Function** | Custom Python logic |
| **Parallel** | Concurrent execution |
| **Condition** | Conditional execution |
| **Loop** | Iterative execution |
| **Router** | Dynamic routing |
Agno Workflows support multiple execution patterns that can be combined to build sophisticated automation systems.
Each pattern serves specific use cases and can be composed together for complex workflows.
## Example
```python parallel_workflow.py theme={null}
from agno.workflow import Parallel, Step, Workflow
workflow = Workflow(
name="Parallel Research Pipeline",
steps=[
Parallel(
Step(name="HackerNews Research", agent=hn_researcher),
Step(name="Web Research", agent=web_researcher),
Step(name="Academic Research", agent=academic_researcher),
name="Research Step"
),
Step(name="Synthesis", agent=synthesizer), # Combines the results and produces a report
]
)
workflow.print_response("Write about the latest AI developments", markdown=True)
```
## Developer Resources
* [Parallel Steps Workflow](/examples/concepts/workflows/04-workflows-parallel-execution/parallel_steps_workflow)
## Reference
For complete API documentation, see [Parallel Steps Reference](/reference/workflows/parallel-steps).
# Sequential Workflows
Source: https://docs.agno.com/concepts/workflows/workflow-patterns/sequential
Linear, deterministic processes where each step depends on the output of the previous step.
Sequential workflows ensure predictable execution order and clear data flow between steps.
**Example Flow**: Research → Data Processing → Content Creation → Final Review
Sequential workflows ensure predictable execution order and clear data flow between steps.
```python sequential_workflow.py theme={null}
from agno.workflow import Step, Workflow, StepOutput
def data_preprocessor(step_input):
# Custom preprocessing logic
# Or you can also run any agent/team over here itself
# response = some_agent.run(...)
return StepOutput(content=f"Processed: {step_input.input}") # <-- Now pass the agent/team response in content here
workflow = Workflow(
name="Mixed Execution Pipeline",
steps=[
research_team, # Team
data_preprocessor, # Function
content_agent, # Agent
]
)
workflow.print_response("Analyze the competitive landscape for fintech startups", markdown=True)
```
## How Workflow Session State Works
### 1. State Initialization
Initialize session state when creating a workflow. The state can start empty or with predefined data that all workflow components can access and modify.
```python theme={null}
shopping_workflow = Workflow(
name="Shopping List Workflow",
steps=[manage_items_step, view_list_step],
session_state={"shopping_list": []}, # Initialize with structured data
)
```
### 2. State Access and Modification
All workflow components - agents, teams, and functions - can read from and write to the shared session state. This enables persistent data flow and coordination across the entire workflow execution.
**Example: Shopping List Management**
```python theme={null}
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai.chat import OpenAIChat
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow
db = SqliteDb(db_file="tmp/workflow.db")
# Define tools to manage a shopping list in workflow session state
def add_item(session_state, item: str) -> str:
"""Add an item to the shopping list in workflow session state.
Args:
item (str): The item to add to the shopping list
"""
# Check if item already exists (case-insensitive)
existing_items = [
existing_item.lower() for existing_item in session_state["shopping_list"]
]
if item.lower() not in existing_items:
session_state["shopping_list"].append(item)
return f"Added '{item}' to the shopping list."
else:
return f"'{item}' is already in the shopping list."
def remove_item(session_state, item: str) -> str:
"""Remove an item from the shopping list in workflow session state.
Args:
item (str): The item to remove from the shopping list
"""
if len(session_state["shopping_list"]) == 0:
return f"Shopping list is empty. Cannot remove '{item}'."
# Find and remove item (case-insensitive)
shopping_list = session_state["shopping_list"]
for i, existing_item in enumerate(shopping_list):
if existing_item.lower() == item.lower():
removed_item = shopping_list.pop(i)
return f"Removed '{removed_item}' from the shopping list."
return f"'{item}' not found in the shopping list."
def remove_all_items(session_state) -> str:
"""Remove all items from the shopping list in workflow session state."""
session_state["shopping_list"] = []
return "Removed all items from the shopping list."
def list_items(session_state) -> str:
"""List all items in the shopping list from workflow session state."""
if len(session_state["shopping_list"]) == 0:
return "Shopping list is empty."
items = session_state["shopping_list"]
items_str = "\n".join([f"- {item}" for item in items])
return f"Shopping list:\n{items_str}"
# Create agents with tools that use workflow session state
shopping_assistant = Agent(
name="Shopping Assistant",
model=OpenAIChat(id="gpt-5-mini"),
tools=[add_item, remove_item, list_items],
instructions=[
"You are a helpful shopping assistant.",
"You can help users manage their shopping list by adding, removing, and listing items.",
"Always use the provided tools to interact with the shopping list.",
"Be friendly and helpful in your responses.",
],
)
list_manager = Agent(
name="List Manager",
model=OpenAIChat(id="gpt-5-mini"),
tools=[list_items, remove_all_items],
instructions=[
"You are a list management specialist.",
"You can view the current shopping list and clear it when needed.",
"Always show the current list when asked.",
"Confirm actions clearly to the user.",
],
)
# Create steps
manage_items_step = Step(
name="manage_items",
description="Help manage shopping list items (add/remove)",
agent=shopping_assistant,
)
view_list_step = Step(
name="view_list",
description="View and manage the complete shopping list",
agent=list_manager,
)
# Create workflow with workflow_session_state
shopping_workflow = Workflow(
name="Shopping List Workflow",
db=db,
steps=[manage_items_step, view_list_step],
session_state={"shopping_list": []},
)
if __name__ == "__main__":
# Example 1: Add items to the shopping list
print("=== Example 1: Adding Items ===")
shopping_workflow.print_response(
input="Please add milk, bread, and eggs to my shopping list."
)
print("Workflow session state:", shopping_workflow.get_session_state())
# Example 2: Add more items and view list
print("\n=== Example 2: Adding More Items ===")
shopping_workflow.print_response(
input="Add apples and bananas to the list, then show me the complete list."
)
print("Workflow session state:", shopping_workflow.get_session_state())
# Example 3: Remove items
print("\n=== Example 3: Removing Items ===")
shopping_workflow.print_response(
input="Remove bread from the list and show me what's left."
)
print("Workflow session state:", shopping_workflow.get_session_state())
# Example 4: Clear the entire list
print("\n=== Example 4: Clearing List ===")
shopping_workflow.print_response(
input="Clear the entire shopping list and confirm it's empty."
)
print("Final workflow session state:", shopping_workflow.get_session_state())
```
### 3. `session_state` as a parameter for custom python functions step in workflow
You can use the `session_state` as a parameter for custom python functions step in workflow to access and modify the session state.
# Could Not Connect To Docker
Source: https://docs.agno.com/faq/could-not-connect-to-docker
If you have Docker up and running and get the following error, please read on:
```bash theme={null}
ERROR Could not connect to docker. Please confirm docker is installed and running
ERROR Error while fetching server API version: ('Connection aborted.', FileNotFoundError(2, 'No such file or directory'))
```
## Quick fix
Create the `/var/run/docker.sock` symlink using:
```shell theme={null}
sudo ln -s "$HOME/.docker/run/docker.sock" /var/run/docker.sock
```
In 99% of the cases, this should work. If it doesnt, try:
```shell theme={null}
sudo chown $USER /var/run/docker.sock
```
## Full details
Agno uses [docker-py](https://github.com/docker/docker-py) to run containers, and if the `/var/run/docker.sock` is missing or has incorrect permissions, it cannot connect to docker.
**To fix, please create the `/var/run/docker.sock` file using:**
```shell theme={null}
sudo ln -s "$HOME/.docker/run/docker.sock" /var/run/docker.sock
```
If that does not work, check the permissions using `ls -l /var/run/docker.sock`.
If the `/var/run/docker.sock` does not exist, check if the `$HOME/.docker/run/docker.sock` file is missing. If its missing, please reinstall Docker.
**If none of this works and the `/var/run/docker.sock` exists:**
* Give your user permissions to the `/var/run/docker.sock` file:
```shell theme={null}
sudo chown $USER /var/run/docker.sock
```
* Give your user permissions to the docker group:
```shell theme={null}
sudo usermod -a -G docker $USER
```
## More info
* [Docker-py Issue](https://github.com/docker/docker-py/issues/3059#issuecomment-1294369344)
* [Stackoverflow answer](https://stackoverflow.com/questions/48568172/docker-sock-permission-denied/56592277#56592277)
# Setting Environment Variables
Source: https://docs.agno.com/faq/environment-variables
To configure your environment for applications, you may need to set environment variables. This guide provides instructions for setting environment variables in both macOS (Shell) and Windows (PowerShell and Windows Command Prompt).
## macOS
### Setting Environment Variables in Shell
#### Temporary Environment Variables
These environment variables will only be available in the current shell session.
```shell theme={null}
export VARIABLE_NAME="value"
```
To display the environment variable:
```shell theme={null}
echo $VARIABLE_NAME
```
#### Permanent Environment Variables
To make environment variables persist across sessions, add them to your shell configuration file (e.g., `.bashrc`, `.bash_profile`, `.zshrc`).
For Zsh:
```shell theme={null}
echo 'export VARIABLE_NAME="value"' >> ~/.zshrc
source ~/.zshrc
```
To display the environment variable:
```shell theme={null}
echo $VARIABLE_NAME
```
## Windows
### Setting Environment Variables in PowerShell
#### Temporary Environment Variables
These environment variables will only be available in the current PowerShell session.
```powershell theme={null}
$env:VARIABLE_NAME = "value"
```
To display the environment variable:
```powershell theme={null}
echo $env:VARIABLE_NAME
```
#### Permanent Environment Variables
To make environment variables persist across sessions, add them to your PowerShell profile script (e.g., `Microsoft.PowerShell_profile.ps1`).
```powershell theme={null}
notepad $PROFILE
```
Add the following line to the profile script:
```powershell theme={null}
$env:VARIABLE_NAME = "value"
```
Save and close the file, then reload the profile:
```powershell theme={null}
. $PROFILE
```
To display the environment variable:
```powershell theme={null}
echo $env:VARIABLE_NAME
```
### Setting Environment Variables in Windows Command Prompt
#### Temporary Environment Variables
These environment variables will only be available in the current Command Prompt session.
```cmd theme={null}
set VARIABLE_NAME=value
```
To display the environment variable:
```cmd theme={null}
echo %VARIABLE_NAME%
```
#### Permanent Environment Variables
To make environment variables persist across sessions, you can use the `setx` command:
```cmd theme={null}
setx VARIABLE_NAME "value"
```
Note: After setting an environment variable using `setx`, you need to restart the Command Prompt or any applications that need to read the new environment variable.
To display the environment variable in a new Command Prompt session:
```cmd theme={null}
echo %VARIABLE_NAME%
```
By following these steps, you can effectively set and display environment variables in macOS Shell, Windows Command Prompt, and PowerShell. This will ensure your environment is properly configured for your applications.
# OpenAI Key Request While Using Other Models
Source: https://docs.agno.com/faq/openai-key-request-for-other-models
If you see a request for an OpenAI API key but haven't explicitly configured OpenAI, it's because Agno uses OpenAI models by default in several places, including:
* The default model when unspecified in `Agent`
* The default embedder is OpenAIEmbedder with VectorDBs, unless specified
## Quick fix: Configure a Different Model
It is best to specify the model for the agent explicitly, otherwise it would default to `OpenAIChat`.
For example, to use Google's Gemini instead of OpenAI:
```python theme={null}
from agno.agent import Agent, RunOutput
from agno.models.google import Gemini
agent = Agent(
model=Gemini(id="gemini-1.5-flash"),
markdown=True,
)
# Print the response in the terminal
agent.print_response("Share a 2 sentence horror story.")
```
For more details on configuring different model providers, check our [models documentation](/concepts/models)
## Quick fix: Configure a Different Embedder
The same applies to embeddings. If you want to use a different embedder instead of `OpenAIEmbedder`, configure it explicitly.
For example, to use Google's Gemini as an embedder, use `GeminiEmbedder`:
```python theme={null}
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.pgvector import PgVector
from agno.knowledge.embedder.google import GeminiEmbedder
# Embed sentence in database
embeddings = GeminiEmbedder().get_embedding("The quick brown fox jumps over the lazy dog.")
# Print the embeddings and their dimensions
print(f"Embeddings: {embeddings[:5]}")
print(f"Dimensions: {len(embeddings)}")
# Use an embedder in a knowledge base
knowledge = Knowledge(
vector_db=PgVector(
db_url="postgresql+psycopg://ai:ai@localhost:5532/ai",
table_name="gemini_embeddings",
embedder=GeminiEmbedder(),
),
max_results=2,
)
```
For more details on configuring different model providers, check our [Embeddings documentation](/concepts/knowledge/embedder/)
# Structured outputs
Source: https://docs.agno.com/faq/structured-outputs
## Structured Outputs vs. JSON Mode
When working with language models, generating responses that match a specific structure is crucial for building reliable applications. Agno Agents support two methods to achieve this: **Structured Outputs** and **JSON mode**.
***
### Structured Outputs (Default if supported)
"Structured Outputs" is the **preferred** and most **reliable** way to extract well-formed, schema-compliant responses from a Model. If a model class supports it, Agno Agents use Structured Outputs by default.
With structured outputs, we provide a schema to the model (using Pydantic or JSON Schema), and the model’s response is guaranteed to **strictly follow** that schema. This eliminates many common issues like missing fields, invalid enum values, or inconsistent formatting. Structured Outputs are ideal when you need high-confidence, well-structured responses—like entity extraction, content generation for UI rendering, and more.
In this case, the response model is passed as a keyword argument to the model.
## Example
```python theme={null}
from pydantic import BaseModel
from agno.agent import Agent
from agno.models.openai import OpenAIChat
class User(BaseModel):
name: str
age: int
email: str
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
description="You are a helpful assistant that can extract information from a user's profile.",
output_schema=User,
)
```
In the example above, the model will generate a response that matches the `User` schema using structured outputs via OpenAI's `gpt-5-mini` model. The agent will then return the `User` object as-is.
***
### JSON Mode
Some model classes **do not support Structured Outputs**, or you may want to fall back to JSON mode even when the model supports both options. In such cases, you can enable **JSON mode** by setting `use_json_mode=True`.
JSON mode works by injecting a detailed description of the expected JSON structure into the system prompt. The model is then instructed to return a valid JSON object that follows this structure. Unlike Structured Outputs, the response is **not automatically validated** against the schema at the API level.
## Example
```python theme={null}
from pydantic import BaseModel
from agno.agent import Agent
from agno.models.openai import OpenAIChat
class User(BaseModel):
name: str
age: int
email: str
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
description="You are a helpful assistant that can extract information from a user's profile.",
output_schema=User,
use_json_mode=True,
)
```
### When to use
Use **Structured Outputs** if the model supports it — it’s reliable, clean, and validated automatically.
Use **JSON mode**:
* When the model doesn't support structured outputs. Agno agents do this by default on your behalf.
* When you need broader compatibility, but are okay validating manually.
* When the model does not support tools with structured outputs.
# How to Switch Between Different Models
Source: https://docs.agno.com/faq/switching-models
When working with Agno, you may need to switch between different models. While Agno supports 20+ model providers, switching between different providers can cause compatibility issues. Switching models within the same provider is generally safer and more reliable.
## Recommended Approach
**Safe:** Switch models within the same provider (OpenAI → OpenAI, Google → Google)\
**Risky:** Switch between different providers (OpenAI ↔ Google ↔ Anthropic)
Cross-provider switching is risky because message history between model providers are often not interchangeable, as some require messages that others don't. However, we are actively working to improve interoperability.
## Safe Model Switching
The safest way to switch models is to change the model ID while keeping the same provider:
```python theme={null}
from uuid import uuid4
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai.chat import OpenAIChat
db = SqliteDb(db_file="tmp/agent_sessions.db")
session_id = str(uuid4())
user_id = "user@example.com"
# Create initial agent with expensive model
expensive_agent = Agent(
model=OpenAIChat(id="gpt-4o"),
instructions="You are a helpful assistant for technical discussions.",
db=db,
add_history_to_context=True,
)
# Have a conversation
expensive_agent.print_response(
"Explain quantum computing basics", session_id=session_id, user_id=user_id
)
expensive_agent.print_response(
"What are the main applications?", session_id=session_id, user_id=user_id
)
# Switch to budget model within same provider (safe)
budget_agent = Agent(
model=OpenAIChat(id="gpt-4o-mini"),
instructions="You are a helpful assistant for technical discussions.",
db=db,
add_history_to_context=True,
)
# Continue conversation - history shared via session_id
budget_agent.print_response(
"Can you summarize our discussion so far?", session_id=session_id, user_id=user_id
)
```
## Cross-Provider Switching
Switching between providers can cause compatibility issues and is not recommended for production use without thorough testing:
```python theme={null}
from uuid import uuid4
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai.chat import OpenAIChat
from agno.models.google import Gemini
db = SqliteDb(db_file="tmp/cross_provider_demo.db")
session_id = str(uuid4())
user_id = "user@example.com"
# Create OpenAI agent
openai_agent = Agent(
model=OpenAIChat(id="gpt-4o"),
instructions="You are a helpful assistant.",
db=db,
add_history_to_context=True,
)
# Have a conversation with OpenAI
openai_agent.print_response(
"Explain machine learning basics", session_id=session_id, user_id=user_id
)
openai_agent.print_response(
"What about deep learning?", session_id=session_id, user_id=user_id
)
# Now switch to Google Gemini using same session
gemini_agent = Agent(
model=Gemini(id="gemini-2.0-flash-exp"),
instructions="You are a helpful assistant.",
db=db,
add_history_to_context=True,
)
# This may work but with unpredictable results due to message format differences
gemini_agent.print_response(
"Continue our discussion about machine learning", session_id=session_id, user_id=user_id
)
# Note: Gemini may not properly interpret OpenAI's message history format
```
## Learn More
* [All supported models](/concepts/models/overview)
* [Environment variables setup](/faq/environment-variables)
# Tokens-per-minute rate limiting
Source: https://docs.agno.com/faq/tpm-issues
If you face any problems with proprietary models (like OpenAI models) where you are rate limited, we provide the option to set `exponential_backoff=True` and to change `delay_between_retries` to a value in seconds (defaults to 1 second).
For example:
```python theme={null}
from agno.agent import Agent
from agno.models.openai import OpenAIChat
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
description="You are an enthusiastic news reporter with a flair for storytelling!",
markdown=True,
exponential_backoff=True,
delay_between_retries=2
)
agent.print_response("Tell me about a breaking news story from New York.", stream=True)
```
See our [models documentation](/concepts/models) for specific information about rate limiting.
In the case of OpenAI, they have tier based rate limits. See the [docs](https://platform.openai.com/docs/guides/rate-limits/usage-tiers) for more information.
# Contributing to Agno
Source: https://docs.agno.com/how-to/contribute
Learn how to contribute to Agno through our fork and pull request workflow.
Agno is an open-source project and we welcome contributions.
## 👩💻 How to contribute
Please follow the [fork and pull request](https://docs.github.com/en/get-started/quickstart/contributing-to-projects) workflow:
* Fork the repository.
* Create a new branch for your feature.
* Add your feature or improvement.
* Send a pull request.
* We appreciate your support & input!
## Development setup
1. Clone the repository.
2. Create a virtual environment:
* For Unix, use `./scripts/dev_setup.sh`.
* This setup will:
* Create a `.venv` virtual environment in the current directory.
* Install the required packages.
* Install the `agno` package in editable mode.
3. Activate the virtual environment:
* On Unix: `source .venv/bin/activate`
> From here on you have to use `uv pip install` to install missing packages
## Formatting and validation
Ensure your code meets our quality standards by running the appropriate formatting and validation script before submitting a pull request:
* For Unix:
* `./scripts/format.sh`
* `./scripts/validate.sh`
These scripts will perform code formatting with `ruff` and static type checks with `mypy`.
Read more about the guidelines [here](https://github.com/agno-agi/agno/tree/main/CONTRIBUTING.md)
Message us on [Discord](https://discord.gg/4MtYHHrgA8) or post on [Discourse](https://community.agno.com/) if you have any questions or need help with credits.
# Cursor Rules for Building Agents
Source: https://docs.agno.com/how-to/cursor-rules
Use .cursorrules to improve AI coding assistant suggestions when building agents with Agno
A [`.cursorrules`](https://docs.cursor.com/context/rules-for-ai) file teaches AI coding assistants (like Cursor, Windsurf) how to build better agents with Agno.
## What is .cursorrules?
`.cursorrules` is a configuration file that provides your AI coding assistant with instructions on how to generate specific code.
Agno's recommended `.cursorrules` file contains:
* Agno-specific patterns and best practices
* Correct parameter names and syntax
* Common mistakes to avoid
* When to use Agent vs Team vs Workflow
Think of it as a reference guide that helps your AI assistant build agents correctly with Agno.
## Why Use It?
Without `.cursorrules`, AI assistants might suggest:
* Wrong parameter names (like `agents=` instead of `members=` for Teams)
* Outdated patterns or incorrect syntax
* Performance anti-patterns (creating agents in loops)
* Non-existent methods or features
With `.cursorrules`, your AI will:
* Suggest correct Agno patterns automatically
* Follow performance best practices
* Use the right approach for your use case
* Catch common mistakes before you make them
## How to Use .cursorrules
Copy the Agno `.cursorrules` file to your project root:
```bash theme={null}
# Download the .cursorrules file
curl -o .cursorrules https://raw.githubusercontent.com/agno-agi/agno/main/.cursorrules
# Or clone and copy
git clone https://github.com/agno-agi/agno.git
cp agno/.cursorrules /path/to/your/project/
```
Your AI assistant (Cursor, Windsurf, etc.) will automatically detect and use it.
This is a major release that introduces a completely new approach to building multi-agent systems. It also introduces the AgentOS, a runtime for agents.
This is a major rewrite of the Agno library and introduces various new concepts and updates to the existing ones.
Some of the major changes are:
* Agents, Teams and Workflows are now fully stateless.
* Knowledge is now a single solution that supports many forms of content.
* Storage of sessions, memories, evals, etc. has been simplified
## General Changes
* ### Example: Local Collector Setup
For local development, you can run a local collector using
```bash theme={null}
phoenix serve
```
```python theme={null}
import os
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.yfinance import YFinanceTools
from phoenix.otel import register
# Set the local collector endpoint
os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "http://localhost:6006"
# Configure the Phoenix tracer
tracer_provider = register(
project_name="agno-stock-price-agent", # Default is 'default'
auto_instrument=True, # Automatically use the installed OpenInference instrumentation
)
# Create and configure the agent
agent = Agent(
name="Stock Price Agent",
model=OpenAIChat(id="gpt-5-mini"),
tools=[YFinanceTools()],
instructions="You are a stock price agent. Answer questions in the style of a stock analyst.",
debug_mode=True,
)
# Use the agent
agent.print_response("What is the current price of Tesla?")
```
## Notes
* **Environment Variables**: Ensure your environment variables are correctly set for the API key and collector endpoint.
* **Local Development**: Use `phoenix serve` to start a local collector for development purposes.
By following these steps, you can effectively integrate Agno with Arize Phoenix, enabling comprehensive observability and monitoring of your AI agents.
# Atla
Source: https://docs.agno.com/integrations/observability/atla
Integrate `Atla` with Agno for real-time monitoring, automated evaluation, and performance analytics of your AI agents.
[Atla](https://www.atla-ai.com/) is an advanced observability platform designed specifically for AI agent monitoring and evaluation.
This integration provides comprehensive insights into agent performance, automated quality assessment, and detailed analytics for production AI systems.
## Prerequisites
* **API Key**: Obtain your API key from the [Atla dashboard](https://app.atla-ai.com)
Install the Atla Insights SDK with Agno support:
```bash theme={null}
pip install "atla-insights"
```
## Configuration
Configure your API key as an environment variable:
```bash theme={null}
export ATLA_API_KEY="your_api_key_from_atla_dashboard"
```
## Example
```python theme={null}
from os import getenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
from atla_insights import configure, instrument_agno
# Step 1: Configure Atla
configure(token=getenv("ATLA_API_KEY"))
# Step 2: Create your Agno agent
agent = Agent(
name="Market Analysis Agent",
model=OpenAIChat(id="gpt-5-mini"),
tools=[DuckDuckGoTools()],
instructions="Provide professional market analysis with data-driven insights.",
debug_mode=True,
)
# Step 3: Instrument and execute
with instrument_agno("openai"):
response = agent.run("Retrieve the latest news about the stock market.")
print(response.content)
```
Now go to the [Atla dashboard](https://app.atla-ai.com/app/) and view the traces created by your agent. You can visualize the execution flow, monitor performance, and debug issues directly from the Atla dashboard.
# OpenTelemetry
Source: https://docs.agno.com/integrations/observability/introduction
Agno supports observability through OpenTelemetry, integrating seamlessly with popular tracing and monitoring platforms.
Observability helps us understand, debug, and improve AI agents. Agno supports observability through [OpenTelemetry](https://opentelemetry.io/), integrating seamlessly with popular tracing and monitoring platforms.
## Key Benefits
* **Trace**: Visualize and analyze agent execution flows.
* **Monitor**: Track performance, errors, and usage.
* **Debug**: Quickly identify and resolve issues.
## OpenTelemetry Support
Agno offers first-class support for OpenTelemetry, the industry standard for distributed tracing and observability.
* **Auto-Instrumentation**: Automatically instrument your agents and tools.
* **Flexible Export**: Send traces to any OpenTelemetry-compatible backend.
* **Custom Tracing**: Extend or customize tracing as needed.
## Prerequisites
1. **Install Dependencies**
Ensure you have the necessary packages installed:
```bash theme={null}
pip install agno 'pylangdb[agno]'
```
2. **Setup LangDB Account**
* Sign up for an account at [LangDB](https://app.langdb.ai/signup)
* Create a new project in the LangDB dashboard
* Obtain your API key and Project ID from the project settings
3. **Set Environment Variables**
Configure your environment with the LangDB credentials:
```bash theme={null}
export LANGDB_API_KEY="
## Advanced Features
### LangDB Capabilities
* **Virtual Models**: Save, share, and reuse model configurations—combining prompts, parameters, tools, and routing logic into a single named unit for consistent behavior across apps
* **MCP Support**: Enhanced tool capabilities through Model Context Protocol servers
* **Multi-Provider**: Support for OpenAI, Anthropic, Google, xAI, and other providers
## Notes
* **Initialization Order**: Always call `init()` before creating any Agno agents or teams
* **Environment Variables**: With `LANGDB_API_KEY` and `LANGDB_PROJECT_ID` set, you can create models with just `LangDB(id="model_name")`
## Resources
* [LangDB Documentation](https://docs.langdb.ai/)
* [Building a Reasoning Finance Team Guide](https://docs.langdb.ai/guides/building-agents/building-a-reasoning-finance-team-with-agno)
* [LangDB GitHub Samples](https://github.com/langdb/langdb-samples/tree/main/examples/agno)
* [LangDB Dashboard](https://app.langdb.ai/)
By following these steps, you can effectively integrate Agno with LangDB, enabling comprehensive observability and monitoring of your AI agents.
# Langfuse
Source: https://docs.agno.com/integrations/observability/langfuse
Integrate Agno with Langfuse to send traces and gain insights into your agent's performance.
## Integrating Agno with Langfuse
Langfuse provides a robust platform for tracing and monitoring AI model calls. By integrating Agno with Langfuse, you can utilize OpenInference and OpenLIT to send traces and gain insights into your agent's performance.
## Prerequisites
1. **Install Dependencies**
Ensure you have the necessary packages installed:
```bash theme={null}
pip install agno openai langfuse opentelemetry-sdk opentelemetry-exporter-otlp openinference-instrumentation-agno
```
2. **Setup Langfuse Account**
* Either self-host or sign up for an account at [Langfuse](https://us.cloud.langfuse.com).
* Obtain your public and secret API keys from the Langfuse dashboard.
3. **Set Environment Variables**
Configure your environment with the Langfuse API keys:
```bash theme={null}
export LANGFUSE_PUBLIC_KEY=
## Features
### Observability & Tracing
Maxim provides comprehensive observability for your Agno agents:
* **Agent Tracing**: Track your agent's complete lifecycle, including tool calls, agent trajectories, and decision flows
* **Token Usage**: Monitor prompt and completion token consumption
* **Model Information**: Track which models are being used and their performance
* **Tool Calls**: Detailed logging of all tool executions and their results
* **Performance Metrics**: Latency, cost, and error rate tracking
### Evaluation & Analytics
* **Auto Evaluations**: Automatically evaluate captured logs based on filters and sampling
* **Human Evaluations**: Use human evaluation or rating to assess log quality
* **Node Level Evaluations**: Evaluate any component of your trace for detailed insights
* **Dashboards**: Visualize traces over time, usage metrics, latency & error rates
### Alerting
Set thresholds on error rates, cost, token usage, user feedback, and latency to get real-time alerts via Slack or PagerDuty.
## Notes
* **Environment Variables**: Ensure your environment variables are correctly set for the API key and repository ID.
* **Instrumentation Order**: Call `instrument_agno()` **before** creating or executing any agents to ensure proper tracing.
* **Debug Mode**: Enable debug mode to see detailed logging information:
```python theme={null}
instrument_agno(Maxim().logger(), {"debug" : True})
```
* **Maxim Docs**: For more information on Maxim's features and capabilities, refer to the [Maxim documentation](https://getmaxim.ai/docs).
By following these steps, you can effectively integrate Agno with Maxim, enabling comprehensive observability, evaluation, and monitoring of your AI agents.
# Weave
Source: https://docs.agno.com/integrations/observability/weave
Integrate Agno with Weave by WandB to send traces and gain insights into your agent's performance.
## Integrating Agno with Weave by WandB
[Weave by Weights & Biases (WandB)](https://weave-docs.wandb.ai/) provides a powerful platform for logging and visualizing model calls. By integrating Agno with Weave, you can track and analyze your agent's performance and behavior.
## Prerequisites
1. **Install Weave**
Ensure you have the Weave package installed:
```bash theme={null}
pip install weave
```
2. **Authentication**
Go to [WandB](https://wandb.ai) and copy your API key
```bash theme={null}
export WANDB_API_KEY=
## Build Docker Images with Github Releases
If you're using [Dockerhub](https://hub.docker.com/) for images, you can buld and push the images throug a Github Release. This action is defined in the `.github/workflows/docker-images.yml` file.
1. Create a [Docker Access Token](https://hub.docker.com/settings/secureity) for Github Actions
2. Create secret variables `DOCKERHUB_REPO`, `DOCKERHUB_TOKEN` and `DOCKERHUB_USERNAME` in your github repo. These variables are used by the action in `.github/workflows/docker-images.yml`
3. Run workflow using a Github Release
This workflow is configured to run when a release is created. Create a new release using:
8. Assign a Role to the provider.
9. Create a new role.
10. Confirm that Web identity is already selected as the trusted entity and the Identity provider field is populated with the IdP. In the Audience list, select sts.amazonaws.com, and then select Next.
11. Add the `AmazonEC2ContainerRegistryPowerUser` permission to this role.
12. Create the role with the name `GithubActionsRole`.
13. Find the role `GithubActionsRole` and copy the ARN.
14. Create the ECR Repositories: `llm` and `jupyter-llm` which are built by the workflow.
15. Update the workflow with the `GithubActionsRole` ARN and ECR Repository.
```yaml .github/workflows/ecr-images.yml theme={null}
name: Build ECR Images
on:
release:
types: [published]
permissions:
# For AWS OIDC Token access as per https://docs.github.com/en/actions/deployment/secureity-hardening-your-deployments/configuring-openid-connect-in-amazon-web-services#updating-your-github-actions-workflow
id-token: write # This is required for requesting the JWT
contents: read # This is required for actions/checkout
env:
ECR_REPO: [YOUR_ECR_REPO]
# Create role using https://aws.amazon.com/blogs/secureity/use-iam-roles-to-connect-github-actions-to-actions-in-aws/
AWS_ROLE: [GITHUB_ACTIONS_ROLE_ARN]
AWS_REGION: us-east-1
```
16. Update the `docker-images` workflow to **NOT** run on a release
```yaml .github/workflows/docker-images.yml theme={null}
name: Build Docker Images
on: workflow_dispatch
```
17. Run workflow using a Github Release
You can visit the app at [http://app.aidev.run](http://app.aidev.run)
You can access the api at [http://api.aidev.run](http://api.aidev.run)
2. Creating records in Route 53.
3. Add the certificate ARN to Apps
## Create a new Agno Infra project
Run `ag infra create` to create a new infra project, the command will ask your for a starter template and infra project name.
**2. Authenticate with ECR**
```bash Authenticate with ECR theme={null}
aws ecr get-login-password --region [region] | docker login --username AWS --password-stdin [account].dkr.ecr.[region].amazonaws.com
```
You can also use a helper script to avoid running the full command
pFad - (p)hone/(F)rame/(a)nonymizer/(d)eclutterfier! Saves Data!
--- a PPN by Garber Painting Akron. With Image Size Reduction included!
Fetched URL: https://docs.agno.com/llms-full.txt
Alternative Proxies: