omnicoreagent 0.3.6

OmniCoreAgent is a powerful Python AI Agent framework for building autonomous AI agents that think, reason, and execute complex tasks. Production-ready agents that use tools, manage memory, coordinate workflows, and handle real-world business logic.

🚀 OmniCoreAgent

The AI Agent Framework Built for Production
Switch memory backends at runtime. Manage context automatically. Deploy with confidence.

Quick Start • See It In Action • 📚 Cookbook • Features • Docs

---

🎬 See It In Action

import asyncio
from omnicoreagent import OmniCoreAgent, MemoryRouter, ToolRegistry

# Create tools in seconds
tools = ToolRegistry()

@tools.register_tool("get_weather")
def get_weather(city: str) -> dict:
    """Get current weather for a city."""
    return {"city": city, "temp": "22°C", "condition": "Sunny"}

# Build a production-ready agent
agent = OmniCoreAgent(
    name="assistant",
    system_instruction="You are a helpful assistant with access to weather data.",
    model_config={"provider": "openai", "model": "gpt-4o"},
    local_tools=tools,
    memory_router=MemoryRouter("redis"),  # Start with Redis
    agent_config={
        "context_management": {"enabled": True},  # Auto-manage long conversations
        "guardrail_config": {"strict_mode": True},  # Block prompt injections
    }
)

async def main():
    # Run the agent
    result = await agent.run("What's the weather in Tokyo?")
    print(result["response"])
    
    # Switch to MongoDB at runtime — no restart needed
    await agent.switch_memory_store("mongodb")
    
    # Keep running with a different backend
    result = await agent.run("How about Paris?")
    print(result["response"])

asyncio.run(main())

What just happened?

• ✅ Registered a custom tool with type hints
• ✅ Built an agent with memory persistence
• ✅ Enabled automatic context management
• ✅ Switched from Redis to MongoDB while running

---

⚡ Quick Start

pip install omnicoreagent

echo "LLM_API_KEY=your_api_key" > .env

from omnicoreagent import OmniCoreAgent

agent = OmniCoreAgent(
    name="my_agent",
    system_instruction="You are a helpful assistant.",
    model_config={"provider": "openai", "model": "gpt-4o"}
)

result = await agent.run("Hello!")
print(result["response"])

That's it. You have an AI agent with session management, memory, and error handling.

📚 Want to learn more? Check out the Cookbook — progressive examples from "Hello World" to production deployments.

---

🎯 What Makes OmniCoreAgent Different?

Feature	What It Means For You
Runtime Backend Switching	Switch Redis ↔ MongoDB ↔ PostgreSQL without restarting
Cloud Workspace Storage	Agent files persist in AWS S3 or Cloudflare R2 ⚡ NEW
Context Engineering	Session memory + agent loop context + tool offloading = no token exhaustion
Tool Response Offloading	Large tool outputs saved to files, 98% token savings
Built-in Guardrails	Prompt injection protection out of the box
MCP Native	Connect to any MCP server (stdio, SSE, HTTP with OAuth)
Background Agents	Schedule autonomous tasks that run on intervals
Workflow Orchestration	Sequential, Parallel, and Router agents for complex tasks
Production Observability	Metrics, tracing, and event streaming built in

---

📋 Table of Contents

Getting Started: See It In Action • Quick Start

Core Features: OmniCoreAgent • Memory System • Context Engineering • Event System • MCP Client • Deep Agent • Local Tools • Agent Skills • Workspace Memory (S3/R2/Local)

Multi-Agent: Sub-Agents • Background Agents • Workflows

Production: BM25 Tool Retrieval • Observability • Guardrails • Model Support • OmniServe

Reference: Examples • Configuration • Testing • Contributing

---

🏗 Architecture Overview

OmniCoreAgent follows a layered architecture that separates concerns while enabling rich integrations between components. Here's how it all fits together:

flowchart TB
    %% Entry Points
    subgraph Entry["🚀 Entry Points"]
        User([👤 User / App])
        Clock([⏰ Scheduler])
        API([🌐 External API])
    end

    %% Security Layer - First Line of Defense
    subgraph Security["🛡️ Security Layer"]
        Guard["Guardrails Engine<br/>━━━━━━━━━━━━━━<br/>• Pattern Matching<br/>• Heuristic Analysis<br/>• Entropy Detection<br/>• Encoding Detection"]
    end

    %% Orchestration Layer
    subgraph Orchestration["🎼 Orchestration Layer"]
        direction LR
        WF["Workflow Engine"]
        SEQ["Sequential<br/>Agent"]
        PAR["Parallel<br/>Agent"]
        RTR["Router<br/>Agent"]
        BG["Background<br/>Agent"]

        WF --> SEQ
        WF --> PAR
        WF --> RTR
    end

    %% Core Engine - The Heart
    subgraph Core["🧠 OmniCore Engine"]
        direction TB
        OCA["OmniCoreAgent<br/>━━━━━━━━━━━━━━<br/>• System Instructions<br/>• Model Config<br/>• Agent Config"]
        
        subgraph Processing["Processing Pipeline"]
            direction LR
            CTX["Context<br/>Manager"]
            SUM["Conversation<br/>Summarizer"]
            REACT["ReAct<br/>Loop"]
            OFF["Tool Response<br/>Offloader"]
        end
        
        LLM["LLM Layer<br/>━━━━━━━━━━━━━━<br/>OpenAI • Anthropic<br/>Gemini • Groq • Ollama<br/>Mistral • DeepSeek"]
    end

    %% Capabilities Layer
    subgraph Capabilities["🛠️ Capabilities Layer"]
        direction TB
        TOOLS["Tool Orchestration<br/>━━━━━━━━━━━━━━<br/>• Validation<br/>• Execution<br/>• Error Handling"]
        
        subgraph ToolTypes["Tool Types"]
            direction LR
            LOCAL["Local Tools<br/>(ToolRegistry)"]
            MCP["MCP Client<br/>(stdio/http/sse)"]
            BM25["BM25 RAG<br/>(Dynamic Discovery)"]
            SKILLS["Agent Skills<br/>(Polyglot Scripts)"]
            SUB["Sub-Agents<br/>(Delegation)"]
        end
    end

    %% Infrastructure Layer
    subgraph Infrastructure["💾 Infrastructure Layer"]
        direction TB
        
        subgraph Routers["Routers (Hot-Swappable)"]
            direction LR
            MEM["Memory Router<br/>━━━━━━━━━━━━━━<br/>Session State<br/>Conversation History"]
            EVT["Event Router<br/>━━━━━━━━━━━━━━<br/>Real-time Streaming<br/>Audit Trail"]
        end
        
        subgraph Storage["Storage Backends"]
            direction LR
            REDIS[("Redis")]
            POSTGRES[("PostgreSQL")]
            MONGO[("MongoDB")]
            MEM_STORE[("In-Memory")]
            FS[("File System")]
        end
        
        OBS["Observability<br/>━━━━━━━━━━━━━━<br/>Metrics • Tracing<br/>Token Usage • Opik"]
    end

    %% Connections - Main Flow
    User --> Guard
    API --> Guard
    Clock --> BG
    
    Guard -->|"✓ Safe"| OCA
    Guard -->|"✗ Blocked"| User
    
    BG --> OCA
    WF --> OCA
    SEQ --> OCA
    PAR --> OCA
    RTR --> OCA
    
    OCA --> CTX
    CTX <--> SUM
    CTX --> REACT
    REACT <--> LLM
    REACT --> TOOLS
    TOOLS --> OFF
    
    TOOLS --> LOCAL
    TOOLS --> MCP
    TOOLS --> BM25
    TOOLS --> SKILLS
    TOOLS --> SUB
    SUB -.->|"Recursive"| OCA
    
    %% Infrastructure Connections
    OCA <-.->|"State"| MEM
    OCA <-.->|"Events"| EVT
    OFF <-.->|"Artifacts"| FS
    
    MEM --> REDIS
    MEM --> POSTGRES
    MEM --> MONGO
    MEM --> MEM_STORE
    EVT --> REDIS
    
    OCA <-.->|"Metrics"| OBS

    %% Styling
    style Entry fill:#1abc9c,stroke:#16a085,color:#fff
    style Security fill:#e74c3c,stroke:#c0392b,color:#fff
    style Orchestration fill:#d35400,stroke:#e67e22,color:#fff
    style Core fill:#2c3e50,stroke:#34495e,color:#fff
    style Processing fill:#34495e,stroke:#2c3e50,color:#fff
    style Capabilities fill:#2980b9,stroke:#3498db,color:#fff
    style ToolTypes fill:#3498db,stroke:#2980b9,color:#fff
    style Infrastructure fill:#8e44ad,stroke:#9b59b6,color:#fff
    style Routers fill:#9b59b6,stroke:#8e44ad,color:#fff
    style Storage fill:#95a5a6,stroke:#7f8c8d,color:#fff

Layer Responsibilities

Layer	Purpose	Key Components
🚀 Entry	Request sources	User Apps, Schedulers, External APIs
🛡️ Security	Threat protection	Guardrails (injection detection, encoding checks)
🎼 Orchestration	Multi-agent coordination	Sequential, Parallel, Router, Background agents
🧠 Core Engine	Agent execution	ReAct loop, Context Management, Summarization, LLM calls
🛠️ Capabilities	Tool execution	Local tools, MCP, BM25 discovery, Skills, Sub-agents
💾 Infrastructure	Persistence & observability	Memory/Event routers, Storage backends, Metrics

Data Flow Highlights

1. Request Path: Entry → Security (guardrails) → Core → Capabilities → Response
2. Memory Persistence: State flows bidirectionally between agent and storage backends
3. Tool Offloading: Large responses saved to file system, only previews in context
4. Event Streaming: Real-time events pushed to Redis Streams for monitoring

🎯 Core Features

1. 🤖 OmniCoreAgent — The Heart of the Framework

from omnicoreagent import OmniCoreAgent, ToolRegistry, MemoryRouter, EventRouter

# Basic Agent
agent = OmniCoreAgent(
    name="assistant",
    system_instruction="You are a helpful assistant.",
    model_config={"provider": "openai", "model": "gpt-4o"}
)

# Production Agent with All Features
agent = OmniCoreAgent(
    name="production_agent",
    system_instruction="You are a production agent.",
    model_config={"provider": "openai", "model": "gpt-4o"},
    local_tools=tool_registry,
    mcp_tools=[...],
    memory_router=MemoryRouter("redis"),
    event_router=EventRouter("redis_stream"),
    agent_config={
        "max_steps": 20,
        "enable_advanced_tool_use": True,
        "enable_agent_skills": True,
        "memory_tool_backend": "local",
        # Memory with summarization
        "memory_config": {
            "mode": "sliding_window",
            "value": 10,
            "summary": {
                "enabled": True,
                "retention_policy": "summarize",
            },
        },
        # Context management for long conversations
        "context_management": {
            "enabled": True,
            "mode": "token_budget",
            "value": 100000,
            "threshold_percent": 75,
            "strategy": "summarize_and_truncate",
            "preserve_recent": 6,
        },
        # Prompt injection guardrails
        "guardrail_config": {
            "enabled": True,
            "strict_mode": True,
        },
    },
)


# Key Methods
await agent.run(query)                      # Execute task
await agent.run(query, session_id="user_1") # With session context
await agent.connect_mcp_servers()           # Connect MCP tools
await agent.list_all_available_tools()      # List all tools
await agent.switch_memory_store("mongodb")   # Switch backend at runtime!
await agent.get_session_history(session_id)      # Retrieve conversation history
await agent.clear_session_history(session_id)     # Clear history (session_id optional, clears all if None)
await agent.get_events(session_id)               # Get event history
await agent.get_memory_store_type()              # Get current memory router type
await agent.cleanup()                       # Clean up resources and remove the agent and the config
await agent.cleanup_mcp_servers()               # Clean up MCP servers without removing the agent and the config
await agent.get_metrics()                       # Get cumulative usage (tokens, requests, time)

[!TIP] Each agent.run() call now returns a metric field containing fine-grained usage for that specific request.

💡 When to Use: OmniCoreAgent is your go-to for any AI task — from simple Q&A to complex multi-step workflows. Start here for any agent project.

2. 🧠 Multi-Tier Memory System (Plug & Play)

5 backends with runtime switching — start with Redis, switch to MongoDB, then PostgreSQL — all on the fly!

from omnicoreagent import OmniCoreAgent, MemoryRouter

# Start with Redis
agent = OmniCoreAgent(
    name="my_agent",
    memory_router=MemoryRouter("redis"),
    model_config={"provider": "openai", "model": "gpt-4o"}
)

# Switch at runtime — no restart needed!
agent.swith_memory_store("mongodb")     # Switch to MongoDB
agent.swith_memory_store("database")    # Switch to PostgreSQL/MySQL/SQLite
agent.swith_memory_store("in_memory")   # Switch to in-memory
agent.swith_memory_store("redis")       # Back to Redis

Backend	Use Case	Environment Variable
in_memory	Fast development	—
redis	Production persistence	REDIS_URL
database	PostgreSQL/MySQL/SQLite	DATABASE_URL
mongodb	Document storage	MONGODB_URI

🧠 Conversation Summarization

OmniCoreAgent includes automatic conversation summarization to manage long conversation histories efficiently. When enabled, older messages are condensed into summaries, keeping context while reducing token usage.

from omnicoreagent import OmniCoreAgent, MemoryRouter

# Configure summarization with sliding window
memory_router = MemoryRouter(
    store_type="redis",
    memory_config={
        "mode": "sliding_window",  # or "token_budget"
        "value": 10,  # Keep last 10 messages (sliding_window) or max tokens (token_budget)
        "summary": {
            "enabled": True,
            "retention_policy": "keep"  # Options: "keep" or "delete"
        }
    }
)

agent = OmniCoreAgent(
    name="summarizing_agent",
    memory_router=memory_router,
    model_config={"provider": "openai", "model": "gpt-4o"}
)

Summarization Modes:

Mode	Description	Best For
sliding_window	Keep last N messages, summarize older ones	Predictable memory size
token_budget	Keep messages within token limit	Cost optimization

Retention Policies:

Policy	Behavior
keep	Mark summarized messages as inactive (recoverable)
delete	Permanently remove summarized messages

How It Works:

1. When conversation exceeds configured limit → summarization triggers
2. Older messages are sent to LLM for summary generation
3. Summary replaces older messages in active context
4. Original messages are retained (with "keep") or deleted per policy

💡 When to Use: Enable summarization for long-running conversations (support bots, research assistants) to maintain context while controlling costs. Use sliding_window for predictable behavior, token_budget for strict cost control.

3. 🔄 Context Engineering System

OmniCoreAgent implements state-of-the-art context engineering inspired by patterns from Anthropic and Cursor. This dual-layer approach ensures your agents never hit token limits — even during marathon coding sessions or multi-step research tasks.

flowchart TB
    subgraph Input["📥 Incoming Context"]
        MSG["Messages<br/>(User + Assistant + Tool)"]
        TOOL_RESP["Tool Responses<br/>(Web Search, APIs, Files)"]
    end

    subgraph Layer1["🧠 Layer 1: Agent Loop Context Management"]
        direction TB
        MONITOR["Context Monitor<br/>━━━━━━━━━━━━━━<br/>• Token counting<br/>• Message counting"]
        
        subgraph Modes["Management Modes"]
            direction LR
            TOKEN["token_budget<br/>━━━━━━━━━━<br/>Max total tokens"]
            SLIDE["sliding_window<br/>━━━━━━━━━━<br/>Max message count"]
        end
        
        subgraph Strategies["Overflow Strategies"]
            direction LR
            TRUNC["truncate<br/>━━━━━━━━<br/>Drop oldest<br/>(fast)"]
            SUMTRUNC["summarize_and_truncate<br/>━━━━━━━━━━━━━━<br/>Condense → Drop<br/>(preserves context)"]
        end
        
        RECENT["preserve_recent<br/>━━━━━━━━━━━━━━<br/>Always keep last N<br/>messages protected"]
    end

    subgraph Layer2["💾 Layer 2: Tool Response Offloading"]
        direction TB
        CHECK["Size Check<br/>━━━━━━━━━━━━━━<br/>threshold_tokens: 500<br/>threshold_bytes: 2000"]
        
        subgraph Offload["Offload Process"]
            direction LR
            SAVE["Save to File<br/>(.omnicoreagent_artifacts/)"]
            PREVIEW["Generate Preview<br/>(first ~150 tokens)"]
        end
        
        subgraph Tools["Built-in Artifact Tools"]
            direction LR
            READ["read_artifact()"]
            TAIL["tail_artifact()"]
            SEARCH["search_artifact()"]
            LIST["list_artifacts()"]
        end
    end

    subgraph Output["📤 Optimized Context"]
        CLEAN["Lean Context<br/>━━━━━━━━━━━━━━<br/>• System prompt<br/>• Recent messages<br/>• Summaries<br/>• Tool previews"]
    end

    %% Flow
    MSG --> MONITOR
    TOOL_RESP --> CHECK
    
    MONITOR --> TOKEN
    MONITOR --> SLIDE
    TOKEN --> TRUNC
    TOKEN --> SUMTRUNC
    SLIDE --> TRUNC
    SLIDE --> SUMTRUNC
    TRUNC --> RECENT
    SUMTRUNC --> RECENT
    
    CHECK -->|"> threshold"| SAVE
    CHECK -->|"≤ threshold"| Output
    SAVE --> PREVIEW
    PREVIEW --> Tools
    Tools -.->|"On demand"| Output
    
    RECENT --> CLEAN
    PREVIEW --> CLEAN

    %% Styling
    style Input fill:#3498db,stroke:#2980b9,color:#fff
    style Layer1 fill:#2c3e50,stroke:#34495e,color:#fff
    style Modes fill:#34495e,stroke:#2c3e50,color:#fff
    style Strategies fill:#34495e,stroke:#2c3e50,color:#fff
    style Layer2 fill:#8e44ad,stroke:#9b59b6,color:#fff
    style Offload fill:#9b59b6,stroke:#8e44ad,color:#fff
    style Tools fill:#9b59b6,stroke:#8e44ad,color:#fff
    style Output fill:#27ae60,stroke:#2ecc71,color:#fff

How the Two Layers Work Together

Layer	Scope	What It Manages	When It Triggers
Context Management	Agent loop messages	User/Assistant conversation, tool call history	When context exceeds threshold_percent of limit
Tool Offloading	Individual tool responses	Large API responses, file contents, search results	When response exceeds threshold_tokens

---

3.1 Agent Loop Context Management

Prevent token exhaustion during long-running tasks with automatic context management. When enabled, the agent monitors context size and applies truncation or summarization when thresholds are exceeded.

agent_config = {
    "context_management": {
        "enabled": True,
        "mode": "token_budget",  # or "sliding_window"
        "value": 100000,  # Max tokens (token_budget) or max messages (sliding_window)
        "threshold_percent": 75,  # Trigger at 75% of limit
        "strategy": "summarize_and_truncate",  # or "truncate"
        "preserve_recent": 4,  # Always keep last N messages
    }
}

Modes:

Mode	Description	Best For
token_budget	Manage by total token count	Cost control, API limits
sliding_window	Manage by message count	Predictable context size

Strategies:

Strategy	Behavior	Trade-off
truncate	Drop oldest messages	Fast, no extra LLM calls
summarize_and_truncate	Summarize then drop	Preserves context, adds latency

---

3.2 Tool Response Offloading

Large tool responses are automatically saved to files, with only a preview in context. The agent can retrieve full content on demand using built-in tools.

agent_config = {
    "tool_offload": {
        "enabled": True,
        "threshold_tokens": 500,  # Offload responses > 500 tokens
        "max_preview_tokens": 150,  # Show first 150 tokens in context
        "storage_dir": ".omnicoreagent_artifacts"
    }
}

Token Savings Example:

Tool Response	Without Offloading	With Offloading	Savings
Web search (50 results)	~10,000 tokens	~200 tokens	98%
Large API response	~5,000 tokens	~150 tokens	97%
File read (1000 lines)	~8,000 tokens	~200 tokens	97%

Built-in Artifact Tools (automatically available when offloading is enabled):

Tool	Purpose
read_artifact(artifact_id)	Read full content when needed
tail_artifact(artifact_id, lines)	Read last N lines (great for logs)
search_artifact(artifact_id, query)	Search within large responses
list_artifacts()	See all offloaded data in current session

---

Combined Power

Enable both for maximum efficiency:

agent = OmniCoreAgent(
    name="research_agent",
    agent_config={
        "context_management": {"enabled": True, "strategy": "summarize_and_truncate"},
        "tool_offload": {"enabled": True, "threshold_tokens": 500}
    }
)
# Result: Agents that can run indefinitely without token exhaustion

💡 When to Use: Enable for long-running tasks (research, multi-step workflows) where context or tool responses can grow unbounded.

4. 📡 Event System (Plug & Play)

Real-time event streaming with runtime switching:

from omnicoreagent import EventRouter

# Start with in-memory
agent = OmniCoreAgent(
    event_router=EventRouter("in_memory"),
    ...
)

# Switch to Redis Streams for production
agent.switch_event_store("redis_stream")
agent.get_event_store_type()                    # Get current event router type
# Stream events in real-time
async for event in agent.stream_events(session_id):
    print(f"{event.type}: {event.payload}")

Event Types: user_message, agent_message, tool_call_started, tool_call_result, final_answer, agent_thought, sub_agent_started, sub_agent_error, sub_agent_result

💡 When to Use: Enable events when you need real-time monitoring, debugging, or building UIs that show agent progress. Essential for production observability.

5. 🔌 Built-in MCP Client

Connect to any MCP-compatible service with support for multiple transport protocols and authentication methods.

Transport Types

1. stdio — Local MCP servers (process communication)

{
    "name": "filesystem",
    "transport_type": "stdio",
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home"]
}

2. streamable_http — Remote servers with HTTP streaming

# With Bearer Token
{
    "name": "github",
    "transport_type": "streamable_http",
    "url": "http://localhost:8080/mcp",
    "headers": {
        "Authorization": "Bearer your-token" # optional
    },
    "timeout": 60 # optional
}

# With OAuth 2.0 (auto-starts callback server on localhost:3000)
{
    "name": "oauth_server",
    "transport_type": "streamable_http",
    "auth": {
        "method": "oauth"
    },
    "url": "http://localhost:8000/mcp"
}

3. sse — Server-Sent Events

{
    "name": "sse_server",
    "transport_type": "sse",
    "url": "http://localhost:3000/sse",
    "headers": {
        "Authorization": "Bearer token" # optional
    },
    "timeout": 60, # optional
    "sse_read_timeout": 120 # optional
}

Complete Example with All 3 Transport Types

agent = OmniCoreAgent(
    name="multi_mcp_agent",
    system_instruction="You have access to filesystem, GitHub, and live data.",
    model_config={"provider": "openai", "model": "gpt-4o"},
    mcp_tools=[
        # 1. stdio - Local filesystem
        {
            "name": "filesystem",
            "transport_type": "stdio",
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home"]
        },
        # 2. streamable_http - Remote API (supports Bearer token or OAuth)
        {
            "name": "github",
            "transport_type": "streamable_http",
            "url": "http://localhost:8080/mcp",
            "headers": {"Authorization": "Bearer github-token"},
            "timeout": 60
        },
        # 3. sse - Real-time streaming
        {
            "name": "live_data",
            "transport_type": "sse",
            "url": "http://localhost:3000/sse",
            "headers": {"Authorization": "Bearer token"},
            "sse_read_timeout": 120
        }
    ]
)

await agent.connect_mcp_servers()
tools = await agent.list_all_available_tools()  # All MCP + local tools
result = await agent.run("List all Python files and get latest commits")

Transport Comparison

Transport	Use Case	Auth Methods
stdio	Local MCP servers, CLI tools	None (local process)
streamable_http	Remote APIs, cloud services	Bearer token, OAuth 2.0
sse	Real-time data, streaming	Bearer token, custom headers

💡 When to Use: Use MCP when you need to connect to external tools and services. Choose stdio for local CLI tools, streamable_http for REST APIs, and sse for real-time streaming data.

---

6. 🧠 DeepAgent (Multi-Agent Orchestration)

DeepAgent = OmniCoreAgent + Multi-Agent Orchestration

DeepAgent automatically breaks down complex tasks and delegates them to specialized subagents running in parallel. The lead agent coordinates the work and synthesizes findings from memory.

How It Works

For complex tasks, DeepAgent can spawn parallel subagents:

1. Analyzes task complexity
2. Creates memory plan (/memories/task_name/)
3. Spawns specialized subagents in parallel
4. Subagents write findings to memory (not context)
5. Lead agent reads from memory to synthesize

Architecture Flow:

User Query → Lead Agent → Spawn Subagents (parallel)
                ↓
            [Subagent A] → Write to /memories/subtask_a/
            [Subagent B] → Write to /memories/subtask_b/
            [Subagent C] → Write to /memories/subtask_c/
                ↓
Lead Agent reads memory → Synthesize → Final Answer

Quick Start

from omnicoreagent import DeepAgent

# Create a DeepAgent for any domain
agent = DeepAgent(
    name="research_coordinator",
    system_instruction="You are a tech research coordinator.",
    model_config={"provider": "openai", "model": "gpt-4o"},
)

await agent.initialize()  # Registers orchestration tools

# Run complex query - automatically spawns subagents
result = await agent.run("""
Research the benefits of Rust vs Go for cloud-native applications.
Consider performance, developer experience, and ecosystem maturity.
""")
# DeepAgent spawns 3 parallel subagents:
# - Performance researcher
# - DevEx analyst  
# - Ecosystem analyst

await agent.cleanup()

DeepAgent vs OmniCoreAgent

Feature	OmniCoreAgent	DeepAgent
Domain	User-defined	User-defined (same)
Tools	User-provided	User-provided + orchestration
Memory Backend	Optional	Always "local" (enforced)
Orchestration	No	Automatic subagent spawning
Best For	Single-agent tasks	Complex multi-step analysis

Built-in Orchestration Tools

DeepAgent provides two specialized tools to the lead agent:

• spawn_subagent - Spawn a single focused subagent
• spawn_parallel_subagents - Spawn multiple subagents in parallel

Subagents inherit:

• Parent's model config
• Parent's tools (MCP + local)
• Parent's agent config (context management, tool offload, etc.)

📚 Learn More: See DeepAgent Documentation for architecture diagrams, use cases, and best practices.

💡 When to Use: Use DeepAgent when your tasks may benefit from multi-agent orchestration (parallel research, divide-and-conquer analysis, multi-domain expertise).

7. 🛠️ Local Tools System

Register any Python function as an AI tool:

from omnicoreagent import ToolRegistry

tools = ToolRegistry()

@tools.register_tool("get_weather")
def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"Weather in {city}: Sunny, 25°C"

@tools.register_tool("calculate_area")
def calculate_area(length: float, width: float) -> str:
    """Calculate rectangle area."""
    return f"Area: {length * width} square units"

agent = OmniCoreAgent(
    name="tool_agent",
    local_tools=tools,  # Your custom tools!
    ...
)

💡 When to Use: Use Local Tools when you need custom business logic, internal APIs, or any Python functionality that isn't available via MCP servers.

---

8. 🧩 Agent Skills System (Packaged Capabilities)

OmniCoreAgent supports the Agent Skills specification — self-contained capability packages that provide specialized knowledge, executable scripts, and documentation.

agent_config = {
    "enable_agent_skills": True  # Enable discovery and tools for skills
}

Key Concepts:

• Discovery: Agents automatically discover skills installed in .agents/skills/[skill-name].
• Activation (SKILL.md): Agents are instructed to read the "Activation Document" first to understand how to use the skill's specific capabilities.
• Polyglot Execution: The run_skill_script tool handles scripts in Python, JavaScript/Node, TypeScript, Ruby, Perl, and Shell (bash/sh).

Directory Structure:

.agents/skills/my-skill-name/
├── SKILL.md        # The "Activation" document (instructions + metadata)
├── scripts/        # Multi-language executable scripts
├── references/     # Deep-dive documentation
└── assets/         # Templates, examples, and resources

Skill Tools:

• read_skill_file(skill_name, file_path): Access any file within a skill (start with SKILL.md).
• run_skill_script(skill_name, script_name, args?): Execute bundled scripts with automatic interpreter detection.

📚 Learn More: To learn how to create your own agent skills, visit agentskills.io.

---

9. 💾 Workspace Memory — Persistent File Storage for Agents

NEW: Cloud Storage Support! Your agents can now store files on AWS S3 or Cloudflare R2 for production-grade, distributed persistence.

A persistent file storage system that gives your agents a dedicated workspace to save, manage, and share files across sessions. Choose from local filesystem for development, or cloud storage (S3/R2) for production deployments where files need to persist across servers, scale globally, and survive restarts.

Storage Backends

Backend	Use Case	Benefits
local	Development, single-server	Zero config, instant setup
s3	Production, AWS infrastructure	Scalable, durable, global access
r2	Production, edge computing	Zero egress fees, Cloudflare ecosystem

Quick Setup

# Local storage (development)
agent_config = {
    "memory_tool_backend": "local"
}

# AWS S3 storage (production)
agent_config = {
    "memory_tool_backend": "s3"
}

# Cloudflare R2 storage (production)
agent_config = {
    "memory_tool_backend": "r2"
}

Environment Variables

For S3:

AWS_S3_BUCKET=my-agent-memories
AWS_ACCESS_KEY_ID=AKIA...
AWS_SECRET_ACCESS_KEY=your-secret
AWS_REGION=us-east-1  # optional, defaults to us-east-1

For R2:

R2_BUCKET_NAME=my-agent-memories
R2_ACCOUNT_ID=your-cloudflare-account-id
R2_ACCESS_KEY_ID=your-r2-key
R2_SECRET_ACCESS_KEY=your-r2-secret

Agent Memory Tools

When enabled, your agent automatically gets these tools:

Tool	Purpose
memory_view	View/list files in memory workspace
memory_create_update	Create, append, or overwrite files
memory_str_replace	Find and replace text within files
memory_insert	Insert text at specific line numbers
memory_delete	Delete files from workspace
memory_rename	Rename or move files
memory_clear_all	Clear entire workspace

Production Features

Feature	Local	S3	R2
Persistent across restarts	✅	✅	✅
Multi-server access	❌	✅	✅
Global CDN distribution	❌	✅	✅
Zero egress fees	N/A	❌	✅
Auto-retry on failure	❌	✅	✅
Concurrent access safety	✅	✅	✅

Use Cases

Use Case	Recommended Backend
Local development	local
Single-server production	local or s3
Multi-server / Kubernetes	s3 or r2
Edge computing / Workers	r2
Cost-sensitive workloads	r2 (zero egress)

Example: Research Agent with Cloud Storage

import os

# Set environment for S3
os.environ["AWS_S3_BUCKET"] = "research-agent-memories"
os.environ["AWS_ACCESS_KEY_ID"] = "AKIA..."
os.environ["AWS_SECRET_ACCESS_KEY"] = "..."
os.environ["AWS_REGION"] = "us-east-1"
os.environ["AWS_ENDPOINT_URL"] = "https://s3.amazonaws.com" # Optional

agent = OmniCoreAgent(
    name="research_agent",
    system_instruction="You are a research assistant. Save your findings to memory.",
    model_config={"provider": "openai", "model": "gpt-4o"},
    agent_config={
        "memory_tool_backend": "s3",  # Files persist in S3
        "max_steps": 50,
    }
)

# Agent can now save research notes that persist across:
# - Server restarts
# - Multiple instances
# - Different geographic locations
result = await agent.run(
    "Research the latest AI developments and save a summary to /notes/ai_trends_2024.md"
)

💡 When to Use: Use local for development. Use s3 or r2 when you need:

• Files to persist across server restarts
• Multiple agent instances accessing the same workspace
• Global teams accessing shared agent knowledge
• Production-grade durability and reliability

---

10. 👥 Sub-Agents System

Delegate tasks to specialized child agents:

weather_agent = OmniCoreAgent(name="weather_agent", ...)
filesystem_agent = OmniCoreAgent(name="filesystem_agent", mcp_tools=MCP_TOOLS, ...)

parent_agent = OmniCoreAgent(
    name="parent_agent",
    sub_agents=[weather_agent, filesystem_agent],
    ...
)

💡 When to Use: Use Sub-Agents when you have specialized agents (e.g., weather, code, data) and want a parent agent to delegate tasks intelligently. Great for building modular, reusable agent architectures.

---

11. 🕰️ Background Agents

Autonomous agents that run on varying schedules (Interval or Cron) or process tasks from a persistent queue.

from omnicoreagent import BackgroundAgentManager, MemoryRouter, EventRouter
    
# Initialize the manager
manager = BackgroundAgentManager(
    memory_router=MemoryRouter("redis"),
    event_router=EventRouter("redis_stream")
)
    
# Create a background agent
background_agent_config = {
    "agent_id": "system_monitor",
    "system_instruction": "Monitor system resources and report anomalies.",
    "model_config": {"provider": "openai", "model": "gpt-4o-mini"},
    "queue_size": 10,  # Max pending tasks
    # Schedule options:
    "interval": 300,        # Integer = seconds (every 5 mins)
    # "interval": "* * * * *", # String = Cron expression (every minute)
    "task_config": {
        "query": "Check metrics and alert if CPU > 80%",
        "timeout": 60,      # Kill task if it hangs
        "max_retries": 3,
        "retry_delay": 10
    }
}
await manager.create_agent(background_agent_config)
    
# Start the system
await manager.start()

# Trigger manually if needed
await manager.run_task_now("system_monitor", {"query": "Immediate check!"})

📚 Deep Dive: Check out the Background Agents Cookbook for full "Kitchen Sink" examples including lifecycle management (pause/resume/delete) and advanced configuration.

🛠️ Comprehensive API Reference

BackgroundAgentManager (Orchestrator)

Lifecycle Management

• await manager.create_agent(config: Dict) -> Dict: Create, register, and schedule a new background agent.
• await manager.start(): Start the manager and all scheduled agents.
• await manager.shutdown(): Gracefully stop the manager and all agents.
• await manager.start_agent(agent_id): Start (schedule) a specific agent.
• await manager.stop_agent(agent_id): Stop (unschedule) a specific agent.
• await manager.pause_agent(agent_id): Pause an agent's schedule without stopping its worker.
• await manager.resume_agent(agent_id): Resume a paused agent's schedule.
• await manager.delete_agent(agent_id): Stop, cleanup, and remove an agent completely.

Task Management

• await manager.register_task(agent_id, task_config): Register or update a task for an agent.
• await manager.run_task_now(agent_id, task_config): Trigger an immediate execution outside the schedule.
• await manager.register_and_run(agent_id, task_config): Register a task and run it immediately.
• await manager.update_task_config(agent_id, task_config): Update an existing task configuration.
• await manager.remove_task(agent_id): Remove a task configuration.
• await manager.list_tasks(): List all agents with registered tasks.

Status & Monitoring

• await manager.get_agent_status(agent_id): Get comprehensive status (running, scheduled, last run, errors).
• await manager.get_manager_status(): Get overall system status (total agents, running count, resource usage).
• await manager.list_agents(): List all registered agent IDs.
• await manager.is_agent_running(agent_id): Check if an agent is currently executing a task.
• await manager.get_running_agents(): List all currently executing agents.
• await manager.get_agent_metrics(agent_id): Get performance metrics (run count, errors, timestamps).
• await manager.get_all_metrics(): Get metrics for all agents.

Configuration & Information

• await manager.update_agent_config(agent_id, new_config): Update agent settings (including model/tools).
• await manager.get_task_config(agent_id): Retrieve current task configuration.
• await manager.get_agent(agent_id): Access the raw BackgroundOmniCoreAgent instance.
• await manager.get_agent_event_info(agent_id): Get event stream connection details.
• await manager.get_all_event_info(): Get event info for all agents and shared stores.
• await manager.get_agent_session_id(agent_id): Get the persistent session ID.
• await manager.get_all_session_ids(): Get a map of all agent session IDs.

---

BackgroundOmniCoreAgent (The Workers)

Execution & Control

• await agent.submit_task(task_config): Queue a task for reliable execution.
• await agent.run_task(task_config): Interface for scheduler/manual triggers.
• await agent.start_worker(): Start the background task processing loop.
• await agent.stop_worker(): Gracefully stop the background worker.
• property agent.is_worker_running: Check if the worker loop is active.
• await agent.connect_mcp_servers(): Establish connections to configured MCP tools.
• await agent.cleanup(): comprehensive cleanup of resources, connections, and tasks.

State & Visibility

• await agent.get_status(): Get health, configuration, and execution state.
• await agent.get_session_id(): Get the persistent session ID.
• await agent.has_task(): Check if a valid task is registered.
• await agent.get_task_query(): Get the current query/instruction being executed.
• await agent.get_task_config(): Get the full task configuration dict.

Events & Streaming

• await agent.stream_events(session_id): Real-time event generator.
• await agent.get_events(session_id): Retrieve past event history.
• await agent.get_event_stream_info(): Connection details for external consumers.
• await agent.update_config(new_config): Hot-reload agent configuration.

💡 When to Use: Perfect for scheduled tasks like system monitoring, periodic reports, data syncing, or any automation that runs independently without user interaction.

---

12. 🔄 Workflow Agents

Orchestrate multiple agents for complex tasks:

from omnicoreagent import SequentialAgent, ParallelAgent, RouterAgent

# Sequential: Chain agents step-by-step
seq_agent = SequentialAgent(sub_agents=[agent1, agent2, agent3])
result = await seq_agent.run(initial_task="Analyze and report")

# Parallel: Run agents concurrently
par_agent = ParallelAgent(sub_agents=[agent1, agent2, agent3])
results = await par_agent.run(agent_tasks={
    "analyzer": "Analyze data",
    "processor": "Process results"
})

# Router: Intelligent task routing
router = RouterAgent(
    sub_agents=[code_agent, data_agent, research_agent],
    model_config={"provider": "openai", "model": "gpt-4o"}
)
result = await router.run(task="Find and summarize AI research")

💡 When to Use:

• SequentialAgent: When tasks depend on each other (output of one → input of next)
• ParallelAgent: When tasks are independent and can run simultaneously for speed
• RouterAgent: When you need intelligent task routing to specialized agents

---

13. 🧠 Advanced Tool Use (BM25 Retrieval)

Automatically discover relevant tools at runtime using BM25 lexical search:

agent_config = {
    "enable_advanced_tool_use": True  # Enable BM25 retrieval
}

How It Works:

1. All MCP tools loaded into in-memory registry
2. BM25 index built over tool names, descriptions, parameters
3. User task used as search query
4. Top 5 relevant tools dynamically injected

Benefits: Scales to 1000+ tools, zero network I/O, deterministic, container-friendly.

💡 When to Use: Enable when you have many MCP tools (10+) and want the agent to automatically discover the right tools for each task without manual selection.

---

14. 📊 Production Observability & Metrics

📈 Real-time Usage Metrics

OmniCoreAgent tracks every token, request, and millisecond. Each run() returns a metric object, and you can get cumulative stats anytime.

result = await agent.run("Analyze this data")
print(f"Request Tokens: {result['metric'].request_tokens}")
print(f"Time Taken: {result['metric'].total_time:.2f}s")

# Get aggregated metrics for the agent's lifecycle
stats = await agent.get_metrics()
print(f"Avg Response Time: {stats['average_time']:.2f}s")

🔍 Opik Tracing

Monitor and optimize your agents with deep traces:

# Add to .env
OPIK_API_KEY=your_opik_api_key
OPIK_WORKSPACE=your_workspace

What's Tracked: LLM call performance, tool execution traces, memory operations, agent workflow, bottlenecks.

Agent Execution Trace:
├── agent_execution: 4.6s
    ├── tools_registry_retrieval: 0.02s ✅
    ├── memory_retrieval_step: 0.08s ✅
    ├── llm_call: 4.5s ⚠️ (bottleneck!)
    └── action_execution: 0.03s ✅

💡 When to Use: Essential for production. Use Metrics for cost/performance monitoring, and Opik for identifying bottlenecks and debugging complex agent logic.

---

15. 🛡️ Prompt Injection Guardrails

Protect your agents against malicious inputs, jailbreaks, and instruction overrides before they reach the LLM.

agent_config = {
    "guardrail_config": {
        "strict_mode": True,      # Block all suspicious inputs
        "sensitivity": 0.85,      # 0.0 to 1.0 (higher = more sensitive)
        "enable_pattern_matching": True,
        "enable_heuristic_analysis": True
    }
}

agent = OmniCoreAgent(..., agent_config=agent_config)

# If a threat is detected:
# result['response'] -> "I'm sorry, but I cannot process this request due to safety concerns..."
# result['guardrail_result'] -> Full metadata about the detected threat

Key Protections:

• Instruction Overrides: "Ignore previous instructions..."
• Jailbreaks: DAN mode, roleplay escapes, etc.
• Toxicity & Abuse: Built-in pattern recognition.
• Payload Splitting: Detects fragmented attack attempts.

⚙️ Configuration Options

Parameter	Type	Default	Description
strict_mode	bool	False	When True, any detection (even low confidence) blocks the request.
sensitivity	float	1.0	Scaling factor for threat scores (0.0 to 1.0). Higher = more sensitive.
max_input_length	int	10000	Maximum allowed query length before blocking.
enable_encoding_detection	bool	True	Detects base64, hex, and other obfuscation attempts.
enable_heuristic_analysis	bool	True	Analyzes prompt structure for typical attack patterns.
enable_sequential_analysis	bool	True	Checks for phased attacks across multiple tokens.
enable_entropy_analysis	bool	True	Detects high-entropy payloads common in injections.
allowlist_patterns	list	[]	List of regex patterns that bypass safety checks.
blocklist_patterns	list	[]	Custom regex patterns to always block.

💡 When to Use: Always enable in user-facing applications to prevent prompt injection attacks and ensure agent reliability.

---

16. 🌐 Universal Model Support

Model-agnostic through LiteLLM — use any provider:

# OpenAI
model_config = {"provider": "openai", "model": "gpt-4o"}

# Anthropic
model_config = {"provider": "anthropic", "model": "claude-3-5-sonnet-20241022"}

# Groq (Ultra-fast)
model_config = {"provider": "groq", "model": "llama-3.1-8b-instant"}

# Ollama (Local)
model_config = {"provider": "ollama", "model": "llama3.1:8b", "ollama_host": "http://localhost:11434"}

# OpenRouter (200+ models)
model_config = {"provider": "openrouter", "model": "anthropic/claude-3.5-sonnet"}

#mistral ai
model_config = {"provider": "mistral", "model": "mistral-7b-instruct"}

#deepseek
model_config = {"provider": "deepseek", "model": "deepseek-chat"}

#google gemini
model_config = {"provider": "google", "model": "gemini-2.0-flash-exp"}

#azure openai
model_config = {"provider": "azure_openai", "model": "gpt-4o"}

Supported: OpenAI, Anthropic, Google Gemini, Groq, DeepSeek, Mistral, Azure OpenAI, OpenRouter, Ollama

💡 When to Use: Switch providers based on your needs — use cheaper models (Groq, DeepSeek) for simple tasks, powerful models (GPT-4o, Claude) for complex reasoning, and local models (Ollama) for privacy-sensitive applications.

---

17. 🚀 OmniServe — Production API Server

Turn any agent into a production-ready REST/SSE API with a single command.

OmniServe transforms OmniCoreAgent or DeepAgent into a full FastAPI server with:

• ✅ REST and SSE streaming endpoints
• ✅ Rate limiting with configurable limits
• ✅ Prometheus metrics at /prometheus
• ✅ OpenTelemetry tracing support
• ✅ Environment variable configuration
• ✅ Docker deployment ready
• ✅ Retry logic and circuit breaker for resilience

Quick Start (Zero Code)

# Start an agent server without writing any code
omniserve quickstart --provider gemini --model gemini-2.0-flash --port 8000

# Or run your existing agent file
omniserve run --agent my_agent.py --port 8000 --rate-limit 100

Python Usage

from omnicoreagent import OmniCoreAgent, OmniServe, OmniServeConfig

agent = OmniCoreAgent(
    name="ProductionAgent",
    system_instruction="You are a helpful assistant.",
    model_config={"provider": "gemini", "model": "gemini-2.0-flash"},
)

# Configure the server
config = OmniServeConfig(
    port=8000,
    cors_origins=["https://myapp.com"],
    auth_enabled=True,
    auth_token="my-secret-token",
    rate_limit_enabled=True,
    rate_limit_requests=100,  # 100 requests per minute
)

# Start the server (blocking)
server = OmniServe(agent, config=config)
server.start()

API Endpoints

Method	Endpoint	Description
POST	/run	SSE streaming response
POST	/run/sync	JSON response
GET	/health	Health check
GET	/ready	Readiness check
GET	/prometheus	Prometheus metrics
GET	/tools	List available tools
GET	/metrics	Agent usage metrics
GET	/events/{session_id}	Get session events
GET	/sessions/{session_id}	Get session history
DELETE	/sessions/{session_id}	Clear session

Environment Variables

All settings can be configured via environment variables with OMNISERVE_ prefix:

Variable	Description	Default
OMNISERVE_HOST	Host to bind to	0.0.0.0
OMNISERVE_PORT	Port to bind to	8000
OMNISERVE_WORKERS	Number of worker processes	1
OMNISERVE_API_PREFIX	API path prefix (e.g., /api/v1)	""
OMNISERVE_ENABLE_DOCS	Enable Swagger UI at /docs	true
OMNISERVE_ENABLE_REDOC	Enable ReDoc at /redoc	true
OMNISERVE_CORS_ENABLED	Enable CORS middleware	true
OMNISERVE_CORS_ORIGINS	Comma-separated allowed origins	*
OMNISERVE_CORS_CREDENTIALS	Allow credentials in CORS	true
OMNISERVE_AUTH_ENABLED	Enable Bearer token auth	false
OMNISERVE_AUTH_TOKEN	Bearer token for authentication	—
OMNISERVE_REQUEST_LOGGING	Log incoming requests	true
OMNISERVE_LOG_LEVEL	Logging level (DEBUG, INFO, WARNING, ERROR)	INFO
OMNISERVE_REQUEST_TIMEOUT	Request timeout in seconds	300
OMNISERVE_RATE_LIMIT_ENABLED	Enable rate limiting	false
OMNISERVE_RATE_LIMIT_REQUESTS	Max requests per time window	100
OMNISERVE_RATE_LIMIT_WINDOW	Rate limit time window in seconds	60

Example .env file:

# Required: Your LLM API key
LLM_API_KEY=your-api-key-here

# Server settings
OMNISERVE_PORT=8000
OMNISERVE_LOG_LEVEL=INFO

# Security
OMNISERVE_AUTH_ENABLED=true
OMNISERVE_AUTH_TOKEN=my-secret-token
OMNISERVE_CORS_ORIGINS=https://myapp.com

# Rate limiting
OMNISERVE_RATE_LIMIT_ENABLED=true
OMNISERVE_RATE_LIMIT_REQUESTS=100
OMNISERVE_RATE_LIMIT_WINDOW=60

Docker Deployment

Deploy with the full observability stack (Prometheus + Grafana):

cd docker
cp .env.example .env
# Edit .env with your LLM_API_KEY
docker-compose up

Services:

• OmniServe: http://localhost:8000 (API + Swagger UI)
• Prometheus: http://localhost:9090 (Metrics)
• Grafana: http://localhost:3000 (Dashboards, login: admin/admin)

Extensibility

Import components to build custom resilience patterns:

from omnicoreagent.omni_agent.omni_serve import (
    RetryConfig,
    CircuitBreaker,
    with_retry,
    get_metrics,
)

# Add retry to any async function
@with_retry(RetryConfig(max_retries=5, strategy="exponential"))
async def call_external_api():
    ...

# Circuit breaker for failure protection
breaker = CircuitBreaker("my-api", failure_threshold=3, timeout=60)
async with breaker:
    result = await risky_call()

# Access Prometheus metrics programmatically
metrics = get_metrics()
print(metrics.to_prometheus())

CLI Commands

# Show help
omniserve --help

# Quickstart (zero code)
omniserve quickstart --provider openai --model gpt-4o

# Run with an agent file
omniserve run --agent my_agent.py --port 8000 --auth-token secret

# Generate .env template
omniserve config --env-example > .env

# View current config from environment
omniserve config --show

# 🐳 Generate Docker Deployment
omniserve generate-deployment --file <agent_file_path>

Production Deployment Generator

Ship your agents to production in seconds with our built-in Docker generator. This tool inspects your agent code and auto-generates a production-ready docker-compose.yml and Dockerfile.

Features:

• Smart Inspection: Auto-configures S3/R2 memory if detected in your code.
• Persistent Storage: Mounts .omnicoreagent_config and .omnicoreagent_artifacts volumes with correct permissions.
• Universal: Works with simple agent scripts (omniserve run) AND full OmniServe apps (python app.py).

Requirements: Your file must define either:

1. An agent variable (e.g., agent = OmniCoreAgent(...), agent = DeepAgent(...))
2. A create_agent() function that returns an agent (for complex setups).

omniserve generate-deployment --file cookbook/omniserve/python_api.py

💡 When to Use: Use OmniServe when you need to deploy agents as APIs — perfect for microservices, webhooks, chatbots, and any application that needs to consume agent capabilities over HTTP.

---

📚 Examples & Cookbook

All examples are in the Cookbook — organized by use case with progressive learning paths.

Quick Links

Category	What You'll Build	Location
Getting Started	Your first agent, tools, memory, events	cookbook/getting_started
Workflows	Sequential, Parallel, Router agents	cookbook/workflows
Background Agents	Scheduled autonomous tasks	cookbook/background_agents
Production	Metrics, guardrails, observability	cookbook/production
🏆 Showcase	Full production applications	cookbook/showcase

🏆 Showcase: Full Production Applications

Application	Description	Features
OmniAudit	Healthcare Claims Audit System	Multi-agent pipeline, ERISA compliance
DevOps Copilot	AI-Powered DevOps Automation	Docker, Prometheus, Grafana
Deep Code Agent	Code Analysis with Sandbox	Sandbox execution, session management

Featured Examples

Agent	Description	Location
E-commerce Shopper	Personal shopping with cart, preferences, recommendations	cookbook/advanced_agent
Flight Booking	Travel agent with search, booking, itineraries	cookbook/advanced_agent
AI Due Diligence	Investment research with web search, analysis	cookbook/advanced_agent/ai_due_diligence_agent

# Start with the basics
python cookbook/getting_started/first_agent.py

# Or explore a full production application
cd cookbook/showcase/devops_copilot_agent && make up

---

⚙️ Configuration

Environment Variables

# Required
LLM_API_KEY=your_api_key

# Optional: Memory backends
REDIS_URL=redis://localhost:6379/0
DATABASE_URL=postgresql://user:pass@localhost:5432/db
MONGODB_URI=mongodb://localhost:27017/omnicoreagent

# Optional: Observability
OPIK_API_KEY=your_opik_key
OPIK_WORKSPACE=your_workspace

Agent Configuration

agent_config = {
    "max_steps": 15,                    # Max reasoning steps
    "tool_call_timeout": 30,            # Tool timeout (seconds)
    "request_limit": 0,                 # 0 = unlimited
    "total_tokens_limit": 0,            # 0 = unlimited
    "memory_config": {"mode": "sliding_window", "value": 10000},
    "enable_advanced_tool_use": True,   # BM25 tool retrieval
    "enable_agent_skills": True,        # Specialized packaged skills
    "memory_tool_backend": "local"      # Persistent working memory
}

Model Configuration

model_config = {
    "provider": "openai",
    "model": "gpt-4o",
    "temperature": 0.7,
    "max_tokens": 2000,
    "top_p": 0.95
}

📋 Additional Model Configurations

# Azure OpenAI
model_config = {
    "provider": "azureopenai",
    "model": "gpt-4",
    "azure_endpoint": "https://your-resource.openai.azure.com",
    "azure_api_version": "2024-02-01"
}

# Ollama (Local)
model_config = {
    "provider": "ollama",
    "model": "llama3.1:8b",
    "ollama_host": "http://localhost:11434"
}

---

🧪 Testing & Development

# Clone
git clone https://github.com/omnirexflora-labs/omnicoreagent.git
cd omnicoreagent

# Setup
uv venv && source .venv/bin/activate
uv sync --dev

# Test
pytest tests/ -v
pytest tests/ --cov=src --cov-report=term-missing

---

🔍 Troubleshooting

Error	Fix
Invalid API key	Check .env: LLM_API_KEY=your_key
ModuleNotFoundError	pip install omnicoreagent
Redis connection failed	Start Redis or use MemoryRouter("in_memory")
MCP connection refused	Ensure MCP server is running

📋 More Troubleshooting

OAuth Server Starts: Normal when using "auth": {"method": "oauth"}. Remove if not needed.

Debug Mode: agent = OmniCoreAgent(..., debug=True)

OmniAgent → OmniCoreAgent Migration: If you were using the old OmniAgent class, update your imports:

# Old (deprecated)
from omnicoreagent import OmniAgent

# New (recommended)
from omnicoreagent import OmniCoreAgent

The OmniAgent alias still works but will be removed in a future release.

Help: Check GitHub Issues

---

🤝 Contributing

# Fork & clone
git clone https://github.com/omnirexflora-labs/omnicoreagent.git

# Setup
uv venv && source .venv/bin/activate
uv sync --dev
pre-commit install

# Submit PR

See CONTRIBUTING.md for guidelines.

---

📄 License

MIT License — see LICENSE

---

👨‍💻 Author & Credits

Created by Abiola Adeshina

• GitHub: @Abiorh001
• X (Twitter): @abiorhmangana
• Email: [e-mail]

🌟 The OmniRexFlora Ecosystem

Project	Description
🧠 OmniMemory	Self-evolving memory for autonomous agents
🤖 OmniCoreAgent	Production-ready AI agent framework (this project)
⚡ OmniDaemon	Event-driven runtime engine for AI agents

🙏 Acknowledgments

Built on: LiteLLM, FastAPI, Redis, Opik, Pydantic, APScheduler

---

Building the future of production-ready AI agent frameworks

⭐ Star us on GitHub • 🐛 Report Bug • 💡 Request Feature • 📖 Documentation