Skip to main content

​
Basic Usage

This guide covers the most common patterns you’ll need when working with OmniCoreAgent — from running your first query to handling errors in production.

​
Running an Agent

Every interaction starts with agent.run(). It returns a dictionary with the agent’s response and metadata. 
import asyncio
from omnicoreagent import OmniCoreAgent

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

    result = await agent.run("What is the capital of France?")

    print(result["response"])       # The agent's text reply
    print(result["session_id"])     # Session used for this run
    print(result.get("metric"))     # Token/timing metric when provider usage is available

    await agent.cleanup()

asyncio.run(main())

​
Session Management

Use session_id to give your agent persistent memory across multiple calls. Without it, each call is stateless. 
# First interaction — agent learns the user's name
await agent.run("My name is Abiola.", session_id="user_42")

# Later interaction — agent remembers
result = await agent.run("What's my name?", session_id="user_42")
# → "Your name is Abiola."

​
Retrieving History

history = await agent.get_history(session_id="user_42")
for message in history:
    print(f"{message['role']}: {message['content'][:80]}")

​
Clearing History

await agent.clear_session_history(session_id="user_42")

​
Adding Memory Persistence

By default, history is stored in-memory (lost on restart). Add a MemoryRouter to persist across restarts. 
from omnicoreagent import OmniCoreAgent, MemoryRouter

agent = OmniCoreAgent(
    name="persistent_agent",
    system_instruction="You are a helpful assistant.",
    model_config={"provider": "openai", "model": "gpt-4o"},
    memory_router=MemoryRouter("redis")  # or "sql", "mongodb", "in_memory"
)
You can switch backends at runtime without restarting:
await agent.switch_memory_store("mongodb")

​
Using Tools

​
MCP Tools (External Servers)

Connect to any MCP-compatible tool server: 
agent = OmniCoreAgent(
    name="tool_agent",
    system_instruction="You can manage files and search the web.",
    model_config={"provider": "openai", "model": "gpt-4o"},
    mcp_tools=[
        {
            "name": "filesystem",
            "transport_type": "stdio",
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
        }
    ]
)

await agent.connect_mcp_servers()
result = await agent.run("List all files in /tmp")

​
Local Tools (Custom Python Functions)

Register any Python function as a tool the agent can call: 
from omnicoreagent import OmniCoreAgent, ToolRegistry

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

agent = OmniCoreAgent(
    name="weather_agent",
    system_instruction="You help with weather queries.",
    model_config={"provider": "openai", "model": "gpt-4o"},
    local_tools=tools
)

​
External Tools

Keep external integrations outside the core package. Use MCP servers for shared capabilities, or wrap project-owned APIs with ToolRegistry when the tool is specific to your application.

​
Event Streaming

Read telemetry events for UIs, logging, and debugging. Use run_id when multiple runs can share the same session_id. 
agent = OmniCoreAgent(
    name="streaming_agent",
    system_instruction="You are a helpful assistant.",
    model_config={"provider": "openai", "model": "gpt-4o"}
)

result = await agent.run("What can you do?")
session_id = result["session_id"]
run_id = result["run_id"]
trace_id = result["trace_id"]

events = await agent.get_telemetry_events_after(
    cursor=None,
    session_id=session_id,
    run_id=run_id,
)
for event in events:
    print(event.event_type, event.model_dump())

exact_trace = await agent.get_trace(trace_id)
latest_session_trace = await agent.get_latest_trace(session_id)
normalized_trace = await agent.get_trace(trace_id, normalize=True)

​
Error Handling

Wrap agent calls with try/except for production use: 
from omnicoreagent.exceptions import OmniCoreAgentError

try:
    result = await agent.run("Analyze this dataset", session_id="user_1")
except OmniCoreAgentError as e:
    print(f"Agent runtime error: {e.message}")
except Exception as e:
    print(f"Unexpected failure: {e}")

​
Common Troubleshooting

ErrorFix
Invalid API keyExport LLM_API_KEY with the key for the provider selected in model_config.
ModuleNotFoundError for an optional backendInstall the matching extra, e.g. pip install "omnicoreagent[redis]"
Redis connection failedStart Redis or use MemoryRouter("in_memory")
MCP connection refusedEnsure MCP server is running and path is correct
Token limit exceededIncrease total_tokens_limit or enable context management

​
Next Steps

Configuration

Full reference for env vars, agent settings, and models

Memory Deep Dive

5 backends with runtime switching and summarization

Agent Harness

Dynamic subagents and workspace-backed coordination for complex tasks