Skip to main content

​
OmniCoreAgent

The OmniCoreAgent class is the primary entry point for the framework. It provides a high-level, user-friendly API for creating and managing AI agents with built-in memory, events, and tool orchestration.

​
Quick Start

from omnicoreagent import OmniCoreAgent

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

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

​
Parameters

ParameterTypeDescription
namestrName of the agent (used for session tracking and logging).
system_instructionstrThe high-level objective or persona for the agent.
model_configdict or ModelConfigConfiguration for the LLM provider and model.
mcp_toolslist(Optional) List of MCP tool configurations.
local_toolsToolRegistry or list(Optional) Custom Python functions or community tools.
sub_agentslist(Optional) Specialized agents the parent can delegate to.
agent_configdict or AgentConfig(Optional) Settings for reasoning steps, timeouts, etc.
memory_routerMemoryRouter(Optional) Custom memory backend (defaults to in-memory).
event_routerEventRouter(Optional) Custom event backend (defaults to in-memory).

​
Production Agent (Full Configuration)

from omnicoreagent import OmniCoreAgent, ToolRegistry, MemoryRouter, EventRouter

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,
        },
    },
)

​
Core Methods

​
run()

Execute a task with the agent. Supports session continuity. 
result = await agent.run("What is the weather today?", session_id="user_1")
print(result['response'])
  • Query: The user’s input string.
  • Session ID: (Optional) Provide a unique ID to maintain conversation history.
  • Returns: A dictionary containing response, session_id, agent_name, and metric.
Each agent.run() call returns a metric field containing fine-grained usage (tokens, time) for that specific request.

​
connect_mcp_servers()

Establishes connections to all configured MCP servers. 
await agent.connect_mcp_servers()

​
list_all_available_tools()

Returns a list of all tools currently available (both MCP and local). 
tools = await agent.list_all_available_tools()

​
cleanup()

Cleans up resources, closes MCP connections, and removes temporary config files. 
await agent.cleanup()

​
Session Management

​
Retrieve History

history = await agent.get_session_history("user_1")
for msg in history:
    print(f"{msg['role']}: {msg['content']}")

​
Clear History

# Clear specific session
await agent.clear_session_history("user_1")

# Clear all history for this agent
await agent.clear_session_history()

​
Runtime Switching

Switch memory and event backends at runtime without restarting: 
# Switch memory backends
await agent.switch_memory_store("mongodb")
await agent.switch_memory_store("database")  # PostgreSQL/MySQL/SQLite
await agent.switch_memory_store("redis")

# Switch event backends
await agent.switch_event_store("redis_stream")

# Check current stores
await agent.get_memory_store_type()
await agent.get_event_store_type()

​
Complete API Reference

# Execution
await agent.run(query)                          # Execute task
await agent.run(query, session_id="user_1")     # With session context

# MCP
await agent.connect_mcp_servers()               # Connect MCP tools
await agent.cleanup_mcp_servers()               # Cleanup MCP without removing agent
await agent.list_all_available_tools()          # List all tools

# Memory
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
await agent.get_memory_store_type()              # Get current memory type

# Events
await agent.switch_event_store("redis_stream")   # Switch event backend
await agent.get_events(session_id)               # Get event history
await agent.get_event_store_type()               # Get current event type

# Metrics
await agent.get_metrics()                        # Cumulative usage stats

# Cleanup
await agent.cleanup()                            # Full cleanup

​
Best Practices

  1. Always Cleanup: Use await agent.cleanup() to avoid orphan processes (especially for MCP).
  2. Session IDs: Use meaningful session IDs (e.g., user database IDs) to persist context across interactions.
  3. Model Choice: Use cheaper models (e.g., gpt-4o-mini) for simple tasks and powerful ones (e.g., gpt-4o) when complex reasoning is required.
OmniCoreAgent is your go-to for any AI task — from simple Q&A to complex multi-step workflows. Start here for any agent project.