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:
## Managing Knowledge via AgentOS control plane
Once your Knowledge bases are attached to AgentOS, you can:
* **View Content**: Browse and search through your Knowledge base contents
* **Add Content**: Upload new documents, add URLs, or input text directly
* **Edit Content**: Modify metadata on existing Knowledge entries
* **Delete Content**: Remove outdated or incorrect information
## Best Practices
* **Separate Knowledge by Domain**: Create separate Knowledge bases for different topics (e.g., technical docs, FAQs, policies)
* **Consistent Naming**: Use descriptive names for your Knowledge bases that reflect their content
* **Regular Updates**: Keep your Knowledge bases current by regularly adding new content and removing outdated information
* **Monitor Performance**: Use different table names for vector storage to avoid conflicts
* **Content Organization**: Use the `name` parameter when adding content to make it easily identifiable
## Troubleshooting
### 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 Security, check out the [AgentOS Security](/agent-os/security) 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_events=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/blob/main/cookbook/knowledge/filters/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,
)
```
# 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/security). ```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:///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:///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) # Notion Tools Source: https://docs.agno.com/concepts/tools/toolkits/others/notion The NotionTools toolkit enables Agents to interact with your Notion pages. Notion is a powerful workspace management tool that allows you to create, organize, and collaborate on your work. Agno provides tools for interacting with Notion databases to help you automate your workflows and streamline your work. ## Prerequisites To use `NotionTools`, you need to install `notion-client`: ```shell theme={null} pip install notion-client ``` ### Step 1: Create a Notion Integration 1. Go to [https://www.notion.so/my-integrations](https://www.notion.so/my-integrations) 2. Click on **"+ New integration"** 3. Fill in the details: * **Name**: Give it a name like "Agno Query Classifier" * **Associated workspace**: Select your workspace * **Type**: Internal integration 4. Click **"Submit"** 5. Copy the **"Internal Integration Token"** (starts with `secret_`) * ⚠️ Keep this secret! This is your `NOTION_API_KEY` ### Step 2: Create a Notion Database 1. Open Notion and create a new page 2. Add a **Database** (you can use "/database" command) 3. Set up the database with these properties: * **Name** (Title) - Already exists by default * **Tag** (Select) - Click "+" to add a new property * Property type: **Select** * Property name: **Tag** * Add these options: * travel * tech * general-blogs * fashion * documents ### Step 3: Share Database with Your Integration 1. Open your database page in Notion 2. Click the **"..."** (three dots) menu in the top right 3. Scroll down and click **"Add connections"** 4. Search for your integration name (e.g., "Agno Query Classifier") 5. Click on it to grant access ### Step 4: Get Your Database ID Your database ID is in the URL of your database page: ``` https://www.notion.so/../{database_id}?v={view_id} ``` The `database_id` is the 32-character string (with hyphens) between the workspace name and the `?v=`. Example: ``` https://www.notion.so/myworkspace/28fee27fd9128039b3f8f47cb7ade7cb?v=... ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ This is your database_id ``` Copy this database ID. ```shell theme={null} export NOTION_API_KEY=your_api_key_here export NOTION_DATABASE_ID=your_database_id_here ``` ## Example The following example demonstrates how to use `NotionTools` to create, update and search for Notion pages with specific tags. ```python theme={null} from agno.agent import Agent from agno.tools.notion import NotionTools # Create an agent with Notion Tools notion_agent = Agent( name="Notion Knowledge Manager", instructions=[ "You are a smart assistant that helps organize information in Notion.", "When given content, analyze it and categorize it appropriately.", "Available categories: travel, tech, general-blogs, fashion, documents", "Always search first to avoid duplicate pages with the same tag.", "Be concise and helpful in your responses.", ], tools=[NotionTools()], markdown=True, ) def demonstrate_tools(): print(" Notion Tools Demonstration\n") print("=" * 60) # Example 1: Travel Notes print("\n Example 1: Organizing Travel Information") print("-" * 60) prompt = """ I found this amazing travel guide: 'Ha Giang Loop in Vietnam - 4 day motorcycle adventure through stunning mountains. Best time to visit: October to March. Must-see spots include Ma Pi Leng Pass.' Save this to Notion under the travel category. """ notion_agent.print_response(prompt) # Example 2: Tech Bookmarks print("\n Example 2: Saving Tech Articles") print("-" * 60) prompt = """ Save this tech article to Notion: 'The Rise of AI Agents in 2025 - How autonomous agents are revolutionizing software development. Key trends include multi-agent systems, agentic workflows, and AI-powered automation.' Categorize this appropriately and add to Notion. """ notion_agent.print_response(prompt) # Example 3: Multiple Items print("\n Example 3: Batch Processing Multiple Items") print("-" * 60) prompt = """ I need to save these items to Notion: 1. 'Best fashion trends for spring 2025 - Sustainable fabrics and minimalist designs' 2. 'My updated resume and cover letter for job applications' 3. 'Quick thoughts on productivity hacks for remote work' Process each one and save them to the appropriate categories. """ notion_agent.print_response(prompt) # Example 4: Search and Update print("\n🔍 Example 4: Finding and Updating Existing Content") print("-" * 60) prompt = """ Search for any pages tagged 'tech' and let me know what you find. Then add this new insight to one of them: 'Update: AI agents now support structured output with Pydantic models for better type safety.' """ notion_agent.print_response(prompt) # Example 5: Smart Categorization print("\n Example 5: Automatic Smart Categorization") print("-" * 60) prompt = """ I have this content but I'm not sure where it belongs: 'Exploring the ancient temples of Angkor Wat in Cambodia. The sunrise view from Angkor Wat is breathtaking. Best visited during the dry season from November to March.' Analyze this content, decide the best category, and save it to Notion. """ notion_agent.print_response(prompt) print("\n" + "=" * 60) print( "\nYour Notion database now contains organized content across different categories." ) print("Check your Notion workspace to see the results!") if __name__ == "__main__": demonstrate_tools() ``` ## Toolkit Functions These are the functions exposed by `NotionTools`: | Function | Description | | ---------------------------------- | ------------------------------------------------------------------------- | | `create_page(title, tag, content)` | Creates a new page in the Notion database with a title, tag, and content. | | `update_page(page_id, content)` | Adds content to an existing Notion page by appending a new paragraph. | | `search_pages(tag)` | Searches for pages in the database by tag and returns matching results. | ## Toolkit Params These parameters are passed to the `NotionTools` constructor: | Parameter | Type | Default | Description | | --------------------- | --------------- | ------- | --------------------------------------------------------------------------------------- | | `api_key` | `Optional[str]` | `None` | Notion API key (integration token). Uses `NOTION_API_KEY` env var if not provided. | | `database_id` | `Optional[str]` | `None` | The ID of the database to work with. Uses `NOTION_DATABASE_ID` env var if not provided. | | `enable_create_page` | `bool` | `True` | Enable the create\_page tool. | | `enable_update_page` | `bool` | `True` | Enable the update\_page tool. | | `enable_search_pages` | `bool` | `True` | Enable the search\_pages tool. | | `all` | `bool` | `False` | Enable all tools. Overrides individual enable flags when `True`. | ## Developer Resources * View [Tools](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/tools/notion.py) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/tools/notion_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 = "
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 MongoVectorDb # 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.
## 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())
```
## Workflow History for Conversational Workflows:
Similar to [workflow history for steps](/concepts/workflows/workflow_with_history), the `WorkflowAgent` has access to the full history of workflow runs for the current session.
This makes it possible to answer questions about previous results, compare outputs from multiple runs, and maintain conversation continuity.
## Example
```python theme={null}
from agno.workflow import Step, Workflow, StepInput, StepOutput
def security_gate(step_input: StepInput) -> StepOutput:
"""Security gate that stops deployment if vulnerabilities found"""
security_result = step_input.previous_step_content or ""
if "VULNERABLE" in security_result.upper():
return StepOutput(
content="🚨 SECURITY ALERT: Critical vulnerabilities detected. Deployment blocked.",
stop=True # Stop the entire workflow
)
else:
return StepOutput(
content="✅ Security check passed. Proceeding with deployment...",
stop=False
)
# Secure deployment pipeline
workflow = Workflow(
name="Secure Deployment Pipeline",
steps=[
Step(name="Security Scan", agent=security_scanner),
Step(name="Security Gate", executor=security_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 security 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 framework 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
## Direct User Interaction
When users interact directly with your workflow (rather than calling it programmatically), you can make it **conversational** by adding a `WorkflowAgent`. This enables natural chat-like interactions where the workflow intelligently decides whether to answer from previous results or execute new workflow runs. Learn more in the [Conversational Workflows](/concepts/workflows/conversational-workflows) guide.
## 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)
```
## Class-based executor
You can also use a class-based executor by defining a class that implements the `__call__` method.
```python theme={null}
class CustomExecutor:
def __call__(self, step_input: StepInput) -> StepOutput:
# 1. Custom preprocessing
# 2. Call agents/teams as needed
# 3. Custom postprocessing
return StepOutput(content=enhanced_content)
content_planning_step = Step(
name="Content Planning Step",
executor=CustomExecutor(),
)
```
**When is this useful?:**
* **Configuration at initialization**: Pass in settings, API keys, or behavior flags when creating the executor
* **Stateful execution**: Maintain counters, caches, or track information across multiple workflow runs
* **Reusable components**: Create configured executor instances that can be shared across multiple workflows
```python theme={null}
class CustomExecutor:
def __init__(self, max_retries: int = 3, use_cache: bool = True):
# Configuration passed during instantiation
self.max_retries = max_retries
self.use_cache = use_cache
self.call_count = 0 # Stateful tracking
def __call__(self, step_input: StepInput) -> StepOutput:
self.call_count += 1
# Access instance configuration and state
if self.use_cache and self.call_count > 1:
return StepOutput(content="Using cached result")
# Your custom logic with access to self.max_retries, etc.
return StepOutput(content=enhanced_content)
# Instantiate with specific configuration
content_planning_step = Step(
name="Content Planning Step",
executor=CustomExecutor(max_retries=5, use_cache=False),
)
```
Also supports async execution by defining the `__call__` method to be an async function.
```python theme={null}
class CustomExecutor:
async def __call__(self, step_input: StepInput) -> StepOutput:
# 1. Custom preprocessing
# 2. Call agents/teams as needed
# 3. Custom postprocessing
return StepOutput(content=enhanced_content)
content_planning_step = Step(
name="Content Planning Step",
executor=CustomExecutor(),
)
```
For a detailed example see [Class-based Executor](/examples/concepts/workflows/01-basic-workflows/class_based_executor).
## 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_events=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)
```
## Handling Session State Data in Parallel Steps
When using custom Python functions in your steps, you can access and update the Worfklow session state via the `run_context` parameter.
If you are performing session state updates in Parallel Steps, be aware that concurrent access to shared state will require coordination to avoid race conditions.
## 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 session 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. Access and Modify State Data
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.
From tools, you can access the session state via `run_context.session_state`.
**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
from agno.run import RunContext
db = SqliteDb(db_file="tmp/workflow.db")
# Define tools to manage a shopping list in workflow session state
def add_item(run_context: RunContext, 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
"""
if not run_context.session_state:
run_context.session_state = {}
# Check if item already exists (case-insensitive)
existing_items = [
existing_item.lower() for existing_item in run_context.session_state["shopping_list"]
]
if item.lower() not in existing_items:
run_context.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(run_context: RunContext, 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 not run_context.session_state:
run_context.session_state = {}
if len(run_context.session_state["shopping_list"]) == 0:
return f"Shopping list is empty. Cannot remove '{item}'."
# Find and remove item (case-insensitive)
shopping_list = run_context.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(run_context: RunContext) -> str:
"""Remove all items from the shopping list in workflow session state."""
if not run_context.session_state:
run_context.session_state = {}
run_context.session_state["shopping_list"] = []
return "Removed all items from the shopping list."
def list_items(run_context: RunContext) -> str:
"""List all items in the shopping list from workflow session state."""
if not run_context.session_state:
run_context.session_state = {}
if len(run_context.session_state["shopping_list"]) == 0:
return "Shopping list is empty."
items = run_context.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())
```
See the [RunContext schema](/reference/run/run_context) for more information.
### 3. `run_context` as a parameter for custom python functions step in workflow
You can add the `run_context` parameter to the Python functions you use as custom steps.
The `run_context` object will be automatically injected when running the function.
You can use it to read and modify the session state, via `run_context.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.
# LangDB
Source: https://docs.agno.com/integrations/observability/langdb
Integrate Agno with LangDB to trace agent execution, tool calls, and gain comprehensive observability into your agent's performance.
## Integrating Agno with LangDB
[LangDB](https://langdb.ai/) provides an AI Gateway platform for comprehensive observability and tracing of AI agents and LLM interactions. By integrating Agno with LangDB, you can gain full visibility into your agent's operations, including agent runs, tool calls, team interactions, and performance metrics.
For detailed integration instructions, see the [LangDB Agno documentation](https://docs.langdb.ai/getting-started/working-with-agent-frameworks/working-with-agno).
## 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.
# OpenLIT
Source: https://docs.agno.com/integrations/observability/openlit
Integrate Agno with OpenLIT for OpenTelemetry-native observability, tracing, and monitoring of your AI agents.
## Integrating Agno with OpenLIT
[OpenLIT](https://github.com/openlit/openlit) is an open-source, self-hosted, OpenTelemetry-native platform for a continuous feedback loop for testing, tracing, and fixing AI agents. By integrating Agno with OpenLIT, you can automatically instrument your agents to gain full visibility into LLM calls, tool usage, costs, performance metrics, and errors.
## Prerequisites
1. **Install Dependencies**
Ensure you have the necessary packages installed:
```bash theme={null}
pip install agno openai openlit
```
2. **Deploy OpenLIT**
OpenLIT is open-source and self-hosted. Quick start with Docker:
```bash theme={null}
git clone https://github.com/openlit/openlit
cd openlit
docker-compose up -d
```
Access the dashboard at `http://127.0.0.1:3000` with default credentials (username: `user@openlit.io`, password: `openlituser`).
**Other Deployment Options:**
For production deployments, Kubernetes with Helm, or other infrastructure setups, see the [OpenLIT Installation Guide](https://docs.openlit.io/latest/openlit/installation) for detailed instructions on:
* Kubernetes deployment with Helm charts
* Custom Docker configurations
* Reusing existing ClickHouse or OpenTelemetry Collector infrastructure
* OpenLIT Operator for zero-code instrumentation in Kubernetes
3. **Set Environment Variables (Optional)**
Configure the OTLP endpoint based on your deployment:
```bash theme={null}
# Local deployment
export OTEL_EXPORTER_OTLP_ENDPOINT="http://127.0.0.1:4318"
# Self-hosted on your infrastructure
export OTEL_EXPORTER_OTLP_ENDPOINT="http://your-openlit-host:4318"
```
## Sending Traces to OpenLIT
### Example: Basic Agent Setup
This example demonstrates how to instrument your Agno agent with OpenLIT for automatic tracing.
```python theme={null}
import openlit
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.yfinance import YFinanceTools
# Initialize OpenLIT instrumentation
openlit.init(
otlp_endpoint="http://127.0.0.1:4318" # Your OpenLIT OTLP endpoint
)
# Create and configure the agent
agent = Agent(
name="Stock Price Agent",
model=OpenAIChat(id="gpt-4o-mini"),
tools=[YFinanceTools(stock_price=True, analyst_recommendations=True)],
instructions="You are a stock price agent. Answer questions in the style of a stock analyst.",
show_tool_calls=True,
)
# Use the agent - all calls are automatically traced
agent.print_response("What is the current price of Tesla and what do analysts recommend?")
```
### Example: Development Mode (Console Output)
For local development without a collector, OpenLIT can output traces directly to the console:
```python theme={null}
import openlit
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
# Initialize OpenLIT without OTLP endpoint for console output
openlit.init()
# Create and configure the agent
agent = Agent(
name="Web Search Agent",
model=OpenAIChat(id="gpt-4o-mini"),
tools=[DuckDuckGoTools()],
instructions="Search the web and provide comprehensive answers.",
markdown=True,
)
# Use the agent - traces will be printed to console
agent.print_response("What are the latest developments in AI agents?")
```
### Example: Multi-Agent Team Tracing
OpenLIT automatically traces complex multi-agent workflows:
```python theme={null}
import openlit
from agno.agent import Agent
from agno.team import Team
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools
# Initialize OpenLIT instrumentation
openlit.init(otlp_endpoint="http://127.0.0.1:4318")
# Research Agent
research_agent = Agent(
name="Market Research Agent",
model=OpenAIChat(id="gpt-4o-mini"),
tools=[DuckDuckGoTools()],
instructions="Research current market conditions and news",
)
# Financial Analysis Agent
finance_agent = Agent(
name="Financial Analyst",
model=OpenAIChat(id="gpt-4o-mini"),
tools=[YFinanceTools(stock_price=True, company_info=True)],
instructions="Perform quantitative financial analysis",
)
# Coordinated Team
finance_team = Team(
name="Finance Research Team",
model=OpenAIChat(id="gpt-4o-mini"),
members=[research_agent, finance_agent],
instructions=[
"Collaborate to provide comprehensive financial insights",
"Consider both fundamental analysis and market sentiment",
],
)
# Execute team workflow - all agent interactions are traced
finance_team.print_response("Analyze Apple (AAPL) investment potential")
```
### Example: Custom Tracer Configuration
For advanced use cases with custom OpenTelemetry configuration:
```python theme={null}
import openlit
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
# Configure custom tracer provider
trace_provider = TracerProvider()
trace_provider.add_span_processor(
SimpleSpanProcessor(
OTLPSpanExporter(endpoint="http://127.0.0.1:4318/v1/traces")
)
)
trace.set_tracer_provider(trace_provider)
# Initialize OpenLIT with custom tracer
openlit.init(
tracer=trace.get_tracer(__name__),
disable_batch=True
)
# Create and configure the agent
agent = Agent(
model=OpenAIChat(id="gpt-4o-mini"),
tools=[DuckDuckGoTools()],
markdown=True,
)
# Use the agent
agent.print_response("What is currently trending on Twitter?")
```
## OpenLIT Dashboard Features
Once your agents are instrumented, you can access the OpenLIT dashboard to:
* **View Traces**: Visualize complete execution flows including agent runs, tool calls, and LLM requests
* **Monitor Performance**: Track latency, token usage, and throughput metrics
* **Analyze Costs**: Monitor API costs across different models and providers
* **Track Errors**: Identify and debug exceptions with detailed stack traces
* **Compare Models**: Evaluate different LLM providers based on performance and cost
## Configuration Options
The `openlit.init()` function accepts several parameters:
```python theme={null}
openlit.init(
otlp_endpoint="http://127.0.0.1:4318", # OTLP collector endpoint
tracer=None, # Custom OpenTelemetry tracer
disable_batch=False, # Disable batch span processing
environment="production", # Environment name for filtering
application_name="my-agent", # Application identifier
)
```
## CLI-Based Instrumentation
For true zero-code instrumentation, you can use the `openlit-instrument` CLI command to run your application without modifying any code:
```bash theme={null}
openlit-instrument \
--service-name my-ai-app \
--environment production \
--otlp-endpoint http://127.0.0.1:4318 \
python your_app.py
```
This approach is particularly useful for:
* Adding observability to existing applications without code changes
* CI/CD pipelines where you want to instrument automatically
* Testing observability before committing to code modifications
## Notes
* **Automatic Instrumentation**: OpenLIT automatically instruments supported LLM providers (OpenAI, Anthropic, etc.) and frameworks
* **Zero Code Changes**: Use either `openlit.init()` in your code or the `openlit-instrument` CLI to trace all LLM calls without modifications
* **OpenTelemetry Native**: OpenLIT uses standard OpenTelemetry protocols, ensuring compatibility with other observability tools
* **Open-Source & Self-Hosted**: OpenLIT is fully open-source and runs on your own infrastructure for complete data privacy and control
## Integration with Other Platforms
[OpenLIT](https://openlit.io/) can export traces to other observability platforms like Grafana Cloud, New Relic and more. See the [Langfuse integration guide](/integrations/observability/langfuse) for an example of using OpenLIT with Langfuse.
# OpenTelemetry
Source: https://docs.agno.com/integrations/observability/overview
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.
## 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/security) 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/security-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/security/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.[YOUR_DOMAIN]`
2. Creating records in Route 53.
3. Add the certificate ARN to Apps
**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