Index

Welcome to Atomic Agents Documentation

``{toctree} :maxdepth: 2 :caption: Documentation

guides/index api/index examples/index contributing

text

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.

text

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 and leverages the power of Pydantic 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
text
Or using uv (recommended):
bash
uv add atomic-agents
text
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
text
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.
text
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        history=history
    " target="_blank" rel="noopener noreferrer">BasicChatInputSchema, BasicChatOutputSchema
)
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)
text
Example Projects 
Check out our example projects in our GitHub repository:
- Quickstart Examples: Simple examples to get started
- Hooks System: Comprehensive monitoring, error handling, and performance metrics
- Basic Multimodal: Analyze images with text
- RAG Chatbot: Build context-aware chatbots
- Web Search Agent: Create agents that perform web searches
- Deep Research: Perform deep research tasks
- YouTube Summarizer: Extract knowledge from videos
- YouTube to Recipe: Convert cooking videos into structured recipes
- Orchestration Agent: Coordinate multiple agents for complex tasks
Community & Support 
- GitHub Repository
- Issue Tracker
- Reddit Community
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
text
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

text

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)

text

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)

text

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
    )
text
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 = AtomicAgentBasicChatInputSchema, BasicChatOutputSchema),
        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)
text
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
text
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."
    )
text
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")
)
text
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)
text
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)
text
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%}")
text
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 = AtomicAgentCustomInput, CustomOutput
)
text
For full API details:
{eval-rst}
.. automodule:: atomic_agents.agents.atomic_agent
   :members:
   :undoc-members:
   :show-inheritance:
text
---
Api/Context 
Context
{seealso}
For a comprehensive guide on memory management, multi-agent patterns, and best practices, see the Memory and Context Guide.
text
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 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()
text
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
text
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
text
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()
text
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()
text
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")
    }
)
text
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")
text
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
text
Key features:
- Structured input/output schemas
- Configuration management
- Title and description overrides
- Error handlingFor 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:
text
---
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
text
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
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
Utils 
The utils module provides helper functions and utilities:
- Message formatting
- Tool response handling
- Schema validation
- Error handling
Learn more about utilities
Getting Started 
For practical examples and guides on using these components, see:
- Quickstart Guide
- Tools Guide
---
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.
text
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
text
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%}")
text
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%}")
text
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:
text
---
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
text
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 - 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 - 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 - 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 - 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 - 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 - 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 - 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 - 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 - 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
📂 View on GitHub - 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)
text
2. STDIO Transport (server_stdio.py):
- Runs as a subprocess
- Communicates through standard input/output
- Ideal for local development3. 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")
}
text
2. Query Processing:
- Uses an orchestrator agent to analyze queries
- Extracts parameters and selects appropriate tools
- Maintains conversation context through ChatHistory3. 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
text
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
text
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
text
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(),
        )
text
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,
)
text
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
text
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
text
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
text
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
text
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")
text
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"
text
Usage Examples 
1. Start the Server (SSE mode):
bash
cd example-mcp-server
uv run example-mcp-server --mode=sse
text
2. Run the Client:Using the main launcher with transport selection:
bash
cd example-client
uv run python -m example_client.main --transport sse
text
Directly calling the SSE client:
bash
cd example-client
uv run python -m example_client.main_sse
text
Directly calling the STDIO client:
bash
cd example-client
uv run python -m example_client.main_stdio
text
3. Example Queries:

You: What is 2+2?
You: Calculate the square root of 144
You: Generate a random number between 1 and 100
text
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 toolExample 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)
text
Then register the tool in the server:
python
def get_available_tools() -> List[Tool]:
    return [
        # ... existing tools ...
        MyNewTool(),
    ]
text
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 URIExample 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)
text
Then register the resource in the server:
python
def get_available_resources() -> List[Resource]:
    return [
        # ... existing resources ...
        MyNewResource(),
    ]
text
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 promptExample 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)
text
Then register the prompt in the server:
python
def get_available_prompts() -> List[Prompt]:
    return [
        # ... existing prompts ...
        MyNewPrompt(),
    ]
text
---
Guides/Cookbook 
Cookbook
Practical recipes for common Atomic Agents use cases.
Quick Reference 
| Recipe | Description |
|--------|-------------|
| Basic Chatbot | Simple conversational agent |
| Chatbot with Memory | Agent that remembers context |
| Custom Output Schema | Structured responses |
| Multi-Provider Agent | Switch between LLM providers |
| Agent with Tools | Agent using external tools |
| Streaming Chatbot | Real-time response streaming |
| Research Agent | Multi-step research workflow |
| 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            history=ChatHistory(" target="_blank" rel="noopener noreferrer">BasicChatInputSchema, BasicChatOutputSchema
        )
    )
    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)
text
Chatbot with Memory 
Agent that maintains conversation history across turns.

/ Detailed source-code truncated for AI context efficiency. /
text
Custom Output Schema 
Agent with structured output including metadata.

/ Detailed source-code truncated for AI context efficiency. /
text
Multi-Provider Agent 
Switch between different LLM providers dynamically.

/ Detailed source-code truncated for AI context efficiency. /
text
Agent with Tools 
Agent that uses tools to extend capabilities.

/ Detailed source-code truncated for AI context efficiency. /
text
Streaming Chatbot 
Real-time streaming responses.

/ Detailed source-code truncated for AI context efficiency. /
text
Research Agent 
Multi-step research workflow.

/ Detailed source-code truncated for AI context efficiency. /
text
RAG Agent 
Retrieval-augmented generation pattern.

/ Detailed source-code truncated for AI context efficiency. /
text
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()
text
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

    }

}

text
Creating Production-Ready Agents 
Agent Factory Pattern 
Create agents with production configuration:

/ Detailed source-code truncated for AI context efficiency. /
text
FastAPI Integration 
Deploy agents as REST APIs:

/ Detailed source-code truncated for AI context efficiency. /
text
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"]
text
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"
text
Rate Limiting 
Implement rate limiting to control API costs:

/ Detailed source-code truncated for AI context efficiency. /
text
Graceful Shutdown 
Handle shutdown signals properly:

/ Detailed source-code truncated for AI context efficiency. /
text
Health Checks 
Implement comprehensive health checks:

/ Detailed source-code truncated for AI context efficiency. /
text
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        history=ChatHistory(" target="_blank" rel="noopener noreferrer">ValidatedInputSchema, ValidatedOutputSchema
    )
)
Handle validation errors

try:
    response = agent.run(ValidatedInputSchema(query="", max_results=5))
except ValueError as e:
    print(f"Validation error: {e}")
text
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
text
API Error Handling 
Handle LLM provider errors gracefully with retry logic.
Basic Retry Pattern 

/ Detailed source-code truncated for AI context efficiency. /
text
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        history=ChatHistory(" target="_blank" rel="noopener noreferrer">BasicChatInputSchema, BasicChatOutputSchema
    )
)
text
Comprehensive Error Handler 

/ Detailed source-code truncated for AI context efficiency. /
text
Graceful Degradation 
Implement fallback behavior when the primary agent fails.
Fallback Agent Pattern 

/ Detailed source-code truncated for AI context efficiency. /
text
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()
text
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
text
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
        }
    )
)
text
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)
text
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
text
Or using uv (recommended):
bash
uv add atomic-agents
text
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]
text
What Python version is required? 
Atomic Agents requires Python 3.12 or higher.
bash
Check your Python version

python --version
text
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

text
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")
text
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        history=ChatHistory(" target="_blank" rel="noopener noreferrer">BasicChatInputSchema, BasicChatOutputSchema
    )
)
Use the agent

response = agent.run(BasicChatInputSchema(chat_message="Hello!"))
print(response.chat_message)
text
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())
text
Anthropic:
python
import instructor
from anthropic import Anthropic
client = instructor.from_anthropic(Anthropic())
text
Groq:
python
import instructor
from groq import Groq
client = instructor.from_groq(Groq(), mode=instructor.Mode.JSON)
text
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
)
text
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
)
text
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        system_prompt_generator=system_prompt
    " target="_blank" rel="noopener noreferrer">BasicChatInputSchema, BasicChatOutputSchema
)
text
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        history=history
    " target="_blank" rel="noopener noreferrer">BasicChatInputSchema, BasicChatOutputSchema
)
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()
text
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 = AtomicAgentCustomInputSchema, CustomOutputSchema
)
response = agent.run(CustomInputSchema(question="What is Python?"))
print(f"Answer: {response.answer}")
print(f"Confidence: {response.confidence}")
text
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
text
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")
text
How do I use the built-in tools? 
Use the Atomic Assembler CLI to download tools:
bash
atomic
text
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
text
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
text
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            history=ChatHistory(" target="_blank" rel="noopener noreferrer">BasicChatInputSchema, BasicChatOutputSchema
        )
    )
    # 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())
text
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"))
text
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']}")
text
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
text
How do I debug agent behavior? 
1. Check the system prompt:
python
print(agent.system_prompt_generator.generate_prompt())
text
2. Inspect history:
python
for msg in agent.history.get_history():
    print(f"{msg['role']}: {msg['content']}")
text
3. Enable logging:
python
import logging
logging.basicConfig(level=logging.DEBUG)
text
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())
text
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
text
2. Class names:
python
Old

from atomic_agents.agents.base_agent import BaseAgent, BaseAgentConfig
New

from atomic_agents import AtomicAgent, AgentConfig
text
3. Schemas as type parameters:
python
Old

agent = BaseAgent(BaseAgentConfig(
    client=client,
    model="gpt-5-mini",
    input_schema=MyInput,
    output_schema=MyOutput
))
New

agent = AtomicAgentMyInput, MyOutput
)
text
See the Upgrade Guide 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        history=ChatHistory(" target="_blank" rel="noopener noreferrer">BasicChatInputSchema, BasicChatOutputSchema
    )
)
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!"))
text
Performance Monitoring 
Track request metrics for performance analysis:

/ Detailed source-code truncated for AI context efficiency. /
text
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)
text
Retry Strategies with Hooks 
Implement intelligent retry logic based on error context:

/ Detailed source-code truncated for AI context efficiency. /
text
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")
text
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()
text
Production Logging Pattern 
A complete production-ready logging setup:

/ Detailed source-code truncated for AI context efficiency. /
text
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)
text
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)
text
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
text
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
text
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
2. Join our Reddit community
3. Read through our examples in the repository
4. Review the example projects in atomic-examples/
See also:
- API Reference - Browse the API reference
- Main Documentation - 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. /
text
Agent Logging with Hooks 
Comprehensive Request Logging 
Use hooks to log all agent interactions:

/ Detailed source-code truncated for AI context efficiency. /
text
Metrics Collection 
Token and Cost Tracking 
Track API usage and costs:

/ Detailed source-code truncated for AI context efficiency. /
text
Monitoring Dashboard 
FastAPI Metrics Endpoint 
Expose metrics via HTTP:

/ Detailed source-code truncated for AI context efficiency. /
text
Distributed Tracing 
OpenTelemetry Integration 
Add distributed tracing for complex systems:

/ Detailed source-code truncated for AI context efficiency. /
text
Alerting 
Alert Conditions 
Define alert conditions for monitoring:

/ Detailed source-code truncated for AI context efficiency. /
text
Log Analysis Patterns 
Structured Log Queries 
Design logs for easy querying:

/ Detailed source-code truncated for AI context efficiency. /
text
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
text
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)
- Understanding of Python classes and async/await
The Problem This Solves 
A common question from developers (see GitHub Issue #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
text
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
)
text
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)
text
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        history=history,
    " target="_blank" rel="noopener noreferrer">ChatInput, ChatOutput
)
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!

text
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
text
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

text
---
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<br/>provided?}
    B -->|Yes| C["history.initialize_turn()<br/>Creates new UUID"]
    C --> D["history.add_message('user', user_input)<br/>Stores user message"]
    B -->|No| E["Skip turn initialization<br/>Use existing history"]
    D --> F["_prepare_messages()<br/>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)<br/>Stores response"]
    J --> K["_manage_overflow()<br/>Trim if needed"]
    K --> L["Return response"]
    style C fill:#e1f5fe
    style D fill:#e1f5fe
    style J fill:#e1f5fe
text
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    history=history
" target="_blank" rel="noopener noreferrer">Input, Output)
--- 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"
text
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
text
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

text
---
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)
text
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    history=history,  # Restored history!
" target="_blank" rel="noopener noreferrer">Input, Output)
Continue the conversation where it left off

response = agent.run(Input(text="Where were we?"))
text
{warning}
Only load serialized data from trusted sources. The

load()

 method reconstructs Python classes from the serialized data.
text
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

text
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
text
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"))
text
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

text
Resetting History:
python
Clear all messages, start fresh

agent.reset_history()
or

history = ChatHistory()  # Create new instance
text
---
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
text
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.
text
{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.
text
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.
---
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)
text
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"]
text
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),
)
text
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.
text
python
Serialize (file paths are preserved)

serialized = history.dump()
When loading, files must be accessible at original paths

history.load(serialized)
text
---
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
text
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
text
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    system_prompt_generator=system_prompt,
" target="_blank" rel="noopener noreferrer">Input, Output)
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?"))
text
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))
text
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')}"
text
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())
text
---
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
text
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    history=shared_history,  # Same history
    system_prompt_generator=SystemPromptGenerator(
        background=["You are a technical expert."]
    " target="_blank" rel="noopener noreferrer">Input, Output,
))
Agent B - Communication Expert

communication_agent = AtomicAgent    history=shared_history,  # Same history!
    system_prompt_generator=SystemPromptGenerator(
        background=["You simplify technical explanations."]
    " target="_blank" rel="noopener noreferrer">Input, Output,
))
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")
)
text
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
text
Use Case: Parallel processing, independent tasks, privacy isolation.
python
Each agent has its own history

agent_a = AtomicAgent    history=ChatHistory(" target="_blank" rel="noopener noreferrer">Input, Output,  # Independent
))
agent_b = AtomicAgent    history=ChatHistory(" target="_blank" rel="noopener noreferrer">Input, Output,  # 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"))
text
(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<br/>added to A.history
    A-->>O: Result A
    O->>O: Manual transfer
    Note over O: B.history.add_message(<br/>"user", Result A)
    O->>B: run(None)
    Note over B: Uses existing history<br/>Turn 2: Response added
    B-->>O: Result B
    O->>O: Manual transfer
    Note over O: A.history.add_message(<br/>"user", Result B)
    O->>A: run(None)
    Note over A: Continues with<br/>B's feedback in context
    A-->>O: Final Result
text
Use Case: Agent loops, evaluation cycles, iterative refinement.

/ Detailed source-code truncated for AI context efficiency. /
text
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
text

/ Detailed source-code truncated for AI context efficiency. /
text
Pattern 5: Memory-Augmented Loops 
Combine conversation history with external memory for long-running processes.

/ Detailed source-code truncated for AI context efficiency. /
text
---
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
text
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        history=fresh_history,
    " target="_blank" rel="noopener noreferrer">Input, Output)
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
text
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}")
text
---
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
text
"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 = AtomicAgentInput, Output  # New instance!
    return agent.run(Input(text=text))
Correct - reuse agent instance

agent = AtomicAgentInput, Output  # Create once
def handle_message(text):
    return agent.run(Input(text=text))  # Reuse
text
"How do I pass memory between agents?" 
See 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
text
"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

text
"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

text
---
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.
| 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 - Get started with Atomic Agents
- Tools Guide - Add capabilities to your agents
- Orchestration Guide - Coordinate multiple agents
- Hooks Guide - Monitor and customize agent behavior
- API Reference - 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 or Reddit community.
---
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. /
text
Sequential Pipeline Pattern 
Chain multiple agents where each agent's output feeds the next:

/ Detailed source-code truncated for AI context efficiency. /
text
Parallel Execution Pattern 
Run multiple agents concurrently for independent tasks:

/ Detailed source-code truncated for AI context efficiency. /
text
Router Pattern 
Route queries to specialized agents based on classification:

/ Detailed source-code truncated for AI context efficiency. /
text
Context Sharing Between Agents 
Share information between agents using context providers:

/ Detailed source-code truncated for AI context efficiency. /
text
Supervisor Pattern 
A supervisor agent that manages and validates worker agents:

/ Detailed source-code truncated for AI context efficiency. /
text
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
text
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)
text
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
text
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.

---

Index Copy Section

Welcome to Atomic Agents Documentation

A Lightweight and Modular Framework for Building AI Agents

Key Features Copy Chapter

Installation Copy Chapter

Quick Example Copy Chapter

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

Set up client with your preferred provider

Create an agent

Interact with your agent (using the agent's input schema)

Or more explicitly:

Example Projects Copy Chapter

Community & Support Copy Chapter

Indices and References Copy Chapter

Api/Agents Copy Section

Agents

Schema Hierarchy Copy Chapter

BaseIOSchema Copy Section

BasicChatInputSchema Copy Section

BasicChatOutputSchema Copy Section

Creating Custom Schemas Copy Section

Base Agent Copy Chapter

Create agent with basic configuration

Run the agent

Stream responses

Configuration Copy Section

Input/Output Schemas Copy Section

Key Methods Copy Section

Context Providers Copy Section

Register with agent

Streaming Support Copy Section

History Management Copy Section

Access history

Reset to initial state

Save/load history state

Token Counting Copy Section

Get accurate token count at any time - always returns a result

Check context utilization if max tokens is known

Custom Schemas Copy Section

Create agent with custom schemas

Api/Context Copy Section

Context

Agent History Copy Chapter

Initialize history with optional max messages

Add messages

Initialize a new turn

Access history

Manage history

Persistence

Create copy

Message Structure Copy Section

Multimodal Support Copy Section

For content with images

System Prompt Generator Copy Chapter

Create generator with static content

Generate prompt

Custom System Prompt Generator Copy Section

Dynamic Context Providers Copy Section

Use with generator

Base Components Copy Chapter

BaseIOSchema Copy Section

BaseTool Copy Section

Api/Index Copy Section

API Reference

Core Components Copy Chapter

Agents Copy Section

Context Components Copy Section

Utils Copy Section

Getting Started Copy Chapter

Api/Utils Copy Section

Utilities

Token Counting Copy Chapter

TokenCountResult Copy Section

TokenCounter Copy Section

Usage Example Copy Section

Direct usage

Count tokens in messages

Index

Key Features

Installation

Quick Example

Example Projects

Community & Support

Indices and References

`Api/Agents`

`Agents`

`Schema Hierarchy`

BaseIOSchema

BasicChatInputSchema

BasicChatOutputSchema

Creating Custom Schemas

Base Agent

Configuration

Input/Output Schemas

Key Methods

`Context Providers`

Streaming Support

History Management

Token Counting

Custom Schemas

Api/Context

Agent History

Message Structure

Multimodal Support

System Prompt Generator

Custom System Prompt Generator

Dynamic Context Providers

Base Components

BaseIOSchema

BaseTool

Api/Index

Core Components

Agents

`Context Components`

`Utils`

`Getting Started`

`Api/Utils`

`Utilities`

`Token Counting`

`TokenCountResult`

TokenCounter

Usage Example

Using with AtomicAgent

Tool Message Formatting

Examples/Index

Quickstart Examples

Hooks System

Basic Multimodal

RAG Chatbot

Web Search Agent

Deep Research

YouTube Summarizer

YouTube to Recipe

Orchestration Agent

MCP Agent

Examples/Mcp Agent

Overview

Architecture

MCP Server

`MCP Client`

`Implementation Details`

`Server Implementation`

`Client Implementation`

MCP Transport Methods

1. STDIO Transport

2. SSE Transport

3. HTTP Stream Transport

Interfaces

Tool Interface

`Resource Interface`

`Prompt Interface`

`Configuration`

`Server Configuration`

Client Configuration

Usage Examples

Best Practices

Extending the Example