### Index
# Welcome to Atomic Agents Documentation
```{toctree}
:maxdepth: 2
:caption: Documentation
guides/index
api/index
examples/index
contributing
```
# A Lightweight and Modular Framework for Building AI Agents
```{admonition} AI Assistant Resources
:class: tip
š„ **Download Documentation for AI Assistants and LLMs**
Choose the resource that best fits your needs:
- **{download}`š Full Package <_static/llms-full.txt>`** - Complete documentation, source code, and examples in one file
- **{download}`š Documentation Only <_static/llms-docs.txt>`** - API documentation, guides, and references
- **{download}`š» Source Code Only <_static/llms-source.txt>`** - Complete atomic-agents framework source code
- **{download}`šÆ Examples Only <_static/llms-examples.txt>`** - All example implementations with READMEs
All files are optimized for AI assistants and Large Language Models, with clear structure and formatting for easy parsing.
```
The Atomic Agents framework is designed around the concept of atomicity to be an extremely lightweight and modular framework for building Agentic AI pipelines and applications without sacrificing developer experience and maintainability. The framework provides a set of tools and agents that can be combined to create powerful applications. It is built on top of [Instructor](https://github.com/jxnl/instructor) and leverages the power of [Pydantic](https://docs.pydantic.dev/latest/) for data and schema validation and serialization.
All logic and control flows are written in Python, enabling developers to apply familiar best practices and workflows from traditional software development without compromising flexibility or clarity.
## Key Features
- **Modularity**: Build AI applications by combining small, reusable components
- **Predictability**: Define clear input and output schemas using Pydantic
- **Extensibility**: Easily swap out components or integrate new ones
- **Control**: Fine-tune each part of the system individually
- **Provider Agnostic**: Works with various LLM providers through Instructor
- **Built for Production**: Robust error handling and async support
## Installation
You can install Atomic Agents using pip:
```bash
pip install atomic-agents
```
Or using uv (recommended):
```bash
uv add atomic-agents
```
Make sure you also install the provider you want to use. Provider SDKs are available as instructor extras:
```bash
pip install instructor[groq] # for Groq
pip install instructor[anthropic] # for Anthropic
pip install instructor[google-genai] # for Gemini
```
OpenAI is included by default.
This also installs the CLI *Atomic Assembler*, which can be used to download Tools (and soon also Agents and Pipelines).
```{note}
The framework supports multiple providers through Instructor, including **OpenAI**, **Anthropic**, **Groq**, **Ollama** (local models), **Gemini**, and more!
For a full list of all supported providers and their setup instructions, have a look at the [Instructor Integrations documentation](https://python.useinstructor.com/integrations/).
```
## Quick Example
Here's a glimpse of how easy it is to create an agent:
```python
import instructor
import openai
from atomic_agents.context import ChatHistory
from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema
# Set up your API key (either in environment or pass directly)
# os.environ["OPENAI_API_KEY"] = "your-api-key"
# or pass it to the client: openai.OpenAI(api_key="your-api-key")
# Initialize agent with history
history = ChatHistory()
# Set up client with your preferred provider
client = instructor.from_openai(openai.OpenAI()) # Pass your API key here if not in environment
# Create an agent
agent = AtomicAgent[BasicChatInputSchema, BasicChatOutputSchema](
config=AgentConfig(
client=client,
model="gpt-5-mini", # Use your provider's model
history=history
)
)
# Interact with your agent (using the agent's input schema)
response = agent.run(agent.input_schema(chat_message="Tell me about quantum computing"))
# Or more explicitly:
response = agent.run(
BasicChatInputSchema(chat_message="Tell me about quantum computing")
)
print(response)
```
## Example Projects
Check out our example projects in our [GitHub repository](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples):
- [Quickstart Examples](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/quickstart): Simple examples to get started
- [Hooks System](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/hooks-example): Comprehensive monitoring, error handling, and performance metrics
- [Basic Multimodal](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/basic-multimodal): Analyze images with text
- [RAG Chatbot](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/rag-chatbot): Build context-aware chatbots
- [Web Search Agent](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/web-search-agent): Create agents that perform web searches
- [Deep Research](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/deep-research): Perform deep research tasks
- [YouTube Summarizer](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/youtube-summarizer): Extract knowledge from videos
- [YouTube to Recipe](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/youtube-to-recipe): Convert cooking videos into structured recipes
- [Orchestration Agent](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/orchestration-agent): Coordinate multiple agents for complex tasks
## Community & Support
- [GitHub Repository](https://github.com/eigenwise/atomic-agents)
- [Issue Tracker](https://github.com/eigenwise/atomic-agents/issues)
- [Reddit Community](https://www.reddit.com/r/AtomicAgents/)
## Indices and References
* {ref}`genindex`
* {ref}`modindex`
* {ref}`search`
---
### Api/Agents
# Agents
## Schema Hierarchy
The Atomic Agents framework uses Pydantic for schema validation and serialization. All input and output schemas follow this inheritance pattern:
```PlainText
pydantic.BaseModel
āāā BaseIOSchema
āāā BasicChatInputSchema
āāā BasicChatOutputSchema
```
### BaseIOSchema
The base schema class that all agent input/output schemas inherit from.
```{eval-rst}
.. py:class:: BaseIOSchema
Base schema class for all agent input/output schemas. Inherits from :class:`pydantic.BaseModel`.
All agent schemas must inherit from this class to ensure proper serialization and validation.
**Inheritance:**
- :class:`pydantic.BaseModel`
```
### BasicChatInputSchema
The default input schema for agents.
```{eval-rst}
.. py:class:: BasicChatInputSchema
Default input schema for agent interactions.
**Inheritance:**
- :class:`BaseIOSchema` ā :class:`pydantic.BaseModel`
.. py:attribute:: chat_message
:type: str
The message to send to the agent.
Example:
>>> input_schema = BasicChatInputSchema(chat_message="Hello, agent!")
>>> agent.run(input_schema)
```
### BasicChatOutputSchema
The default output schema for agents.
```{eval-rst}
.. py:class:: BasicChatOutputSchema
Default output schema for agent responses.
**Inheritance:**
- :class:`BaseIOSchema` ā :class:`pydantic.BaseModel`
.. py:attribute:: chat_message
:type: str
The response message from the agent.
Example:
>>> response = agent.run(input_schema)
>>> print(response.chat_message)
```
### Creating Custom Schemas
You can create custom input/output schemas by inheriting from `BaseIOSchema`:
```python
from pydantic import Field
from typing import List
from atomic_agents import BaseIOSchema
class CustomInputSchema(BaseIOSchema):
chat_message: str = Field(..., description="User's message")
context: str = Field(None, description="Optional context for the agent")
class CustomOutputSchema(BaseIOSchema):
chat_message: str = Field(..., description="Agent's response")
follow_up_questions: List[str] = Field(
default_factory=list,
description="Suggested follow-up questions"
)
confidence: float = Field(
...,
description="Confidence score for the response",
ge=0.0,
le=1.0
)
```
## Base Agent
The `AtomicAgent` class is the foundation for building AI agents in the Atomic Agents framework. It handles chat interactions, history management, system prompts, and responses from language models.
```python
from atomic_agents import AtomicAgent, AgentConfig
from atomic_agents.context import ChatHistory, SystemPromptGenerator
# Create agent with basic configuration
agent = AtomicAgent[BasicChatInputSchema, BasicChatOutputSchema](
config=AgentConfig(
client=instructor.from_openai(OpenAI()),
model="gpt-4-turbo-preview",
history=ChatHistory(),
system_prompt_generator=SystemPromptGenerator()
)
)
# Run the agent
response = agent.run(user_input)
# Stream responses
async for partial_response in agent.run_async(user_input):
print(partial_response)
```
### Configuration
The `AgentConfig` class provides configuration options:
```python
class AgentConfig:
client: instructor.Instructor # Client for interacting with the language model
model: str = "gpt-4-turbo-preview" # Model to use
history: Optional[ChatHistory] = None # History component
system_prompt_generator: Optional[SystemPromptGenerator] = None # Prompt generator
input_schema: Optional[Type[BaseModel]] = None # Custom input schema
output_schema: Optional[Type[BaseModel]] = None # Custom output schema
model_api_parameters: Optional[dict] = None # Additional API parameters
```
### Input/Output Schemas
Default schemas for basic chat interactions:
```python
class BasicChatInputSchema(BaseIOSchema):
"""Input from the user to the AI agent."""
chat_message: str = Field(
...,
description="The chat message sent by the user."
)
class BasicChatOutputSchema(BaseIOSchema):
"""Response generated by the chat agent."""
chat_message: str = Field(
...,
description="The markdown-enabled response generated by the chat agent."
)
```
### Key Methods
- `run(user_input: Optional[BaseIOSchema] = None) -> BaseIOSchema`: Process user input and get response
- `run_async(user_input: Optional[BaseIOSchema] = None)`: Stream responses asynchronously
- `get_response(response_model=None) -> Type[BaseModel]`: Get direct model response
- `reset_history()`: Reset history to initial state
- `get_context_provider(provider_name: str)`: Get a registered context provider
- `register_context_provider(provider_name: str, provider: BaseDynamicContextProvider)`: Register a new context provider
- `unregister_context_provider(provider_name: str)`: Remove a context provider
- `get_context_token_count() -> TokenCountResult`: Get token count for current context (system prompt + history)
### Context Providers
Context providers can be used to inject dynamic information into the system prompt:
```python
from atomic_agents.context import BaseDynamicContextProvider
class SearchResultsProvider(BaseDynamicContextProvider):
def __init__(self, title: str):
super().__init__(title=title)
self.results = []
def get_info(self) -> str:
return "\n\n".join([
f"Result {idx}:\n{result}"
for idx, result in enumerate(self.results, 1)
])
# Register with agent
agent.register_context_provider(
"search_results",
SearchResultsProvider("Search Results")
)
```
### Streaming Support
The agent supports streaming responses for more interactive experiences:
```python
async def chat():
async for partial_response in agent.run_async(user_input):
# Handle each chunk of the response
print(partial_response.chat_message)
```
### History Management
The agent automatically manages conversation history through the `ChatHistory` component:
```python
# Access history
history = agent.history.get_history()
# Reset to initial state
agent.reset_history()
# Save/load history state
serialized = agent.history.dump()
agent.history.load(serialized)
```
### Token Counting
Monitor context usage with the `get_context_token_count()` method. Token counts are computed accurately on-demand by serializing the context exactly as Instructor does, including the output schema overhead. This works with any provider (OpenAI, Anthropic, Google, etc.) and supports multimodal content:
```python
# Get accurate token count at any time - always returns a result
token_info = agent.get_context_token_count()
print(f"Total tokens: {token_info.total}")
print(f"System prompt (with schema): {token_info.system_prompt} tokens")
print(f"History: {token_info.history} tokens")
print(f"Model: {token_info.model}")
# Check context utilization if max tokens is known
if token_info.max_tokens:
print(f"Max context: {token_info.max_tokens} tokens")
if token_info.utilization:
print(f"Context utilization: {token_info.utilization:.1%}")
```
The `TokenCountResult` contains:
- `total`: Total tokens in context (system + history + schema overhead)
- `system_prompt`: Tokens used by system prompt and output schema
- `history`: Tokens used by conversation history (including multimodal content)
- `model`: The model name used for counting
- `max_tokens`: Maximum context window (if known)
- `utilization`: Percentage of context used (if max_tokens known)
### Custom Schemas
You can use custom input/output schemas for structured interactions:
```python
from pydantic import BaseModel, Field
from typing import List
class CustomInput(BaseIOSchema):
"""Custom input with specific fields"""
question: str = Field(..., description="User's question")
context: str = Field(..., description="Additional context")
class CustomOutput(BaseIOSchema):
"""Custom output with structured data"""
answer: str = Field(..., description="Answer to the question")
sources: List[str] = Field(..., description="Source references")
# Create agent with custom schemas
agent = AtomicAgent[CustomInput, CustomOutput](
config=AgentConfig(
client=client,
model=model,
)
)
```
For full API details:
```{eval-rst}
.. automodule:: atomic_agents.agents.atomic_agent
:members:
:undoc-members:
:show-inheritance:
```
---
### Api/Context
# Context
```{seealso}
For a comprehensive guide on memory management, multi-agent patterns, and best practices, see the **[Memory and Context Guide](/guides/memory)**.
```
## Agent History
The `ChatHistory` class manages conversation history and state for AI agents. It implements `BaseChatHistory`, an interface-only abstract base class that declares the memory contract `AtomicAgent` depends on (the type `AgentConfig.history` accepts). Implement `BaseChatHistory` directly, or subclass `ChatHistory`, to plug in your own persistent or backend-specific memory store. See the [Memory guide's "Writing a Custom Memory Backend" section](/guides/memory#writing-a-custom-memory-backend) for the full contract and a recommended pattern.
```python
from atomic_agents.context import ChatHistory
from atomic_agents import BaseIOSchema
# Initialize history with optional max messages
history = ChatHistory(max_messages=10)
# Add messages
history.add_message(
role="user",
content=BaseIOSchema(...)
)
# Initialize a new turn
history.initialize_turn()
turn_id = history.get_current_turn_id()
# Access history
history = history.get_history()
# Manage history
history.get_message_count() # Get number of messages
history.delete_turn_id(turn_id) # Delete messages by turn
# Persistence
serialized = history.dump() # Save to string
history.load(serialized) # Load from string
# Create copy
new_history = history.copy()
```
Key features:
- Message history management with role-based messages
- Turn-based conversation tracking
- Support for multimodal content (images, etc.)
- Serialization and persistence
- History size management
- Deep copy functionality
### Message Structure
Messages in history are structured as:
```python
class Message(BaseModel):
role: str # e.g., 'user', 'assistant', 'system'
content: BaseIOSchema # Message content following schema
turn_id: Optional[str] # Unique ID for grouping messages
```
### Multimodal Support
The history system automatically handles multimodal content:
```python
# For content with images
history = history.get_history()
for message in history:
if isinstance(message.content, list):
text_content = message.content[0] # JSON string
images = message.content[1:] # List of images
```
## System Prompt Generator
The `SystemPromptGenerator` creates structured system prompts for AI agents:
```python
from atomic_agents.context import (
SystemPromptGenerator,
BaseDynamicContextProvider
)
# Create generator with static content
generator = SystemPromptGenerator(
background=[
"You are a helpful AI assistant.",
"You specialize in technical support."
],
steps=[
"1. Understand the user's request",
"2. Analyze available information",
"3. Provide clear solutions"
],
output_instructions=[
"Use clear, concise language",
"Include step-by-step instructions",
"Cite relevant documentation"
]
)
# Generate prompt
prompt = generator.generate_prompt()
```
### Custom System Prompt Generator
If you require finer control over system prompt construction, subclass `BaseSystemPromptGenerator` and implement `generate_prompt()`. This approach is useful when prompt content should be maintained in a human-readable format (e.g., Markdown or text file) to allow review or editing by non-developers.
```python
from pathlib import Path
from typing import Dict, Optional, Union
from atomic_agents.context import (
BaseDynamicContextProvider,
BaseSystemPromptGenerator
)
class MarkdownFileSystemPromptGenerator(BaseSystemPromptGenerator):
def __init__(
self,
md_file: Union[Path, str],
context_providers: Optional[Dict[str, BaseDynamicContextProvider]] = None,
):
super().__init__(context_providers=context_providers)
path = Path(md_file)
if not path.exists():
raise FileNotFoundError(f"System prompt file not found: {md_file}")
self.system_prompt = path.read_text(encoding="utf-8")
def generate_prompt(self) -> str:
return f"{self.system_prompt}\n\n{self._build_context_string()}"
def _build_context_string(self) -> str:
if not self.context_providers:
return ""
context_sections = ["# Additional Context"]
for provider in self.context_providers.values():
info = provider.get_info()
if info:
context_sections.append(f"## {provider.title}")
context_sections.append(info)
context_sections.append("")
return "\n".join(context_sections).strip()
generator = MarkdownFileSystemPromptGenerator("path/to/system_prompt.md")
prompt = generator.generate_prompt()
```
### Dynamic Context Providers
Context providers inject dynamic information into prompts:
```python
from dataclasses import dataclass
from typing import List
@dataclass
class SearchResult:
content: str
metadata: dict
class SearchResultsProvider(BaseDynamicContextProvider):
def __init__(self, title: str):
super().__init__(title=title)
self.results: List[SearchResult] = []
def get_info(self) -> str:
"""Format search results for the prompt"""
if not self.results:
return "No search results available."
return "\n\n".join([
f"Result {idx}:\nMetadata: {result.metadata}\nContent:\n{result.content}\n{'-' * 80}"
for idx, result in enumerate(self.results, 1)
])
# Use with generator
generator = SystemPromptGenerator(
background=["You answer based on search results."],
context_providers={
"search_results": SearchResultsProvider("Search Results")
}
)
```
The generated prompt will include:
1. Background information
2. Processing steps (if provided)
3. Dynamic context from providers
4. Output instructions
## Base Components
### BaseIOSchema
Base class for all input/output schemas:
```python
from atomic_agents import BaseIOSchema
from pydantic import Field
class CustomSchema(BaseIOSchema):
"""Schema description (required)"""
field: str = Field(..., description="Field description")
```
Key features:
- Requires docstring description
- Rich representation support
- Automatic schema validation
- JSON serialization
### BaseTool
Base class for creating tools:
```python
from atomic_agents import BaseTool, BaseToolConfig
from pydantic import Field
class MyToolConfig(BaseToolConfig):
"""Tool configuration"""
api_key: str = Field(
default=os.getenv("API_KEY"),
description="API key for the service"
)
class MyTool(BaseTool[MyToolInputSchema, MyToolOutputSchema]):
"""Tool implementation"""
input_schema = MyToolInputSchema
output_schema = MyToolOutputSchema
def __init__(self, config: MyToolConfig = MyToolConfig()):
super().__init__(config)
self.api_key = config.api_key
def run(self, params: MyToolInputSchema) -> MyToolOutputSchema:
# Implement tool logic
pass
```
Key features:
- Structured input/output schemas
- Configuration management
- Title and description overrides
- Error handling
For full API details:
```{eval-rst}
.. automodule:: atomic_agents.context.chat_history
:members:
:undoc-members:
:show-inheritance:
.. automodule:: atomic_agents.context.base_chat_history
:members:
:undoc-members:
:show-inheritance:
.. automodule:: atomic_agents.context.system_prompt_generator
:members:
:undoc-members:
:show-inheritance:
.. automodule:: atomic_agents.base.base_io_schema
:members:
:undoc-members:
:show-inheritance:
.. automodule:: atomic_agents.base.base_tool
:members:
:undoc-members:
:show-inheritance:
```
---
### Api/Index
# API Reference
This section contains the API reference for all public modules and classes in Atomic Agents.
```{toctree}
:maxdepth: 2
:caption: API Reference
agents
context
utils
```
## Core Components
The Atomic Agents framework is built around several core components that work together to provide a flexible and powerful system for building AI agents.
### Agents
The agents module provides the base classes for creating AI agents:
- `AtomicAgent`: The foundational agent class that handles interactions with LLMs
- `AgentConfig`: Configuration class for customizing agent behavior
- `BasicChatInputSchema`: Standard input schema for agent interactions
- `BasicChatOutputSchema`: Standard output schema for agent responses
[Learn more about agents](agents.md)
### Context Components
The context module contains essential building blocks:
- `ChatHistory`: Manages conversation history and state with support for:
- Message history with role-based messages
- Turn-based conversation tracking
- Multimodal content
- Serialization and persistence
- History size management
- `SystemPromptGenerator`: Creates structured system prompts with:
- Background information
- Processing steps
- Output instructions
- Dynamic context through context providers
- `BaseDynamicContextProvider`: Base class for creating custom context providers that can inject dynamic information into system prompts
[Learn more about context components](context.md)
### Utils
The utils module provides helper functions and utilities:
- Message formatting
- Tool response handling
- Schema validation
- Error handling
[Learn more about utilities](utils.md)
## Getting Started
For practical examples and guides on using these components, see:
- [Quickstart Guide](../guides/quickstart.md)
- [Tools Guide](../guides/tools.md)
---
### Api/Utils
# Utilities
## Token Counting
The `TokenCounter` utility provides provider-agnostic token counting for any model supported by LiteLLM. This allows you to monitor context usage regardless of whether you're using OpenAI, Anthropic, Google, or any other supported provider.
### TokenCountResult
A named tuple containing token count information:
```{eval-rst}
.. py:class:: TokenCountResult
Named tuple containing token count information.
.. py:attribute:: total
:type: int
Total tokens in the context (system prompt + history + schema overhead).
.. py:attribute:: system_prompt
:type: int
Tokens used by the system prompt and output schema.
.. py:attribute:: history
:type: int
Tokens used by conversation history (including multimodal content).
.. py:attribute:: model
:type: str
The model used for token counting.
.. py:attribute:: max_tokens
:type: Optional[int]
Maximum context window for the model (if known).
.. py:attribute:: utilization
:type: Optional[float]
Context utilization percentage (0.0 to 1.0) if max_tokens is known.
```
### TokenCounter
The main utility class for counting tokens:
```{eval-rst}
.. py:class:: TokenCounter
Utility class for counting tokens in messages using LiteLLM.
.. py:method:: count_messages(model: str, messages: List[Dict[str, Any]]) -> int
Count tokens in a list of messages.
:param model: The model name (e.g., "gpt-4", "claude-3-opus-20240229")
:param messages: List of message dictionaries with "role" and "content" keys
:return: Number of tokens
.. py:method:: count_text(model: str, text: str) -> int
Count tokens in a text string.
:param model: The model name
:param text: The text to count tokens for
:return: Number of tokens
.. py:method:: get_max_tokens(model: str) -> Optional[int]
Get the maximum context window for a model.
:param model: The model name
:return: Maximum tokens, or None if unknown
.. py:method:: count_context(model: str, system_messages: List[Dict], history_messages: List[Dict]) -> TokenCountResult
Count tokens for a complete context (system prompt + history).
:param model: The model name
:param system_messages: System prompt messages
:param history_messages: Conversation history messages
:return: TokenCountResult with detailed breakdown
```
### Usage Example
```python
from atomic_agents.utils import TokenCounter, TokenCountResult
# Direct usage
counter = TokenCounter()
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello!"},
{"role": "assistant", "content": "Hi there! How can I help?"},
]
# Count tokens in messages
token_count = counter.count_messages("gpt-4", messages)
# Get max context window
max_tokens = counter.get_max_tokens("gpt-4")
# Count complete context with breakdown
result = counter.count_context(
model="gpt-4",
system_messages=[{"role": "system", "content": "You are helpful."}],
history_messages=[{"role": "user", "content": "Hello!"}],
)
print(f"Total: {result.total}, System: {result.system_prompt}, History: {result.history}")
if result.utilization:
print(f"Context utilization: {result.utilization:.1%}")
```
### Using with AtomicAgent
The easiest way to get token counts is through the agent's `get_context_token_count()` method. The agent computes accurate token counts on-demand by serializing the context exactly as Instructor does, including output schema overhead and multimodal content:
```python
# Get accurate token count at any time - always returns a result
token_info = agent.get_context_token_count()
print(f"Total tokens: {token_info.total}")
print(f"System prompt (with schema): {token_info.system_prompt} tokens")
print(f"History: {token_info.history} tokens")
if token_info.utilization:
print(f"Context utilization: {token_info.utilization:.1%}")
```
The token count includes:
- System prompt content
- Output schema overhead (the JSON schema Instructor sends for structured output)
- Conversation history (including multimodal content like images, PDFs, audio)
This gives you an accurate count that matches what would be sent to the API.
## Tool Message Formatting
```{eval-rst}
.. automodule:: atomic_agents.utils.format_tool_message
:members:
:undoc-members:
:show-inheritance:
```
---
### Examples/Index
# Example Projects
This section contains detailed examples of using Atomic Agents in various scenarios.
```{note}
All examples are available in optimized formats for AI assistants:
- **{download}`Examples with documentation <../_static/llms-examples.txt>`** - All examples with source code and READMEs
- **{download}`Full framework package <../_static/llms-full.txt>`** - Complete documentation, source, and examples
```
## Quickstart Examples
Simple examples to get started with the framework:
- Basic chatbot with history
- Custom chatbot with personality
- Streaming responses
- Custom input/output schemas
- Multiple provider support
š **[View on GitHub](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/quickstart)** - Browse the complete source code and run the examples
## Hooks System
Comprehensive monitoring and error handling with the AtomicAgent hook system:
- Parse error handling and validation
- API call monitoring and metrics
- Response time tracking and performance analysis
- Intelligent retry mechanisms
- Production-ready error isolation
- Real-time performance dashboards
š **[View on GitHub](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/hooks-example)** - Browse the complete source code and run the examples
## Basic Multimodal
Examples of working with images and text:
- Image analysis with text descriptions
- Image-based question answering
- Visual content generation
- Multi-image comparisons
š **[View on GitHub](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/basic-multimodal)** - Browse the complete source code and run the examples
## RAG Chatbot
Build context-aware chatbots with retrieval-augmented generation:
- Document indexing and embedding
- Semantic search integration
- Context-aware responses
- Source attribution
- Follow-up suggestions
š **[View on GitHub](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/rag-chatbot)** - Browse the complete source code and run the examples
## Web Search Agent
Create agents that can search and analyze web content:
- Web search integration
- Content extraction
- Result synthesis
- Multi-source research
- Citation tracking
š **[View on GitHub](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/web-search-agent)** - Browse the complete source code and run the examples
## Deep Research
Perform comprehensive research tasks:
- Multi-step research workflows
- Information synthesis
- Source validation
- Structured output generation
- Citation management
š **[View on GitHub](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/deep-research)** - Browse the complete source code and run the examples
## YouTube Summarizer
Extract and analyze information from videos:
- Transcript extraction
- Content summarization
- Key point identification
- Timestamp linking
- Chapter generation
š **[View on GitHub](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/youtube-summarizer)** - Browse the complete source code and run the examples
## YouTube to Recipe
Convert cooking videos into structured recipes:
- Video analysis
- Recipe extraction
- Ingredient parsing
- Step-by-step instructions
- Time and temperature conversion
š **[View on GitHub](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/youtube-to-recipe)** - Browse the complete source code and run the examples
## Orchestration Agent
Coordinate multiple agents for complex tasks:
- Agent coordination
- Task decomposition
- Progress tracking
- Error handling
- Result aggregation
š **[View on GitHub](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/orchestration-agent)** - Browse the complete source code and run the examples
## MCP Agent
Build intelligent agents using the Model Context Protocol:
- Server implementation with multiple transport methods
- Dynamic tool discovery and registration
- Natural language query processing
- Stateful conversation handling
- Extensible tool architecture
[View MCP Agent Documentation](mcp_agent.md)
š **[View on GitHub](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/mcp-agent)** - Browse the complete source code and run the examples
---
### Examples/Mcp Agent
# MCP Server and Client Example
This guide provides a detailed overview of the Model Context Protocol (MCP) server and client example implementation in the Atomic Agents framework.
## Overview
The MCP example demonstrates how to build an intelligent agent system using the Model Context Protocol, showcasing both server and client implementations. The example supports three transport methods: STDIO, Server-Sent Events (SSE), and HTTP Stream.
## Architecture
### MCP Server
The server component (`example-mcp-server`) is built using:
- FastMCP: A high-performance MCP server implementation
- Starlette: A lightweight ASGI framework
- Uvicorn: An ASGI server implementation
Key components:
1. **Transport Layers**:
- `server_stdio.py`: Implements STDIO-based communication
- `server_sse.py`: Implements SSE-based HTTP communication
- `server_http.py`: Implements unified HTTP streaming transport
2. **Tools Service**:
- Manages registration and execution of MCP tools
- Handles tool discovery and metadata
3. **Resource Service**:
- Manages static resources
- Handles resource discovery and access
4. **Built-in Tools**:
- `AddNumbersTool`: Performs addition
- `SubtractNumbersTool`: Performs subtraction
- `MultiplyNumbersTool`: Performs multiplication
- `DivideNumbersTool`: Performs division
### MCP Client
The client component (`example-client`) is an intelligent agent that:
1. **Tool Discovery**:
- Dynamically discovers available tools from the MCP server
- Builds a schema-based tool registry
2. **Query Processing**:
- Uses GPT models for natural language understanding
- Extracts parameters from user queries
- Selects appropriate tools based on intent
3. **Execution Flow**:
- Maintains conversation context
- Handles tool execution results
- Provides conversational responses
## Implementation Details
### Server Implementation
The server supports three transport methods:
1. **SSE Transport** (`server_sse.py`):
```python
# Initialize FastMCP server
mcp = FastMCP("example-mcp-server")
# Register tools and resources
tool_service.register_tools(get_available_tools())
resource_service.register_resources(get_available_resources())
# Create Starlette app with CORS support
app = create_starlette_app(mcp_server)
```
2. **STDIO Transport** (`server_stdio.py`):
- Runs as a subprocess
- Communicates through standard input/output
- Ideal for local development
3. **HTTP Stream Transport** (`server_http.py`):
- Single `/mcp` endpoint for JSON-RPC and SSE-style streaming
- Handles session via `Mcp-Session-Id` header; allows resumable and cancelable streams
### Client Implementation
The client uses a sophisticated orchestration system:
1. **Tool Management**:
```python
# Fetch available tools (synchronous)
tools = fetch_mcp_tools(
mcp_endpoint=config.mcp_server_url,
transport_type=MCPTransportType.HTTP_STREAM,
)
# Or fetch tools asynchronously (must be called within async context)
tools = await fetch_mcp_tools_async(
mcp_endpoint=config.mcp_server_command,
transport_type=MCPTransportType.STDIO,
client_session=session, # Optional pre-initialized session
)
# Build tool schema mapping
tool_schema_to_class_map = {
ToolClass.input_schema: ToolClass
for ToolClass in tools
if hasattr(ToolClass, "input_schema")
}
```
2. **Query Processing**:
- Uses an orchestrator agent to analyze queries
- Extracts parameters and selects appropriate tools
- Maintains conversation context through ChatHistory
3. **Async vs Sync Tool Fetching**:
```python
# Synchronous fetching (suitable for HTTP/SSE)
tools = fetch_mcp_tools(
mcp_endpoint="http://localhost:6969",
transport_type=MCPTransportType.HTTP_STREAM
)
# Asynchronous fetching (optimized for STDIO with persistent sessions)
async def setup_tools():
tools = await fetch_mcp_tools_async(
transport_type=MCPTransportType.STDIO,
client_session=session # Reuse existing session
)
return tools
```
## MCP Transport Methods
The example implements three distinct transport methods via the `MCPTransportType` enum, each with its own advantages:
```python
from atomic_agents.connectors.mcp.mcp_definition_service import MCPTransportType
# Available transport types
MCPTransportType.STDIO # Standard input/output transport
MCPTransportType.SSE # Server-Sent Events transport
MCPTransportType.HTTP_STREAM # HTTP Stream transport
```
### 1. STDIO Transport
STDIO transport uses standard input/output streams for communication between the client and server:
```python
# Client-side STDIO setup (from main_stdio.py)
async def _bootstrap_stdio():
stdio_exit_stack = AsyncExitStack()
command_parts = shlex.split(config.mcp_stdio_server_command)
server_params = StdioServerParameters(command=command_parts[0], args=command_parts[1:], env=None)
read_stream, write_stream = await stdio_exit_stack.enter_async_context(stdio_client(server_params))
session = await stdio_exit_stack.enter_async_context(ClientSession(read_stream, write_stream))
await session.initialize()
return session
```
**Key advantages**:
- No network configuration required
- Simple local setup
- Direct process communication
- Lower latency for local usage
**Use cases**:
- Development and testing
- Single-user environments
- Embedded agent applications
- Offline operation
### 2. SSE Transport
Server-Sent Events (SSE) transport uses HTTP long-polling for real-time, one-way communication:
```python
# Server-side SSE setup (from server_sse.py)
async def handle_sse(request: Request) -> None:
async with sse.connect_sse(
request.scope,
request.receive,
request._send, # noqa: SLF001
) as (read_stream, write_stream):
await mcp_server.run(
read_stream,
write_stream,
mcp_server.create_initialization_options(),
)
```
**Key advantages**:
- Multiple clients can connect to a single server
- Network-based communication
- Stateless server architecture
- Suitable for distributed systems
**Use cases**:
- Production deployments
- Multi-user environments
- Scalable agent infrastructure
- Cross-network operation
### 3. HTTP Stream Transport
HTTP Stream transport uses a single `/mcp` endpoint for JSON-RPC and streaming communication:
```python
# Client-side HTTP Stream setup
tools = fetch_mcp_tools(
mcp_endpoint="http://localhost:6969",
transport_type=MCPTransportType.HTTP_STREAM,
)
```
**Key advantages**:
- Single endpoint for all MCP operations
- Session management via headers
- Resumable and cancelable streams
- Modern HTTP-based architecture
**Use cases**:
- Modern web applications
- Cloud-native deployments
- Microservice architectures
- API gateway integration
## Interfaces
### Tool Interface
The MCP server defines a standardized tool interface that all tools must implement:
```python
class Tool(ABC):
"""Abstract base class for all tools."""
name: ClassVar[str]
description: ClassVar[str]
input_model: ClassVar[Type[BaseToolInput]]
output_model: ClassVar[Optional[Type[BaseModel]]] = None
@abstractmethod
async def execute(self, input_data: BaseToolInput) -> ToolResponse:
"""Execute the tool with given arguments."""
pass
def get_schema(self) -> Dict[str, Any]:
"""Get JSON schema for the tool."""
schema = {
"name": self.name,
"description": self.description,
"input": self.input_model.model_json_schema(),
}
if self.output_model:
schema["output"] = self.output_model.model_json_schema()
return schema
```
The tool interface consists of:
1. **Class Variables**:
- `name`: Tool identifier used in MCP communications
- `description`: Human-readable tool description
- `input_model`: Pydantic model defining input parameters
- `output_model`: Pydantic model defining output structure (optional)
2. **Execute Method**:
- Asynchronous method that performs the tool's functionality
- Takes strongly-typed input data
- Returns a structured ToolResponse
3. **Schema Method**:
- Provides JSON Schema for tool discovery
- Enables automatic documentation generation
- Facilitates client-side validation
### Resource Interface
The MCP server defines a standardized resource interface that all resources must implement:
```python
class Resource(ABC):
"""Abstract base class for all resources."""
name: ClassVar[str]
description: ClassVar[str]
uri: ClassVar[str]
mime_type: ClassVar[str]
input_model: ClassVar[Type[BaseResourceInput]]
output_model: ClassVar[Optional[Type[BaseModel]]] = None
@abstractmethod
async def read(self, input_data: BaseResourceInput) -> ResourceResponse:
"""Read data from the resource."""
pass
def get_schema(self) -> Dict[str, Any]:
"""Get JSON schema for the resource."""
schema = {
"name": self.name,
"description": self.description,
"uri": self.uri,
"mime_type": self.mime_type,
"input": self.input_model.model_json_schema(),
}
if self.output_model:
schema["output"] = self.output_model.model_json_schema()
return schema
```
The resource interface consists of:
1. **Class Variables**:
- `name`: Resource identifier used in MCP communications
- `description`: Human-readable resource description
- `uri`: URI pattern for accessing the resource
- `mime_type`: MIME type of the resource content
- `input_model`: Pydantic model defining input parameters
- `output_model`: Pydantic model defining output structure (optional)
2. **Read Method**:
- Asynchronous method that retrieves data from the resource
- Takes strongly-typed input data
- Returns a structured ResourceResponse
3. **Schema Method**:
- Provides URI Template for resource discovery
- Enables automatic documentation generation
### Prompt Interface
The MCP client uses a standardized prompt interface for managing prompts:
```python
class Prompt(ABC):
"""Abstract base class for all prompts."""
name: ClassVar[str]
description: ClassVar[str]
input_model: ClassVar[Type[BasePromptInput]]
output_model: ClassVar[Optional[Type[BaseModel]]] = None
@abstractmethod
async def generate(self, input_data: BasePromptInput) -> PromptResponse:
"""Generate the prompt with given arguments."""
pass
def get_schema(self) -> Dict[str, Any]:
"""Get JSON schema for the prompt."""
schema = {
"name": self.name,
"description": self.description,
"input": self.input_model.model_json_schema(),
}
if self.output_model:
schema["output"] = self.output_model.model_json_schema()
return schema
```
The prompt interface consists of:
1. **Class Variables**:
- `name`: Prompt identifier used in MCP communications
- `description`: Human-readable prompt description
- `input_model`: Pydantic model defining input parameters
- `output_model`: Pydantic model defining output structure (optional)
2. **Generate Method**:
- Asynchronous method that generates the prompt
- Takes strongly-typed input data
- Returns a structured PromptResponse
3. **Schema Method**:
- Provides JSON Schema for prompt discovery
- Enables automatic documentation generation
## Configuration
### Server Configuration
The server can be configured through command-line arguments:
```bash
uv run example-mcp-server --mode=sse --host=0.0.0.0 --port=6969 --reload
uv run example-mcp-server --mode=stdio
uv run example-mcp-server --mode=http_stream --host=0.0.0.0 --port=6969
```
Options:
- `--mode`: Transport mode (sse/stdio/http_stream)
- `--host`: Host to bind to
- `--port`: Port to listen on
- `--reload`: Enable auto-reload for development
### Client Configuration
The client uses a configuration class:
```python
@dataclass
class MCPConfig:
mcp_server_url: str = "http://localhost:6969"
openai_model: str = "gpt-5-mini"
openai_api_key: str = os.getenv("OPENAI_API_KEY")
```
For STDIO transport, additional options are available:
```python
@dataclass
class MCPConfig:
openai_model: str = "gpt-5-mini"
openai_api_key: str = os.getenv("OPENAI_API_KEY")
mcp_stdio_server_command: str = "uv run example-mcp-server --mode stdio"
```
## Usage Examples
1. **Start the Server (SSE mode)**:
```bash
cd example-mcp-server
uv run example-mcp-server --mode=sse
```
2. **Run the Client**:
Using the main launcher with transport selection:
```bash
cd example-client
uv run python -m example_client.main --transport sse
```
Directly calling the SSE client:
```bash
cd example-client
uv run python -m example_client.main_sse
```
Directly calling the STDIO client:
```bash
cd example-client
uv run python -m example_client.main_stdio
```
3. **Example Queries**:
```
You: What is 2+2?
You: Calculate the square root of 144
You: Generate a random number between 1 and 100
```
## Best Practices
1. **Development**:
- Use STDIO transport for local development
- Enable server auto-reload during development
- Implement proper error handling
2. **Production**:
- Use SSE transport for production deployments
- Configure appropriate CORS settings
- Implement authentication if needed
3. **Tool Development**:
- Follow the Tool interface contract
- Provide clear input/output schemas
- Include comprehensive documentation
## Extending the Example
### Adding New Tools
To add new tools:
1. Create a new tool class implementing the Tool interface
2. Register the tool in the server's tool service
3. The client will automatically discover and use the new tool
Example tool structure:
```python
class MyNewTool(Tool):
name = "my_new_tool"
description = "This tool performs a custom operation"
input_model = create_model(
"MyNewToolInput",
param1=(str, Field(..., description="First parameter")),
param2=(int, Field(..., description="Second parameter")),
__base__=BaseToolInput
)
async def execute(self, input_data: BaseToolInput) -> ToolResponse:
# Access params with input_data.param1, input_data.param2
result = f"Processed {input_data.param1} with {input_data.param2}"
return ToolResponse.from_text(result)
```
Then register the tool in the server:
```python
def get_available_tools() -> List[Tool]:
return [
# ... existing tools ...
MyNewTool(),
]
```
### Adding New Resources
To add new resources:
1. Create a new resource class implementing the Resource interface
2. Register the resource in the server's resource service
3. The client can access the new resource via its URI
Example resource structure:
```python
class MyNewResource(Resource):
name = "my_new_resource"
description = "This resource provides custom data"
uri = "resource://my_new_resource/{param1}"
mime_type = "application/json"
input_model = create_model(
"MyNewResourceInput",
param1=(str, Field(..., description="Resource parameter")),
__base__=BaseResourceInput
)
async def read(self, input_data: BaseResourceInput) -> ResourceResponse:
# Access params with input_data.param1
data = {"message": f"Data for {input_data.param1}"}
return ResourceResponse.from_data(data, self.mime_type)
```
Then register the resource in the server:
```python
def get_available_resources() -> List[Resource]:
return [
# ... existing resources ...
MyNewResource(),
]
```
### Adding New Prompts
To add new prompts:
1. Create a new prompt class implementing the Prompt interface
2. Register the prompt in the server's prompt service
3. The client will automatically discover and use the new prompt
Example prompt structure:
```python
class MyNewPrompt(Prompt):
name = "my_new_prompt"
description = "This prompt generates a custom response"
input_model = create_model(
"MyNewPromptInput",
param1=(str, Field(..., description="First parameter")),
param2=(int, Field(..., description="Second parameter")),
__base__=BasePromptInput
)
async def generate(self, input_data: BasePromptInput) -> PromptResponse:
# Access params with input_data.param1, input_data.param2
result = f"Generated response for {input_data.param1} with {input_data.param2}"
return PromptResponse.from_text(result)
```
Then register the prompt in the server:
```python
def get_available_prompts() -> List[Prompt]:
return [
# ... existing prompts ...
MyNewPrompt(),
]
```
---
### Guides/Cookbook
# Cookbook
Practical recipes for common Atomic Agents use cases.
## Quick Reference
| Recipe | Description |
|--------|-------------|
| [Basic Chatbot](#basic-chatbot) | Simple conversational agent |
| [Chatbot with Memory](#chatbot-with-memory) | Agent that remembers context |
| [Custom Output Schema](#custom-output-schema) | Structured responses |
| [Multi-Provider Agent](#multi-provider-agent) | Switch between LLM providers |
| [Agent with Tools](#agent-with-tools) | Agent using external tools |
| [Streaming Chatbot](#streaming-chatbot) | Real-time response streaming |
| [Research Agent](#research-agent) | Multi-step research workflow |
| [RAG Agent](#rag-agent) | Retrieval-augmented generation |
## Basic Chatbot
A minimal chatbot implementation.
```python
"""
Basic Chatbot Recipe
A simple conversational agent that responds to user messages.
Requirements:
- pip install atomic-agents openai
- Set OPENAI_API_KEY environment variable
"""
import os
import instructor
import openai
from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema
from atomic_agents.context import ChatHistory
def create_basic_chatbot():
"""Create a basic chatbot agent."""
client = instructor.from_openai(openai.OpenAI())
agent = AtomicAgent[BasicChatInputSchema, BasicChatOutputSchema](
config=AgentConfig(
client=client,
model="gpt-5-mini",
history=ChatHistory()
)
)
return agent
def chat_loop(agent):
"""Interactive chat loop."""
print("Chatbot ready! Type 'quit' to exit.\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() in ['quit', 'exit', 'q']:
print("Goodbye!")
break
if not user_input:
continue
response = agent.run(BasicChatInputSchema(chat_message=user_input))
print(f"Bot: {response.chat_message}\n")
if __name__ == "__main__":
agent = create_basic_chatbot()
chat_loop(agent)
```
## Chatbot with Memory
Agent that maintains conversation history across turns.
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Custom Output Schema
Agent with structured output including metadata.
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Multi-Provider Agent
Switch between different LLM providers dynamically.
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Agent with Tools
Agent that uses tools to extend capabilities.
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Streaming Chatbot
Real-time streaming responses.
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Research Agent
Multi-step research workflow.
```
/* Detailed source-code truncated for AI context efficiency. */
```
## RAG Agent
Retrieval-augmented generation pattern.
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Summary
These recipes demonstrate common patterns:
| Pattern | Key Components | Use Case |
|---------|---------------|----------|
| Basic Chatbot | AtomicAgent, ChatHistory | Simple Q&A |
| Memory | ChatHistory persistence | Context retention |
| Custom Schema | BaseIOSchema subclass | Structured output |
| Multi-Provider | Provider switching | Flexibility |
| Tools | BaseTool | Extended capabilities |
| Streaming | run_async_stream | Real-time UX |
| Research | Multiple agents | Complex workflows |
| RAG | Context providers | Knowledge-augmented |
Combine these patterns to build sophisticated AI applications.
---
### Guides/Deployment
# Deployment Guide
This guide covers best practices for deploying Atomic Agents applications to production environments.
## Overview
Deploying AI agents requires attention to:
- **Configuration Management**: Environment-specific settings
- **API Key Security**: Secure credential handling
- **Scaling**: Handling concurrent requests
- **Monitoring**: Observability and alerting
- **Error Handling**: Graceful degradation
## Environment Configuration
### Using Environment Variables
Store configuration in environment variables:
```python
import os
from dataclasses import dataclass
from typing import Optional
@dataclass
class AgentDeploymentConfig:
"""Production configuration for agents."""
# Required
openai_api_key: str
model: str
# Optional with defaults
max_tokens: int = 2048
temperature: float = 0.7
timeout: float = 30.0
max_retries: int = 3
@classmethod
def from_env(cls) -> "AgentDeploymentConfig":
"""Load configuration from environment variables."""
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise ValueError("OPENAI_API_KEY environment variable is required")
return cls(
openai_api_key=api_key,
model=os.getenv("AGENT_MODEL", "gpt-4o-mini"),
max_tokens=int(os.getenv("AGENT_MAX_TOKENS", "2048")),
temperature=float(os.getenv("AGENT_TEMPERATURE", "0.7")),
timeout=float(os.getenv("AGENT_TIMEOUT", "30.0")),
max_retries=int(os.getenv("AGENT_MAX_RETRIES", "3")),
)
# Usage
config = AgentDeploymentConfig.from_env()
```
### Configuration File Pattern
For complex deployments, use configuration files:
```python
import os
import json
from pathlib import Path
def load_config(env: str = None) -> dict:
"""Load environment-specific configuration."""
env = env or os.getenv("DEPLOYMENT_ENV", "development")
config_path = Path(f"config/{env}.json")
if not config_path.exists():
raise FileNotFoundError(f"Config not found: {config_path}")
with open(config_path) as f:
config = json.load(f)
# Override with environment variables
if os.getenv("OPENAI_API_KEY"):
config["openai_api_key"] = os.getenv("OPENAI_API_KEY")
return config
# config/production.json example:
# {
# "model": "gpt-4o",
# "max_tokens": 4096,
# "timeout": 60,
# "rate_limit": {
# "requests_per_minute": 100,
# "tokens_per_minute": 100000
# }
# }
```
## Creating Production-Ready Agents
### Agent Factory Pattern
Create agents with production configuration:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## FastAPI Integration
Deploy agents as REST APIs:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Docker Deployment
### Dockerfile
```dockerfile
FROM python:3.12-slim
WORKDIR /app
# Install uv for faster dependency installation
RUN pip install uv
# Copy dependency files
COPY pyproject.toml uv.lock ./
# Install dependencies
RUN uv sync --frozen --no-dev
# Copy application code
COPY . .
# Set environment variables
ENV PYTHONUNBUFFERED=1
ENV DEPLOYMENT_ENV=production
# Expose port
EXPOSE 8000
# Run the application
CMD ["uv", "run", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
```
### Docker Compose
```yaml
version: '3.8'
services:
agent-api:
build: .
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- AGENT_MODEL=gpt-4o-mini
- DEPLOYMENT_ENV=production
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
deploy:
replicas: 3
resources:
limits:
memory: 512M
redis:
image: redis:7-alpine
ports:
- "6379:6379"
```
## Rate Limiting
Implement rate limiting to control API costs:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Graceful Shutdown
Handle shutdown signals properly:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Health Checks
Implement comprehensive health checks:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Best Practices Summary
| Area | Recommendation |
|------|----------------|
| Configuration | Use environment variables, never hardcode secrets |
| API Keys | Store in secrets manager (AWS Secrets Manager, Vault) |
| Scaling | Use async clients, implement connection pooling |
| Monitoring | Add health checks, log request/response metrics |
| Error Handling | Implement retries, circuit breakers, fallbacks |
| Rate Limiting | Respect API limits, implement client-side limiting |
| Shutdown | Handle signals, drain connections gracefully |
## Deployment Checklist
- [ ] Environment variables configured
- [ ] API keys stored securely
- [ ] Health check endpoint implemented
- [ ] Rate limiting configured
- [ ] Error handling and retries implemented
- [ ] Logging and monitoring set up
- [ ] Graceful shutdown handling
- [ ] Docker/container configuration
- [ ] Load balancing configured (if scaling)
- [ ] Backup/fallback providers configured
---
### Guides/Error Handling
# Error Handling Guide
This guide covers best practices for handling errors in Atomic Agents applications, including validation errors, API failures, and custom error handling patterns.
## Overview
Atomic Agents provides multiple layers of error handling:
1. **Schema Validation** - Pydantic validates input/output at runtime
2. **API Error Handling** - Handle LLM provider errors gracefully
3. **Hook System** - Monitor and respond to errors via hooks
4. **Custom Exception Handling** - Build robust error recovery patterns
## Schema Validation Errors
Pydantic schemas catch invalid data before it reaches the LLM.
### Basic Validation
```python
import os
from typing import List
from pydantic import Field, field_validator
import instructor
import openai
from atomic_agents import AtomicAgent, AgentConfig, BaseIOSchema
from atomic_agents.context import ChatHistory
class ValidatedInputSchema(BaseIOSchema):
"""Input schema with validation rules."""
query: str = Field(..., description="User query", min_length=1, max_length=1000)
max_results: int = Field(default=10, ge=1, le=100, description="Maximum results to return")
@field_validator('query')
@classmethod
def query_not_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError("Query cannot be empty or whitespace only")
return v.strip()
class ValidatedOutputSchema(BaseIOSchema):
"""Output schema with validation."""
answer: str = Field(..., description="The response")
confidence: float = Field(..., ge=0.0, le=1.0, description="Confidence score 0-1")
sources: List[str] = Field(default_factory=list, description="Source references")
# Initialize client and agent
client = instructor.from_openai(openai.OpenAI())
agent = AtomicAgent[ValidatedInputSchema, ValidatedOutputSchema](
config=AgentConfig(
client=client,
model="gpt-5-mini",
history=ChatHistory()
)
)
# Handle validation errors
try:
response = agent.run(ValidatedInputSchema(query="", max_results=5))
except ValueError as e:
print(f"Validation error: {e}")
```
### Custom Validators
```python
from pydantic import Field, field_validator, model_validator
from typing import Optional
from atomic_agents import BaseIOSchema
class SearchInputSchema(BaseIOSchema):
"""Search input with complex validation."""
query: str = Field(..., description="Search query")
category: Optional[str] = Field(None, description="Category filter")
date_from: Optional[str] = Field(None, description="Start date YYYY-MM-DD")
date_to: Optional[str] = Field(None, description="End date YYYY-MM-DD")
@field_validator('category')
@classmethod
def validate_category(cls, v: Optional[str]) -> Optional[str]:
valid_categories = ['technology', 'science', 'business', 'health']
if v is not None and v.lower() not in valid_categories:
raise ValueError(f"Category must be one of: {valid_categories}")
return v.lower() if v else None
@model_validator(mode='after')
def validate_dates(self):
if self.date_from and self.date_to:
if self.date_from > self.date_to:
raise ValueError("date_from must be before date_to")
return self
```
## API Error Handling
Handle LLM provider errors gracefully with retry logic.
### Basic Retry Pattern
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Using the Hook System for Error Handling
The Atomic Agents hook system provides powerful error monitoring capabilities.
### Error Logging Hook
```python
import os
import logging
from datetime import datetime
from typing import Any, Optional
import instructor
import openai
from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema
from atomic_agents.context import ChatHistory
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def on_error_hook(error: Exception, context: dict) -> None:
"""Hook called when an error occurs during agent execution."""
logger.error(f"Agent error: {type(error).__name__}: {error}")
logger.error(f"Context: {context}")
def on_completion_hook(response: Any, duration_ms: float) -> None:
"""Hook called on successful completion."""
logger.info(f"Agent completed in {duration_ms:.2f}ms")
# Create agent with hooks using Instructor's hook system
client = instructor.from_openai(openai.OpenAI())
# Register hooks with the instructor client
client.on("completion", lambda *args: on_completion_hook(*args))
agent = AtomicAgent[BasicChatInputSchema, BasicChatOutputSchema](
config=AgentConfig(
client=client,
model="gpt-5-mini",
history=ChatHistory()
)
)
```
### Comprehensive Error Handler
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Graceful Degradation
Implement fallback behavior when the primary agent fails.
### Fallback Agent Pattern
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Best Practices
### 1. Always Validate Input
```python
from pydantic import Field, field_validator
from atomic_agents import BaseIOSchema
class SafeInputSchema(BaseIOSchema):
"""Input schema with comprehensive validation."""
message: str = Field(..., min_length=1, max_length=10000)
@field_validator('message')
@classmethod
def sanitize_message(cls, v: str) -> str:
# Remove potential prompt injection attempts
dangerous_patterns = ['ignore previous', 'disregard instructions']
for pattern in dangerous_patterns:
if pattern.lower() in v.lower():
raise ValueError("Invalid input detected")
return v.strip()
```
### 2. Log All Errors
```python
import logging
from functools import wraps
logger = logging.getLogger(__name__)
def log_errors(func):
"""Decorator to log all errors from agent operations."""
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
logger.exception(f"Error in {func.__name__}: {e}")
raise
return wrapper
```
### 3. Set Timeouts
```python
import os
import instructor
import openai
from atomic_agents import AtomicAgent, AgentConfig
from atomic_agents.context import ChatHistory
# Configure timeout at client level
client = instructor.from_openai(
openai.OpenAI(timeout=30.0) # 30 second timeout
)
agent = AtomicAgent(
config=AgentConfig(
client=client,
model="gpt-5-mini",
history=ChatHistory(),
model_api_parameters={
"max_tokens": 500 # Limit response length
}
)
)
```
### 4. Implement Circuit Breaker
```python
import time
from typing import Optional, Callable
from dataclasses import dataclass
@dataclass
class CircuitBreaker:
"""Simple circuit breaker for agent calls."""
failure_threshold: int = 5
reset_timeout: float = 60.0
_failure_count: int = 0
_last_failure_time: float = 0
_state: str = "closed" # closed, open, half-open
def call(self, func: Callable, *args, **kwargs):
"""Execute function with circuit breaker protection."""
if self._state == "open":
if time.time() - self._last_failure_time > self.reset_timeout:
self._state = "half-open"
else:
raise Exception("Circuit breaker is open")
try:
result = func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
self._failure_count = 0
self._state = "closed"
def _on_failure(self):
self._failure_count += 1
self._last_failure_time = time.time()
if self._failure_count >= self.failure_threshold:
self._state = "open"
# Usage
circuit_breaker = CircuitBreaker(failure_threshold=3, reset_timeout=30.0)
def safe_agent_call(agent, input_data):
return circuit_breaker.call(agent.run, input_data)
```
## Summary
Key error handling strategies in Atomic Agents:
| Strategy | Use Case | Implementation |
|----------|----------|----------------|
| Schema Validation | Prevent invalid inputs | Pydantic validators |
| Retry Logic | Transient failures | Exponential backoff |
| Hook System | Monitoring & logging | Instructor hooks |
| Fallback Chain | High availability | Multiple agents |
| Circuit Breaker | Prevent cascade failures | State machine |
Always combine multiple strategies for robust production applications.
---
### Guides/Faq
# Frequently Asked Questions
Common questions and answers about using Atomic Agents.
## Installation & Setup
### How do I install Atomic Agents?
Install using pip:
```bash
pip install atomic-agents
```
Or using uv (recommended):
```bash
uv add atomic-agents
```
You also need to install your LLM provider. OpenAI is included by default. For other providers, use instructor extras:
```bash
# For Anthropic
pip install instructor[anthropic]
# For Groq
pip install instructor[groq]
# For Gemini
pip install instructor[google-genai]
```
### What Python version is required?
Atomic Agents requires **Python 3.12 or higher**.
```bash
# Check your Python version
python --version
```
### How do I set up my API key?
Set your API key as an environment variable:
```bash
# OpenAI
export OPENAI_API_KEY="your-api-key"
# Anthropic
export ANTHROPIC_API_KEY="your-api-key"
# Or use a .env file with python-dotenv
```
In your code:
```python
import os
from dotenv import load_dotenv
load_dotenv() # Load from .env file
# Keys are read from environment
api_key = os.getenv("OPENAI_API_KEY")
```
## Agent Configuration
### How do I create a basic agent?
```python
import instructor
import openai
from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema
from atomic_agents.context import ChatHistory
# Create instructor client
client = instructor.from_openai(openai.OpenAI())
# Create agent
agent = AtomicAgent[BasicChatInputSchema, BasicChatOutputSchema](
config=AgentConfig(
client=client,
model="gpt-5-mini",
history=ChatHistory()
)
)
# Use the agent
response = agent.run(BasicChatInputSchema(chat_message="Hello!"))
print(response.chat_message)
```
### How do I use different LLM providers?
Atomic Agents works with any provider supported by Instructor:
**OpenAI:**
```python
import instructor
import openai
client = instructor.from_openai(openai.OpenAI())
```
**Anthropic:**
```python
import instructor
from anthropic import Anthropic
client = instructor.from_anthropic(Anthropic())
```
**Groq:**
```python
import instructor
from groq import Groq
client = instructor.from_groq(Groq(), mode=instructor.Mode.JSON)
```
**Ollama (local models):**
```python
import instructor
from openai import OpenAI
client = instructor.from_openai(
OpenAI(
base_url="http://localhost:11434/v1",
api_key="ollama"
),
mode=instructor.Mode.JSON
)
```
**Google Gemini:**
```python
import instructor
from openai import OpenAI
import os
client = instructor.from_openai(
OpenAI(
api_key=os.getenv("GEMINI_API_KEY"),
base_url="https://generativelanguage.googleapis.com/v1beta/openai/"
),
mode=instructor.Mode.JSON
)
```
### How do I customize the system prompt?
Use `SystemPromptGenerator` to define agent behavior:
```python
from atomic_agents.context import SystemPromptGenerator
system_prompt = SystemPromptGenerator(
background=[
"You are a helpful coding assistant.",
"You specialize in Python programming."
],
steps=[
"Analyze the user's question.",
"Provide clear, working code examples.",
"Explain the code step by step."
],
output_instructions=[
"Always include code examples.",
"Use markdown formatting.",
"Keep explanations concise."
]
)
agent = AtomicAgent[BasicChatInputSchema, BasicChatOutputSchema](
config=AgentConfig(
client=client,
model="gpt-5-mini",
system_prompt_generator=system_prompt
)
)
```
### How do I add memory/conversation history?
Use `ChatHistory` to maintain conversation context:
```python
from atomic_agents.context import ChatHistory
# Create history
history = ChatHistory()
# Create agent with history
agent = AtomicAgent[BasicChatInputSchema, BasicChatOutputSchema](
config=AgentConfig(
client=client,
model="gpt-5-mini",
history=history
)
)
# Conversation is automatically maintained
agent.run(BasicChatInputSchema(chat_message="My name is Alice"))
agent.run(BasicChatInputSchema(chat_message="What's my name?")) # Will remember "Alice"
# Reset history when needed
agent.reset_history()
```
## Custom Schemas
### How do I create custom input/output schemas?
Inherit from `BaseIOSchema`:
```python
from typing import List, Optional
from pydantic import Field
from atomic_agents import BaseIOSchema
class CustomInputSchema(BaseIOSchema):
"""Custom input with additional fields."""
question: str = Field(..., description="The user's question")
context: Optional[str] = Field(None, description="Additional context")
max_length: int = Field(default=500, description="Max response length")
class CustomOutputSchema(BaseIOSchema):
"""Custom output with structured data."""
answer: str = Field(..., description="The answer to the question")
confidence: float = Field(..., ge=0.0, le=1.0, description="Confidence score")
sources: List[str] = Field(default_factory=list, description="Source references")
follow_up_questions: List[str] = Field(default_factory=list, description="Suggested follow-ups")
# Use with agent
agent = AtomicAgent[CustomInputSchema, CustomOutputSchema](
config=AgentConfig(client=client, model="gpt-5-mini")
)
response = agent.run(CustomInputSchema(question="What is Python?"))
print(f"Answer: {response.answer}")
print(f"Confidence: {response.confidence}")
```
### How do I add validation to schemas?
Use Pydantic validators:
```python
from pydantic import Field, field_validator, model_validator
from atomic_agents import BaseIOSchema
class ValidatedInputSchema(BaseIOSchema):
"""Input with validation rules."""
query: str = Field(..., min_length=1, max_length=1000)
category: str = Field(...)
@field_validator('category')
@classmethod
def validate_category(cls, v: str) -> str:
valid = ['tech', 'science', 'business']
if v.lower() not in valid:
raise ValueError(f"Category must be one of: {valid}")
return v.lower()
@field_validator('query')
@classmethod
def sanitize_query(cls, v: str) -> str:
return v.strip()
@model_validator(mode='after')
def validate_combination(self):
# Cross-field validation
if self.category == 'tech' and len(self.query) < 10:
raise ValueError("Tech queries must be at least 10 characters")
return self
```
## Tools
### How do I create a custom tool?
Inherit from `BaseTool`:
```python
import os
from pydantic import Field
from atomic_agents import BaseTool, BaseToolConfig, BaseIOSchema
class WeatherInputSchema(BaseIOSchema):
"""Input for weather tool."""
city: str = Field(..., description="City name to get weather for")
class WeatherOutputSchema(BaseIOSchema):
"""Output from weather tool."""
temperature: float = Field(..., description="Temperature in Celsius")
condition: str = Field(..., description="Weather condition")
humidity: int = Field(..., description="Humidity percentage")
class WeatherToolConfig(BaseToolConfig):
"""Configuration for weather tool."""
api_key: str = Field(default_factory=lambda: os.getenv("WEATHER_API_KEY"))
class WeatherTool(BaseTool[WeatherInputSchema, WeatherOutputSchema]):
"""Tool to fetch current weather."""
def __init__(self, config: WeatherToolConfig = None):
super().__init__(config or WeatherToolConfig())
self.api_key = self.config.api_key
def run(self, params: WeatherInputSchema) -> WeatherOutputSchema:
# Implement your tool logic here
# This is a mock implementation
return WeatherOutputSchema(
temperature=22.5,
condition="Sunny",
humidity=45
)
# Use the tool
tool = WeatherTool()
result = tool.run(WeatherInputSchema(city="London"))
print(f"Temperature: {result.temperature}Ā°C")
```
### How do I use the built-in tools?
Use the Atomic Assembler CLI to download tools:
```bash
atomic
```
Then import and use them:
```python
from calculator.tool.calculator import CalculatorTool, CalculatorInputSchema
calculator = CalculatorTool()
result = calculator.run(CalculatorInputSchema(expression="2 + 2 * 3"))
print(result.value) # 8.0
```
## Streaming & Async
### How do I stream responses?
Use `run_stream()` for synchronous streaming:
```python
from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema
# Synchronous streaming
for partial in agent.run_stream(BasicChatInputSchema(chat_message="Write a poem")):
print(partial.chat_message, end='', flush=True)
print() # Newline at end
```
### How do I use async methods?
Use `run_async()` for async operations:
```python
import asyncio
from openai import AsyncOpenAI
import instructor
from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema
from atomic_agents.context import ChatHistory
async def main():
# Use async client
client = instructor.from_openai(AsyncOpenAI())
agent = AtomicAgent[BasicChatInputSchema, BasicChatOutputSchema](
config=AgentConfig(
client=client,
model="gpt-5-mini",
history=ChatHistory()
)
)
# Non-streaming async
response = await agent.run_async(BasicChatInputSchema(chat_message="Hello"))
print(response.chat_message)
# Streaming async
async for partial in agent.run_async_stream(BasicChatInputSchema(chat_message="Write a story")):
print(partial.chat_message, end='', flush=True)
asyncio.run(main())
```
## Context Providers
### How do I inject dynamic context?
Create a custom context provider:
```python
from typing import List
from atomic_agents.context import BaseDynamicContextProvider
class SearchResultsProvider(BaseDynamicContextProvider):
"""Provides search results as context."""
def __init__(self, title: str = "Search Results"):
super().__init__(title=title)
self.results: List[str] = []
def add_result(self, result: str):
self.results.append(result)
def clear(self):
self.results = []
def get_info(self) -> str:
if not self.results:
return "No search results available."
return "\n".join(f"- {r}" for r in self.results)
# Register with agent
provider = SearchResultsProvider()
provider.add_result("Python is a programming language")
provider.add_result("Python was created by Guido van Rossum")
agent.register_context_provider("search_results", provider)
# The context is now included in the system prompt
response = agent.run(BasicChatInputSchema(chat_message="Tell me about Python"))
```
## Common Issues
### Why am I getting validation errors?
Check that your input matches the schema:
```python
from pydantic import ValidationError
try:
response = agent.run(BasicChatInputSchema(chat_message=""))
except ValidationError as e:
print("Validation errors:")
for error in e.errors():
print(f" {error['loc']}: {error['msg']}")
```
### How do I handle API rate limits?
Implement retry logic:
```python
import time
from openai import RateLimitError
def run_with_retry(agent, input_data, max_retries=3):
for attempt in range(max_retries):
try:
return agent.run(input_data)
except RateLimitError:
if attempt < max_retries - 1:
wait = 2 ** attempt # Exponential backoff
print(f"Rate limited. Waiting {wait}s...")
time.sleep(wait)
else:
raise
```
### How do I debug agent behavior?
1. **Check the system prompt:**
```python
print(agent.system_prompt_generator.generate_prompt())
```
2. **Inspect history:**
```python
for msg in agent.history.get_history():
print(f"{msg['role']}: {msg['content']}")
```
3. **Enable logging:**
```python
import logging
logging.basicConfig(level=logging.DEBUG)
```
## MCP Integration
### How do I connect to an MCP server?
```python
from atomic_agents.connectors.mcp import fetch_mcp_tools_async, MCPTransportType
async def setup_mcp_tools():
tools = await fetch_mcp_tools_async(
server_url="http://localhost:8000",
transport_type=MCPTransportType.HTTP_STREAM
)
return tools
# Use tools with your agent
tools = asyncio.run(setup_mcp_tools())
```
## Migration
### How do I upgrade from v1.x to v2.0?
Key changes:
1. **Import paths:**
```python
# Old
from atomic_agents.lib.base.base_io_schema import BaseIOSchema
# New
from atomic_agents import BaseIOSchema
```
2. **Class names:**
```python
# Old
from atomic_agents.agents.base_agent import BaseAgent, BaseAgentConfig
# New
from atomic_agents import AtomicAgent, AgentConfig
```
3. **Schemas as type parameters:**
```python
# Old
agent = BaseAgent(BaseAgentConfig(
client=client,
model="gpt-5-mini",
input_schema=MyInput,
output_schema=MyOutput
))
# New
agent = AtomicAgent[MyInput, MyOutput](
AgentConfig(client=client, model="gpt-5-mini")
)
```
See the [Upgrade Guide](../UPGRADE_DOC.md) for complete migration instructions.
---
### Guides/Hooks
# Hooks Guide
This guide covers the hook system in Atomic Agents, enabling comprehensive monitoring, error handling, and intelligent retry mechanisms.
## Overview
The Atomic Agents hook system integrates with Instructor's event system to provide:
- **Comprehensive Monitoring**: Track all aspects of agent execution
- **Robust Error Handling**: Graceful handling of validation and completion errors
- **Intelligent Retry Patterns**: Implement smart retry logic based on error context
- **Performance Metrics**: Monitor response times, success rates, and error patterns
- **Zero Overhead**: Hooks only execute when registered and enabled
## Supported Hook Events
| Event | Description | When Triggered |
|-------|-------------|----------------|
| `parse:error` | Pydantic validation failures | When LLM output doesn't match schema |
| `completion:kwargs` | Before API calls | Just before sending request to LLM |
| `completion:response` | After API responses | When LLM returns a response |
| `completion:error` | API or network errors | On connection failures, timeouts, etc. |
## Basic Hook Registration
Register hooks using the `register_hook` method on any `AtomicAgent`:
```python
import os
import instructor
import openai
from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema
from atomic_agents.context import ChatHistory
def on_parse_error(error):
"""Handle validation errors."""
print(f"Validation failed: {error}")
def on_completion_kwargs(**kwargs):
"""Log API call details before request."""
model = kwargs.get("model", "unknown")
print(f"Calling model: {model}")
def on_completion_response(response, **kwargs):
"""Process successful responses."""
if hasattr(response, "usage"):
print(f"Tokens used: {response.usage.total_tokens}")
def on_completion_error(error, **kwargs):
"""Handle API errors."""
print(f"API error: {type(error).__name__}: {error}")
# Create agent
client = instructor.from_openai(openai.OpenAI())
agent = AtomicAgent[BasicChatInputSchema, BasicChatOutputSchema](
config=AgentConfig(
client=client,
model="gpt-4o-mini",
history=ChatHistory()
)
)
# Register hooks
agent.register_hook("parse:error", on_parse_error)
agent.register_hook("completion:kwargs", on_completion_kwargs)
agent.register_hook("completion:response", on_completion_response)
agent.register_hook("completion:error", on_completion_error)
# Use the agent normally - hooks are called automatically
response = agent.run(BasicChatInputSchema(chat_message="Hello!"))
```
## Performance Monitoring
Track request metrics for performance analysis:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Detailed Validation Error Handling
Extract detailed information from validation errors:
```python
from pydantic import ValidationError
def detailed_parse_error_handler(error):
"""Extract detailed validation error information."""
if isinstance(error, ValidationError):
print("Validation Error Details:")
for err in error.errors():
# Get field path (e.g., "confidence" or "nested.field")
field_path = " -> ".join(str(x) for x in err["loc"])
error_type = err["type"]
message = err["msg"]
print(f" Field: {field_path}")
print(f" Type: {error_type}")
print(f" Message: {message}")
# Access input value if available
if "input" in err:
print(f" Invalid Value: {err['input']}")
else:
print(f"Parse Error: {error}")
agent.register_hook("parse:error", detailed_parse_error_handler)
```
## Retry Strategies with Hooks
Implement intelligent retry logic based on error context:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Managing Hooks
### Enable/Disable Hooks
Temporarily disable hooks without unregistering:
```python
# Disable all hooks
agent.disable_hooks()
# Run without hook overhead
response = agent.run(input_data)
# Re-enable hooks
agent.enable_hooks()
# Check if hooks are enabled
if agent.hooks_enabled():
print("Hooks are active")
```
### Unregister Hooks
Remove specific hooks or clear all:
```python
# Unregister a specific hook
agent.unregister_hook("parse:error", on_parse_error)
# Clear all hooks
agent.clear_hooks()
```
## Production Logging Pattern
A complete production-ready logging setup:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Best Practices
### 1. Keep Hooks Lightweight
Hooks run synchronously - avoid heavy operations:
```python
# Good: Quick logging
def on_response(response, **kwargs):
logger.info(f"Response received")
# Avoid: Heavy processing in hooks
def on_response_slow(response, **kwargs):
# Don't do this - blocks the response
save_to_database(response)
send_to_analytics(response)
generate_report(response)
```
### 2. Handle Hook Exceptions
Wrap hook logic to prevent failures from disrupting the agent:
```python
def safe_hook(func):
"""Decorator to catch hook exceptions."""
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
logger.error(f"Hook error in {func.__name__}: {e}")
return wrapper
@safe_hook
def on_completion_response(response, **kwargs):
# If this fails, the agent continues working
process_response(response)
```
### 3. Use Hooks for Cross-Cutting Concerns
Hooks are ideal for:
- Logging and monitoring
- Metrics collection
- Error tracking
- Performance profiling
- Audit trails
### 4. Don't Modify Responses in Hooks
Hooks are for observation, not transformation:
```python
# Good: Observe and log
def on_response(response, **kwargs):
logger.info(f"Got response: {response}")
# Avoid: Trying to modify response
def on_response_bad(response, **kwargs):
response.chat_message = "Modified" # Don't do this
```
## Summary
| Feature | Method | Description |
|---------|--------|-------------|
| Register hook | `agent.register_hook(event, callback)` | Add a hook callback |
| Unregister hook | `agent.unregister_hook(event, callback)` | Remove specific hook |
| Clear all hooks | `agent.clear_hooks()` | Remove all hooks |
| Enable hooks | `agent.enable_hooks()` | Activate hook system |
| Disable hooks | `agent.disable_hooks()` | Deactivate hook system |
| Check status | `agent.hooks_enabled()` | Check if hooks active |
Use hooks to add monitoring and error handling to your agents without modifying core business logic.
---
### Guides/Index
# User Guide
This section contains detailed guides for working with Atomic Agents.
```{toctree}
:maxdepth: 2
:caption: Guides
quickstart
memory
tools
hooks
orchestration
cookbook
error-handling
testing
deployment
performance
security
logging
faq
```
## Implementation Patterns
The framework supports various implementation patterns and use cases:
### Chatbots and Assistants
- Basic chat interfaces with any LLM provider
- Streaming responses
- Custom response schemas
- Suggested follow-up questions
- History management and context retention
- Multi-turn conversations
### RAG Systems
- Query generation and optimization
- Context-aware responses
- Document Q&A with source tracking
- Information synthesis and summarization
- Custom embedding and retrieval strategies
- Hybrid search approaches
### Specialized Agents
- YouTube video summarization and analysis
- Web search and deep research
- Recipe generation from various sources
- Multimodal interactions (text, images, etc.)
- Custom tool integration
- Custom MCP integration to support tools, resources, and prompts
- Task orchestration
## Provider Integration Guide
Atomic Agents is designed to be provider-agnostic. Here's how to work with different providers:
### Provider Selection
- Choose any provider supported by Instructor
- Configure provider-specific settings
- Handle rate limits and quotas
- Implement fallback strategies
### Local Development
- Use Ollama for local testing
- Mock responses for development
- Debug provider interactions
- Test provider switching
### Production Deployment
- Load balancing between providers
- Failover configurations
- Cost optimization strategies
- Performance monitoring
### Custom Provider Integration
- Extend Instructor for new providers
- Implement custom client wrappers
- Add provider-specific features
- Handle unique response formats
## Best Practices
### Error Handling
- Implement proper exception handling
- Add retry mechanisms
- Log provider errors
- Handle rate limits gracefully
### Performance Optimization
- Use streaming for long responses
- Implement caching strategies
- Optimize prompt lengths
- Batch operations when possible
### Security
- Secure API key management
- Input validation and sanitization
- Output filtering
- Rate limiting and quotas
## Getting Help
If you need help, you can:
1. Check our [GitHub Issues](https://github.com/eigenwise/atomic-agents/issues)
2. Join our [Reddit community](https://www.reddit.com/r/AtomicAgents/)
3. Read through our examples in the repository
4. Review the example projects in `atomic-examples/`
**See also**:
- [API Reference](/api/index) - Browse the API reference
- [Main Documentation](/index) - Return to main documentation
---
### Guides/Logging
# Logging and Monitoring Guide
This guide covers logging, monitoring, and observability best practices for Atomic Agents applications.
## Overview
Effective logging and monitoring enables:
- **Debugging**: Trace issues in agent behavior
- **Performance Tracking**: Identify bottlenecks
- **Cost Monitoring**: Track API usage and costs
- **Alerting**: Detect anomalies and failures
- **Auditing**: Maintain records for compliance
## Basic Logging Setup
### Configure Python Logging
Set up structured logging for agents:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Agent Logging with Hooks
### Comprehensive Request Logging
Use hooks to log all agent interactions:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Metrics Collection
### Token and Cost Tracking
Track API usage and costs:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Monitoring Dashboard
### FastAPI Metrics Endpoint
Expose metrics via HTTP:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Distributed Tracing
### OpenTelemetry Integration
Add distributed tracing for complex systems:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Alerting
### Alert Conditions
Define alert conditions for monitoring:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Log Analysis Patterns
### Structured Log Queries
Design logs for easy querying:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Best Practices
### Logging Guidelines
| What to Log | Why | Example |
|-------------|-----|---------|
| Request IDs | Trace requests | `request_id: "abc123"` |
| Timestamps | Timeline analysis | `timestamp: "2024-01-15T10:30:00Z"` |
| Model used | Cost attribution | `model: "gpt-4o-mini"` |
| Token counts | Usage tracking | `tokens: {"prompt": 100, "completion": 50}` |
| Latency | Performance monitoring | `duration_ms: 1523` |
| Error types | Debugging | `error_type: "ValidationError"` |
| User IDs | Audit trails | `user_id: "user456"` |
### What NOT to Log
- Full request/response content (privacy)
- API keys or secrets
- Personal identifiable information (PII)
- Sensitive business data
## Summary
| Component | Purpose | Tools |
|-----------|---------|-------|
| Logging | Debug & audit | Python logging, structured JSON |
| Metrics | Performance tracking | Custom collectors, Prometheus |
| Tracing | Request flow | OpenTelemetry, Jaeger |
| Alerting | Issue detection | Custom rules, PagerDuty |
| Dashboards | Visualization | Grafana, custom endpoints |
Implement logging and monitoring from the start - it's much harder to add later.
---
### Guides/Memory
# Memory and Context Management
This guide covers everything you need to know about managing conversation memory and dynamic context in Atomic Agents. Whether you're building a simple chatbot or orchestrating complex multi-agent systems, understanding memory management is essential.
```{contents}
:local:
:depth: 2
```
## Introduction
### What You'll Learn
- How conversation history works in Atomic Agents
- What "turns" are and how they're tracked
- How messages are automatically managed during agent execution
- How to persist and restore conversation state
- How to use context providers for dynamic information injection
- Advanced multi-agent memory patterns
### Prerequisites
- Basic familiarity with Atomic Agents ([Quickstart Guide](quickstart.md))
- Understanding of Python classes and async/await
### The Problem This Solves
A common question from developers (see [GitHub Issue #58](https://github.com/eigenwise/atomic-agents/issues/58)):
> "In most of the examples only the initial message is added, not any subsequent runs. Is this automatic?"
**Yes, it is automatic!** When you call `agent.run(user_input)`, the framework automatically:
1. Adds your input to the conversation history
2. Sends the full history to the LLM
3. Adds the LLM's response to history
This guide explains exactly how this works and how to leverage it for complex use cases.
---
## Understanding Memory in Atomic Agents
### The Conversation Model
Atomic Agents uses a **turn-based conversation model** where each interaction between user and assistant forms a "turn". The `ChatHistory` class manages this conversation state.
```{mermaid}
flowchart LR
subgraph Turn1["Turn 1 (turn_id: abc-123)"]
U1[User Message]
A1[Assistant Response]
end
subgraph Turn2["Turn 2 (turn_id: def-456)"]
U2[User Message]
A2[Assistant Response]
end
subgraph Turn3["Turn 3 (turn_id: ghi-789)"]
U3[User Message]
A3[Assistant Response]
end
U1 --> A1
A1 -.-> U2
U2 --> A2
A2 -.-> U3
U3 --> A3
```
**Key Concepts:**
- **Message**: A single piece of content with a role (user, assistant, system)
- **Turn**: A logical grouping of related messages (typically user input + assistant response)
- **Turn ID**: A UUID that links messages belonging to the same turn
- **History**: The complete sequence of messages in a conversation
### Messages and Turns
Each message in the history has three components:
```python
from atomic_agents.context import Message
# Message structure
message = Message(
role="user", # "user", "assistant", or "system"
content=some_schema, # Must be a BaseIOSchema instance
turn_id="abc-123" # UUID linking related messages
)
```
**Why Turn IDs Matter:**
- Group related messages together
- Enable deletion of complete turns (user message + response)
- Track conversation flow for debugging
- Support conversation branching patterns
---
## ChatHistory Fundamentals
### Creating and Configuring History
```python
from atomic_agents.context import ChatHistory
# Basic history (unlimited messages)
history = ChatHistory()
# History with message limit (oldest messages removed when exceeded)
history = ChatHistory(max_messages=50)
```
### Using History with an Agent
```python
import instructor
import openai
from atomic_agents import AtomicAgent, AgentConfig, BaseIOSchema
from atomic_agents.context import ChatHistory
from pydantic import Field
# Define schemas
class ChatInput(BaseIOSchema):
"""User chat message"""
message: str = Field(..., description="The user's message")
class ChatOutput(BaseIOSchema):
"""Assistant response"""
response: str = Field(..., description="The assistant's response")
# Create history and agent
history = ChatHistory(max_messages=100)
client = instructor.from_openai(openai.OpenAI())
agent = AtomicAgent[ChatInput, ChatOutput](
config=AgentConfig(
client=client,
model="gpt-5-mini",
history=history,
)
)
# Each run automatically manages history
response1 = agent.run(ChatInput(message="Hello!"))
response2 = agent.run(ChatInput(message="What did I just say?"))
# The agent remembers the previous message!
```
### The Turn Lifecycle
```{mermaid}
stateDiagram-v2
[*] --> NoTurn: ChatHistory created
NoTurn --> ActiveTurn: initialize_turn() called
NoTurn --> ActiveTurn: add_message() called
ActiveTurn --> ActiveTurn: add_message() same turn
ActiveTurn --> NewTurn: initialize_turn() called
NewTurn --> ActiveTurn: Generates new UUID
ActiveTurn --> NoTurn: All turns deleted
note right of ActiveTurn: current_turn_id = UUID
note right of NoTurn: current_turn_id = None
```
**Turn Lifecycle Methods:**
```python
# Initialize a new turn (generates new UUID)
history.initialize_turn()
# Get the current turn ID
turn_id = history.get_current_turn_id()
print(f"Current turn: {turn_id}") # e.g., "abc-123-def-456"
# Add a message to the current turn
history.add_message("user", ChatInput(message="Hello"))
# Messages added without initialize_turn() use the existing turn
# or auto-initialize if no turn exists
```
---
## Automatic Memory Management
This section addresses the core question from GitHub Issue #58: **How does automatic message management work?**
### How .run() Manages Memory
When you call `agent.run(user_input)`, here's exactly what happens:
```{mermaid}
flowchart TD
A["agent.run(user_input)"] --> B{user_input
provided?}
B -->|Yes| C["history.initialize_turn()
Creates new UUID"]
C --> D["history.add_message('user', user_input)
Stores user message"]
B -->|No| E["Skip turn initialization
Use existing history"]
D --> F["_prepare_messages()
Build message list"]
E --> F
F --> G["System prompt + history"]
G --> H["LLM API call"]
H --> I["Receive response"]
I --> J["history.add_message('assistant', response)
Stores response"]
J --> K["_manage_overflow()
Trim if needed"]
K --> L["Return response"]
style C fill:#e1f5fe
style D fill:#e1f5fe
style J fill:#e1f5fe
```
### Step-by-Step Trace
Let's trace through a complete conversation:
```python
from atomic_agents import AtomicAgent, AgentConfig, BaseIOSchema
from atomic_agents.context import ChatHistory
from pydantic import Field
class Input(BaseIOSchema):
"""Input"""
text: str = Field(...)
class Output(BaseIOSchema):
"""Output"""
reply: str = Field(...)
# Create agent with history
history = ChatHistory()
agent = AtomicAgent[Input, Output](config=AgentConfig(
client=client,
model="gpt-5-mini",
history=history
))
# --- TURN 1 ---
print(f"Before run: {history.get_message_count()} messages") # 0 messages
response1 = agent.run(Input(text="Hi, my name is Alice"))
# Internally:
# 1. history.initialize_turn() -> turn_id = "abc-123"
# 2. history.add_message("user", Input(text="Hi..."))
# 3. LLM called with history
# 4. history.add_message("assistant", Output(reply="Hello Alice!"))
print(f"After run 1: {history.get_message_count()} messages") # 2 messages
print(f"Turn ID: {history.get_current_turn_id()}") # "abc-123"
# --- TURN 2 ---
response2 = agent.run(Input(text="What's my name?"))
# Internally:
# 1. history.initialize_turn() -> turn_id = "def-456" (NEW turn)
# 2. history.add_message("user", Input(text="What's..."))
# 3. LLM called with FULL history (all 4 messages)
# 4. history.add_message("assistant", Output(reply="Your name is Alice!"))
print(f"After run 2: {history.get_message_count()} messages") # 4 messages
print(f"Turn ID: {history.get_current_turn_id()}") # "def-456"
```
### Running Without Input
You can call `.run()` without input to continue within the same turn:
```python
# First call with input - starts new turn
response = agent.run(Input(text="Start a story"))
# Subsequent call without input - same turn continues
# Useful for: tool follow-ups, multi-step reasoning
continuation = agent.run() # No new turn created, uses existing history
```
### Streaming and Async Behavior
All execution methods handle memory the same way:
| Method | Memory Behavior |
|--------|-----------------|
| `agent.run(input)` | Automatic turn init + message add |
| `agent.run_stream(input)` | Same as run(), streams response |
| `agent.run_async(input)` | Same as run(), async execution |
| `agent.run_async_stream(input)` | Same as run(), async + streaming |
```python
# Streaming example - memory works identically
async for chunk in agent.run_async_stream(Input(text="Hello")):
print(chunk.reply, end="", flush=True)
# History is updated with complete response after stream finishes
```
---
## History Persistence and Management
### Serialization: Saving Conversations
Save conversation history to disk or database:
```python
from atomic_agents.context import ChatHistory
# ... after some conversation ...
# Serialize to JSON string
serialized = history.dump()
# Save to file
with open("conversation.json", "w") as f:
f.write(serialized)
# Save to database
db.save_conversation(user_id=123, data=serialized)
```
### Deserialization: Restoring Conversations
```python
# Load from file
with open("conversation.json", "r") as f:
serialized = f.read()
# Create new history and load
history = ChatHistory()
history.load(serialized)
# Use with agent
agent = AtomicAgent[Input, Output](config=AgentConfig(
client=client,
model="gpt-5-mini",
history=history, # Restored history!
))
# Continue the conversation where it left off
response = agent.run(Input(text="Where were we?"))
```
```{warning}
Only load serialized data from trusted sources. The `load()` method reconstructs Python classes from the serialized data.
```
### Overflow Management
Control memory usage with `max_messages`:
```python
# Keep only last 20 messages
history = ChatHistory(max_messages=20)
# When 21st message is added, oldest message is removed
# This is FIFO (First In, First Out) - oldest messages go first
```
**Strategy for Long Conversations:**
```python
# Option 1: Simple limit
history = ChatHistory(max_messages=50)
# Option 2: Monitor and handle manually
if history.get_message_count() > 40:
# Maybe summarize old messages before they're lost
old_messages = history.get_history()[:10]
summary = summarize_messages(old_messages)
# Store summary in context provider instead
```
### History Manipulation
**Copying History:**
```python
# Create independent copy (deep copy)
history_copy = history.copy()
# Modifications don't affect original
history_copy.add_message("user", Input(text="This only goes in copy"))
```
**Deleting Turns:**
```python
# Get the turn ID you want to delete
turn_id = history.get_current_turn_id()
# Delete all messages with that turn ID
history.delete_turn_id(turn_id)
# Useful for: removing failed attempts, undo functionality
```
**Resetting History:**
```python
# Clear all messages, start fresh
agent.reset_history()
# or
history = ChatHistory() # Create new instance
```
---
## Writing a Custom Memory Backend
Memory is intentionally kept outside the framework core. `ChatHistory` is a solid in-memory default, but if you want conversations to survive a restart, live in Redis, Postgres, or some hosted memory service, that integration belongs in **your** codebase, not as a framework dependency. `BaseChatHistory` is the stable seam that makes this possible: build against it and `AtomicAgent` doesn't care what's underneath.
### The Contract
`BaseChatHistory` is an interface-only abstract base class (`ABC`). It declares no state and no behavior of its own, just the methods a history implementation must provide:
- `initialize_turn(self) -> None`
- `add_message(self, role: str, content: BaseIOSchema) -> None`
- `get_history(self) -> List[Dict]`
- `get_current_turn_id(self) -> Optional[str]`
- `delete_turn_id(self, turn_id: str)`
- `get_message_count(self) -> int`
- `dump(self) -> str`
- `load(self, serialized_data: str) -> None`
- `copy(self) -> "BaseChatHistory"`
Every implementation also has to maintain two attributes, because `AtomicAgent` reads them directly:
- `history: List[Message]` - the in-memory working list of messages. `AtomicAgent._trim_context` reads this list (via each message's `turn_id`) to decide what to trim when a context-length limit is hit.
- `current_turn_id: Optional[str]` - the ID of the current turn, or `None` if no turn has started yet. Set by `initialize_turn()`.
### Recommended Pattern: Subclass ChatHistory
Don't implement `BaseChatHistory` from scratch unless you have a good reason to. Subclassing `ChatHistory` gets you turn handling and serialization for free, you only need to override the couple of methods that touch your storage layer:
```python
from atomic_agents.context import ChatHistory
class PersistentHistory(ChatHistory):
def __init__(self, session_id, store, max_messages=None):
super().__init__(max_messages=max_messages)
self.session_id = session_id
self.store = store
saved = store.get(session_id)
if saved:
self.load(saved)
def add_message(self, role, content):
super().add_message(role, content)
self.store.put(self.session_id, self.dump())
def copy(self):
# AtomicAgent calls copy() for its initial_history snapshot and reset_history().
# Without this override, copy() returns a plain ChatHistory and your backend is
# silently dropped after reset_history() (later writes stop reaching the store).
new_history = PersistentHistory(self.session_id, self.store, max_messages=self.max_messages)
new_history.load(self.dump())
new_history.current_turn_id = self.current_turn_id
return new_history
```
The core of the pattern is small: load on init if there's saved state, persist on every `add_message`. `dump()`/`load()` already round-trip the full history, so there's no serialization logic to write yourself.
```{important}
Override `copy()` too. `AtomicAgent` snapshots `initial_history` with `copy()` at construction and restores it in `reset_history()`. The built-in `ChatHistory.copy()` hard-codes a plain `ChatHistory`, so a subclass that carries extra state (a store handle, a `session_id`) that forgets to override `copy()` will be **silently replaced by a non-persistent in-memory history** the first time `reset_history()` runs. Any subclass with its own state must override `copy()` to return its own type.
```
```{note}
This pattern re-serializes and rewrites the **entire** history on every message (`self.dump()`), which is fine for a demo but O(nĀ²) over a long conversation. A production backend against Redis/Postgres/etc. should persist incrementally (append the new message) rather than dumping the whole history each write.
```
### From-Scratch Alternative
If you don't want the in-memory list at all (say your backend is a query, not a cache), implement `BaseChatHistory` directly. You give up the free turn tracking and `get_history()` serialization that `ChatHistory` provides, so you own all of it: generating turn IDs, building the wire-format list of dicts, and keeping `history` / `current_turn_id` in sync with whatever `AtomicAgent` expects to read.
### Try It
There's a full runnable example of this pattern (a dependency-free SQLite-backed history that survives across process runs) at [`atomic-examples/persistent-memory`](https://github.com/eigenwise/atomic-agents/tree/main/atomic-examples/persistent-memory).
---
## Multimodal Content in History
ChatHistory supports images, PDFs, and audio through Instructor's multimodal types, plus
video through the framework's own `VideoURL` type.
### Adding Multimodal Messages
```python
from instructor import Image, PDF, Audio
from atomic_agents import BaseIOSchema
from pydantic import Field
from typing import List
class ImageAnalysisInput(BaseIOSchema):
"""Input with images for analysis"""
question: str = Field(..., description="Question about the images")
images: List[Image] = Field(..., description="Images to analyze")
# Create input with images
input_with_images = ImageAnalysisInput(
question="What's in these images?",
images=[
Image.from_path("photo1.jpg"),
Image.from_path("photo2.png"),
]
)
# Run agent - images are stored in history
response = agent.run(input_with_images)
```
### Multimodal Message Structure
When history contains multimodal content, `get_history()` returns a special structure:
```python
history_data = history.get_history()
for message in history_data:
if isinstance(message["content"], list):
# Multimodal message
json_content = message["content"][0] # Text/JSON data
multimodal_objects = message["content"][1:] # Images, PDFs, etc.
else:
# Text-only message
json_content = message["content"]
```
### Video
Instructor has no video type yet, so Atomic Agents ships its own `VideoURL` for providers
that accept OpenAI-compatible `video_url` content parts (MiniMax, Qwen-VL, ...):
```python
from atomic_agents import BaseIOSchema, VideoURL
from pydantic import Field
class VideoAnalysisInput(BaseIOSchema):
"""Input with a video for analysis"""
question: str = Field(..., description="Question about the video")
video: VideoURL = Field(..., description="Video to analyze")
input_with_video = VideoAnalysisInput(
question="What happens in this clip?",
video=VideoURL(url="https://example.com/clip.mp4", fps=1.0),
)
```
`get_history()` emits the video as a `{"type": "video_url", ...}` content part, which
Instructor passes through to the provider unchanged.
### Serialization with Multimodal
```{note}
Multimodal content with file paths is serialized by path. Ensure files exist at the same paths when loading.
```
```python
# Serialize (file paths are preserved)
serialized = history.dump()
# When loading, files must be accessible at original paths
history.load(serialized)
```
---
## Dynamic Context with Providers
Context providers inject dynamic information into agent system prompts at runtime, complementing the static conversation history.
### Understanding the Difference
| Aspect | ChatHistory (Memory) | Context Providers |
|--------|---------------------|-------------------|
| **Purpose** | Store conversation turns | Inject dynamic context |
| **Location** | Message history | System prompt |
| **Persistence** | Saved with history | Regenerated each call |
| **Use Case** | Conversation continuity | Real-time data (RAG, user info, time) |
```{mermaid}
flowchart TB
subgraph SystemPrompt["System Prompt (sent to LLM)"]
BG[Background Instructions]
ST[Steps]
subgraph DC["Dynamic Context"]
CP1[Context Provider 1]
CP2[Context Provider 2]
CP3[Context Provider 3]
end
OI[Output Instructions]
end
subgraph Messages["Conversation Messages"]
H[ChatHistory Messages]
end
SystemPrompt --> LLM
Messages --> LLM
LLM --> Response
```
### Creating Custom Context Providers
```python
from atomic_agents.context import BaseDynamicContextProvider
class UserContextProvider(BaseDynamicContextProvider):
"""Provides current user information to the agent."""
def __init__(self):
super().__init__(title="Current User")
self.user_name: str = ""
self.user_role: str = ""
self.preferences: dict = {}
def get_info(self) -> str:
"""Called every time the agent runs."""
if not self.user_name:
return "No user logged in."
info = f"User: {self.user_name} (Role: {self.user_role})"
if self.preferences:
prefs = ", ".join(f"{k}: {v}" for k, v in self.preferences.items())
info += f"\nPreferences: {prefs}"
return info
```
### Registering Context Providers
```python
from atomic_agents import AtomicAgent, AgentConfig
from atomic_agents.context import SystemPromptGenerator
# Create provider
user_provider = UserContextProvider()
# Option 1: Register with SystemPromptGenerator
system_prompt = SystemPromptGenerator(
background=["You are a helpful assistant."],
context_providers={"user": user_provider}
)
agent = AtomicAgent[Input, Output](config=AgentConfig(
client=client,
model="gpt-5-mini",
system_prompt_generator=system_prompt,
))
# Option 2: Register after agent creation
agent.register_context_provider("user", user_provider)
# Update provider state before running
user_provider.user_name = "Alice"
user_provider.user_role = "Admin"
# Now the agent knows about Alice!
response = agent.run(Input(text="What can I do?"))
```
### Common Context Provider Patterns
**RAG (Retrieval-Augmented Generation):**
```python
class RAGContextProvider(BaseDynamicContextProvider):
"""Injects retrieved documents into the prompt."""
def __init__(self, vector_db):
super().__init__(title="Relevant Documents")
self.vector_db = vector_db
self.current_query: str = ""
self._cached_results: list = []
def search(self, query: str, top_k: int = 3):
"""Call before agent.run() to update context."""
self.current_query = query
self._cached_results = self.vector_db.search(query, top_k=top_k)
def get_info(self) -> str:
if not self._cached_results:
return "No relevant documents found."
docs = []
for i, doc in enumerate(self._cached_results, 1):
docs.append(f"Document {i}:\n{doc['content']}\nSource: {doc['source']}")
return "\n\n".join(docs)
# Usage
rag_provider = RAGContextProvider(vector_db)
agent.register_context_provider("documents", rag_provider)
# Before each query
user_query = "How do I reset my password?"
rag_provider.search(user_query) # Update context
response = agent.run(Input(text=user_query))
```
**Time-Aware Context:**
```python
from datetime import datetime
class TimeContextProvider(BaseDynamicContextProvider):
"""Provides current time information."""
def __init__(self):
super().__init__(title="Current Time")
def get_info(self) -> str:
now = datetime.now()
return f"Current date/time: {now.strftime('%Y-%m-%d %H:%M:%S %Z')}"
```
**Session Context:**
```python
class SessionContextProvider(BaseDynamicContextProvider):
"""Tracks session-specific state."""
def __init__(self):
super().__init__(title="Session State")
self.data: dict = {}
def set(self, key: str, value: str):
self.data[key] = value
def get_info(self) -> str:
if not self.data:
return "No session data."
return "\n".join(f"- {k}: {v}" for k, v in self.data.items())
```
---
## Multi-Agent Memory Patterns
This section addresses the question from GitHub Issue #58:
> "How do I handle a scenario where one agent performs an action, a second agent evaluates it, and then passes results back to the first agent's memory?"
Here are five patterns for managing memory across multiple agents.
### Pattern 1: Shared History
Multiple agents share the same `ChatHistory` instance, seeing each other's messages.
```{mermaid}
flowchart LR
subgraph SharedHistory["Shared ChatHistory"]
M1[Message 1]
M2[Message 2]
M3[Message 3]
M4[Message 4]
end
A1[Agent A] --> SharedHistory
A2[Agent B] --> SharedHistory
A3[Agent C] --> SharedHistory
```
**Use Case:** Agents that need full conversation context (e.g., specialist + generalist).
```python
from atomic_agents import AtomicAgent, AgentConfig
from atomic_agents.context import ChatHistory
# One history shared by all
shared_history = ChatHistory()
# Agent A - Technical Expert
technical_agent = AtomicAgent[Input, Output](config=AgentConfig(
client=client,
model="gpt-5-mini",
history=shared_history, # Same history
system_prompt_generator=SystemPromptGenerator(
background=["You are a technical expert."]
),
))
# Agent B - Communication Expert
communication_agent = AtomicAgent[Input, Output](config=AgentConfig(
client=client,
model="gpt-5-mini",
history=shared_history, # Same history!
system_prompt_generator=SystemPromptGenerator(
background=["You simplify technical explanations."]
),
))
# Conversation flow
user_input = Input(text="Explain quantum computing")
# Technical agent adds to shared history
technical_response = technical_agent.run(user_input)
# Communication agent sees technical response in history
simple_response = communication_agent.run(
Input(text="Simplify the above explanation for a child")
)
```
### Pattern 2: Independent Histories
Each agent maintains its own isolated history.
```{mermaid}
flowchart TB
subgraph Agent_A["Agent A"]
HA[History A]
end
subgraph Agent_B["Agent B"]
HB[History B]
end
subgraph Agent_C["Agent C"]
HC[History C]
end
User --> Agent_A
User --> Agent_B
User --> Agent_C
```
**Use Case:** Parallel processing, independent tasks, privacy isolation.
```python
# Each agent has its own history
agent_a = AtomicAgent[Input, Output](config=AgentConfig(
client=client,
model="gpt-5-mini",
history=ChatHistory(), # Independent
))
agent_b = AtomicAgent[Input, Output](config=AgentConfig(
client=client,
model="gpt-5-mini",
history=ChatHistory(), # Independent
))
# They don't see each other's conversations
response_a = agent_a.run(Input(text="Research topic A"))
response_b = agent_b.run(Input(text="Research topic B"))
```
(pattern-3-agent-to-agent-messaging)=
### Pattern 3: Agent-to-Agent Messaging
Manually transfer outputs between agent memories. **This directly addresses Issue #58.**
```{mermaid}
sequenceDiagram
participant U as User
participant O as Orchestrator
participant A as Agent A
participant B as Agent B
U->>O: Initial request
O->>A: run(user_input)
Note over A: Turn 1: User + Response
added to A.history
A-->>O: Result A
O->>O: Manual transfer
Note over O: B.history.add_message(
"user", Result A)
O->>B: run(None)
Note over B: Uses existing history
Turn 2: Response added
B-->>O: Result B
O->>O: Manual transfer
Note over O: A.history.add_message(
"user", Result B)
O->>A: run(None)
Note over A: Continues with
B's feedback in context
A-->>O: Final Result
```
**Use Case:** Agent loops, evaluation cycles, iterative refinement.
```
/* Detailed source-code truncated for AI context efficiency. */
```
### Pattern 4: Supervisor-Worker with Context Providers
Use context providers to share state between supervisor and worker agents.
```{mermaid}
flowchart TB
subgraph SharedContext["Shared Context Provider"]
SC[Task State & Results]
end
SUP[Supervisor Agent] --> SharedContext
W1[Worker 1] --> SharedContext
W2[Worker 2] --> SharedContext
W3[Worker 3] --> SharedContext
SUP -->|Delegates| W1
SUP -->|Delegates| W2
SUP -->|Delegates| W3
W1 -->|Updates| SharedContext
W2 -->|Updates| SharedContext
W3 -->|Updates| SharedContext
```
```
/* Detailed source-code truncated for AI context efficiency. */
```
### Pattern 5: Memory-Augmented Loops
Combine conversation history with external memory for long-running processes.
```
/* Detailed source-code truncated for AI context efficiency. */
```
---
## Best Practices
### When to Use Each Pattern
| Scenario | Recommended Pattern |
|----------|-------------------|
| Single agent chatbot | Basic ChatHistory |
| Multi-turn with context | ChatHistory + Context Providers |
| Parallel independent tasks | Independent Histories |
| Sequential pipeline | Agent-to-Agent Messaging |
| Iterative refinement loops | Agent-to-Agent Messaging |
| Supervisor-worker | Shared Context Providers |
| Long-running processes | Memory-Augmented Loops |
### Managing Context Window Limits
```python
from atomic_agents.utils import get_context_token_count
# Monitor token usage
token_info = agent.get_context_token_count()
print(f"Total tokens: {token_info.total}")
print(f"System prompt: {token_info.system_prompt}")
print(f"History: {token_info.history}")
print(f"Utilization: {token_info.utilization:.1%}")
# Set appropriate limits
if token_info.utilization > 0.8:
# Consider trimming history or summarizing
pass
```
### Testing Agents with Memory
```python
import pytest
from atomic_agents.context import ChatHistory
@pytest.fixture
def fresh_history():
"""Provide clean history for each test."""
return ChatHistory()
@pytest.fixture
def agent_with_history(fresh_history):
"""Agent with clean history."""
return AtomicAgent[Input, Output](config=AgentConfig(
client=mock_client,
model="gpt-5-mini",
history=fresh_history,
))
def test_conversation_continuity(agent_with_history):
"""Test that agent remembers previous messages."""
agent_with_history.run(Input(text="My name is Bob"))
response = agent_with_history.run(Input(text="What's my name?"))
assert "Bob" in response.response
def test_history_persistence(agent_with_history):
"""Test serialization/deserialization."""
agent_with_history.run(Input(text="Remember: secret=42"))
# Serialize
serialized = agent_with_history.history.dump()
# Create new history and load
new_history = ChatHistory()
new_history.load(serialized)
assert new_history.get_message_count() == 2
```
### Debugging Memory Issues
```python
# Inspect current history
for msg in history.history:
print(f"[{msg.role}] Turn: {msg.turn_id}")
print(f" Content: {msg.content.model_dump_json()[:100]}...")
print()
# Check turn state
print(f"Current turn ID: {history.get_current_turn_id()}")
print(f"Message count: {history.get_message_count()}")
print(f"Max messages: {history.max_messages}")
```
---
## Troubleshooting
### "Messages aren't being added to history"
**Cause:** Calling `run()` without input after resetting history.
```python
# Wrong - no messages to work with
agent.reset_history()
agent.run() # Nothing in history!
# Correct
agent.reset_history()
agent.run(Input(text="Start fresh")) # Provides input
```
### "Agent doesn't remember previous conversation"
**Cause:** Creating new agent instances instead of reusing.
```python
# Wrong - new agent = new history each time
def handle_message(text):
agent = AtomicAgent[Input, Output](config=config) # New instance!
return agent.run(Input(text=text))
# Correct - reuse agent instance
agent = AtomicAgent[Input, Output](config=config) # Create once
def handle_message(text):
return agent.run(Input(text=text)) # Reuse
```
### "How do I pass memory between agents?"
See [Pattern 3: Agent-to-Agent Messaging](#pattern-3-agent-to-agent-messaging).
```python
# Transfer output to another agent's memory
agent_b.history.add_message("user", agent_a_output)
agent_b.run() # Now has context from agent A
```
### "What exactly is a 'turn'?"
A **turn** is a logical unit of conversation, typically containing:
- One user message
- One assistant response
- Both sharing the same `turn_id` (UUID)
```python
# This is ONE turn:
response = agent.run(Input(text="Hello"))
# turn_id "abc-123" assigned to both user message and response
# This starts a NEW turn:
response2 = agent.run(Input(text="Next question"))
# turn_id "def-456" assigned to new pair
```
### "History is too large / context overflow"
```python
# Option 1: Limit history size
history = ChatHistory(max_messages=30)
# Option 2: Monitor and handle
if history.get_message_count() > 40:
# Summarize or archive old messages
pass
# Option 3: Use context providers for persistent data
# instead of relying on conversation history
```
---
## API Quick Reference
### ChatHistory
| Method | Description |
|--------|-------------|
| `ChatHistory(max_messages=None)` | Create history with optional limit |
| `add_message(role, content)` | Add message to current turn |
| `initialize_turn()` | Start new turn with new UUID |
| `get_current_turn_id()` | Get current turn's UUID |
| `get_history()` | Get all messages as list of dicts |
| `get_message_count()` | Get number of messages |
| `delete_turn_id(turn_id)` | Delete all messages in a turn |
| `dump()` | Serialize to JSON string |
| `load(data)` | Deserialize from JSON string |
| `copy()` | Create deep copy |
### BaseChatHistory
The abstract base class `ChatHistory` implements. Subclass it (directly, or via `ChatHistory`) to plug in a custom memory backend. See [Writing a Custom Memory Backend](#writing-a-custom-memory-backend).
| Method | Description |
|--------|-------------|
| `initialize_turn()` | Start new turn with new UUID |
| `add_message(role, content)` | Add message to current turn |
| `get_history()` | Get all messages as list of dicts |
| `get_current_turn_id()` | Get current turn's UUID |
| `delete_turn_id(turn_id)` | Delete all messages in a turn |
| `get_message_count()` | Get number of messages |
| `dump()` | Serialize to JSON string |
| `load(data)` | Deserialize from JSON string |
| `copy()` | Create deep copy |
Every implementation must also maintain a `history: List[Message]` attribute and a `current_turn_id: Optional[str]` attribute.
### Message
| Field | Type | Description |
|-------|------|-------------|
| `role` | str | "user", "assistant", or "system" |
| `content` | BaseIOSchema | Message content |
| `turn_id` | Optional[str] | UUID linking related messages |
### BaseDynamicContextProvider
| Method | Description |
|--------|-------------|
| `__init__(title)` | Create with display title |
| `get_info() -> str` | Return context string (override this) |
---
## Next Steps
- [Quickstart Guide](quickstart.md) - Get started with Atomic Agents
- [Tools Guide](tools.md) - Add capabilities to your agents
- [Orchestration Guide](orchestration.md) - Coordinate multiple agents
- [Hooks Guide](hooks.md) - Monitor and customize agent behavior
- [API Reference](/api/context) - Full API documentation
---
## Summary
Key takeaways:
1. **Automatic Memory**: `agent.run(input)` automatically manages history - you don't need to manually add messages
2. **Turns**: A turn groups user input + assistant response with a shared UUID
3. **Persistence**: Use `dump()`/`load()` to save and restore conversations
4. **Context Providers**: Inject dynamic information (RAG, user data, time) into system prompts
5. **Multi-Agent**: Use shared history, agent-to-agent messaging, or context providers depending on your needs
For questions or issues, visit our [GitHub repository](https://github.com/eigenwise/atomic-agents) or [Reddit community](https://www.reddit.com/r/AtomicAgents/).
---
### Guides/Orchestration
# Orchestration and Multi-Agent Patterns
This guide covers patterns for building multi-agent systems and orchestrating complex workflows with Atomic Agents.
## Overview
Orchestration in Atomic Agents enables:
- **Tool Selection**: Agents that choose appropriate tools based on input
- **Multi-Agent Pipelines**: Chain agents for complex workflows
- **Dynamic Routing**: Route queries to specialized agents
- **Parallel Execution**: Run multiple agents concurrently
- **Agent Composition**: Combine agents for sophisticated behavior
## Tool Orchestration Pattern
The most common pattern: an orchestrator agent that selects and invokes tools.
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Sequential Pipeline Pattern
Chain multiple agents where each agent's output feeds the next:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Parallel Execution Pattern
Run multiple agents concurrently for independent tasks:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Router Pattern
Route queries to specialized agents based on classification:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Context Sharing Between Agents
Share information between agents using context providers:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Supervisor Pattern
A supervisor agent that manages and validates worker agents:
```
/* Detailed source-code truncated for AI context efficiency. */
```
## Best Practices
### 1. Design Clear Interfaces
Define explicit input/output schemas for each agent:
```python
# Good: Clear, typed interfaces
class AgentAOutput(BaseIOSchema):
data: str
metadata: dict
class AgentBInput(BaseIOSchema):
data: str # Explicitly matches AgentAOutput.data
```
### 2. Handle Failures Gracefully
Implement fallbacks and error handling:
```python
def execute_with_fallback(primary_agent, fallback_agent, input_data):
try:
return primary_agent.run(input_data)
except Exception as e:
print(f"Primary failed: {e}, using fallback")
return fallback_agent.run(input_data)
```
### 3. Monitor Agent Interactions
Log inter-agent communication:
```python
def logged_handoff(from_agent: str, to_agent: str, data):
print(f"[{from_agent}] -> [{to_agent}]: {type(data).__name__}")
return data
```
### 4. Keep Agents Focused
Each agent should have a single responsibility:
```python
# Good: Single responsibility
query_generator = AtomicAgent[...] # Only generates queries
analyzer = AtomicAgent[...] # Only analyzes
# Avoid: Multiple responsibilities in one agent
do_everything_agent = AtomicAgent[...] # Too complex
```
## Summary
| Pattern | Use Case | Key Benefit |
|---------|----------|-------------|
| Tool Orchestration | Dynamic tool selection | Flexible routing |
| Sequential Pipeline | Multi-step processing | Clear data flow |
| Parallel Execution | Independent analyses | Performance |
| Router Pattern | Query classification | Specialization |
| Context Sharing | Knowledge accumulation | Collaboration |
| Supervisor Pattern | Quality assurance | Validation |
Choose patterns based on your workflow requirements and combine them for sophisticated agent systems.
---