Quick Start: Build Your First Agent in 10 Minutes

This tutorial will get you from zero to a working AI agent in under 10 minutes. By the end, you’ll have:

✅ OmniDaemon installed and running
✅ Your first agent listening for events
✅ A publisher sending events
✅ Results being stored and retrieved

Let’s go!

Step 1: Install Event Bus & Storage Backend

For this Quick Start, we’ll use Redis (the current production-ready backend for both event bus and storage).

💡 OmniDaemon is pluggable! Redis Streams is our first event bus implementation. Coming soon: Kafka, RabbitMQ, NATS. For storage, we support JSON (dev) and Redis (production), with PostgreSQL, MongoDB, and S3 planned.

macOS

brew install redis
brew services start redis

Ubuntu/Debian

sudo apt update
sudo apt install redis-server
sudo systemctl start redis-server
sudo systemctl enable redis-server

Windows

# Option 1: Using WSL (recommended)
wsl --install
# Then follow Ubuntu steps above

# Option 2: Download installer from https://redis.io/download

Docker (All Platforms - Easiest!)

docker run -d -p 6379:6379 --name redis redis:latest

✅ Verify Event Bus is running (Redis Streams for this Quick Start):

redis-cli ping

Expected output: PONG ❌ If you see “command not found” or connection error, the event bus backend isn’t running. Try the Docker method above.

Step 2: Install OmniDaemon

Using uv (Recommended - Fast!)

# Install uv (if not already installed)
curl -LsSf https://astral.sh/uv/install.sh | sh
# Or: pip install uv

# Create a new project
mkdir my-omnidaemon-project
cd my-omnidaemon-project

# Initialize project
uv init

# Create virtual environment
uv venv

# Activate environment
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install OmniDaemon
uv add omnidaemon

# Verify installation
python -c "import omnidaemon; print('✅ OmniDaemon installed!')"

Using pip (Traditional)

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install OmniDaemon
pip install omnidaemon

# Verify installation
python -c "import omnidaemon; print('✅ OmniDaemon installed!')"

✅ Expected output:

✅ OmniDaemon installed!

Step 3: Create Your First Agent (Production-Ready)

We’ll use the Supervisor pattern - for process isolation and auto-recovery.

Create Agent Directory

mkdir my_first_agent
cd my_first_agent

Create callback.py:

# my_first_agent/callback.py
def greeter_callback(message: dict):
    """
    Agent logic runs here - in an isolated process!
    In production: LangGraph, CrewAI, AutoGen, etc.
    """
    content = message.get("content", {})
    name = content.get("name", "stranger")
    return {"reply": f"Hello, {name}! 👋"}

Create __init__.py:

touch __init__.py

Directory structure:

my-omnidaemon-project/
├── my_first_agent/
│   ├── __init__.py
│   └── callback.py
└── (agent_runner.py - next step)

Go back to parent:

cd ..

###Create Agent Runner Create agent_runner.py:

# agent_runner.py - Production-ready with Supervisor
import asyncio
import logging
import sys
from omnidaemon import OmniDaemonSDK, AgentConfig
from omnidaemon.agent_runner.agent_supervisor_runner import (
    create_supervisor_from_directory,
    shutdown_all_supervisors,
)

logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger(__name__)

sdk = OmniDaemonSDK()
_supervisor = None

async def get_supervisor():
    global _supervisor
    if _supervisor is None:
        _supervisor = await create_supervisor_from_directory(
            agent_name="greeter",
            agent_dir="./my_first_agent",
            callback_function="callback.greeter_callback",
        )
    return _supervisor

async def greeter_handler(message: dict):
    supervisor = await get_supervisor()
    return await supervisor.handle_event(message)

async def main():
    try:
        await get_supervisor()  # Pre-initialize
        
        await sdk.register_agent(
            agent_config=AgentConfig(
                topic="greet.user",
                callback=greeter_handler,
            )
        )
        
        await sdk.start()
        logger.info("🎧 Agent running. Press Ctrl+C to stop.")
        await asyncio.Event().wait()
        
    except (KeyboardInterrupt, asyncio.CancelledError):
        logger.info("Shutdown...")
    finally:
        await sdk.shutdown()
        await shutdown_all_supervisors()

if __name__ == "__main__":
    asyncio.run(main())

What this provides:

✅ Process isolation (like containers)
✅ Auto-restart on crash
✅ Production-ready for Python agents
✅ Fault containment

📚 Working Examples

Want to see complete, production-ready implementations? Check these out:

Multiple Supervised Agents

examples/agents_with_supervisors/ Shows real-world patterns:

Multiple agents with supervisors in one runner
Google ADK with MCP filesystem tools
Proper initialization and graceful shutdown

Key files to study:

supervisor_agent_runners.py - Runner setup
google_adk_with_supervisor/callback.py - Agent logic
google_adk_with_supervisor/adk_agent.py - ADK setup

💡 These examples are your best teacher. Copy their patterns for production use.

Development Alternative (Simple)

For quick prototyping only (no isolation):

# agent_runner_simple.py - DEV ONLY
import asyncio
from omnidaemon import OmniDaemonSDK, AgentConfig

sdk = OmniDaemonSDK()

async def greeter(message: dict):
    name = message.get("content", {}).get("name", "stranger")
    return {"reply": f"Hello, {name}!"}

async def main():
    await sdk.register_agent(
        agent_config=AgentConfig(topic="greet.user", callback=greeter)
    )
    await sdk.start()
    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        await sdk.shutdown()

if __name__ == "__main__":
    asyncio.run(main())

⚠️ Limitations: No isolation, no auto-restart. Use Supervisor for production.

Step 4: Run Your Agent

python agent_runner.py

✅ Expected output:

[Runner abc-123] Registered agent 'greeter-agent' on topic 'greet.user'
🎧 Agent running. Press Ctrl+C to stop.

Success indicators:

Shows “Registered agent” message
Shows “Agent running” message
Process doesn’t exit (stays running, waiting for messages)

❌ Common errors and fixes:

Error	Cause	Fix
`Connection refused [Errno 111]`	Event bus not running	Go back to Step 1, start event bus backend
`ModuleNotFoundError: No module named 'omnidaemon'`	Not installed	Go back to Step 2
`ImportError: cannot import name 'OmniDaemonSDK'`	Wrong import	Try `from omnidaemon import OmniDaemonSDK`

Keep this terminal running - your agent is now alive and listening!

Step 5: Publish an Event

Open a NEW terminal (keep the agent running in the first one!) and create publisher.py:

SIMPLE VERSION (Required parameters only)

# publisher.py - SIMPLE VERSION
import asyncio
from omnidaemon import OmniDaemonSDK, EventEnvelope, PayloadBase

sdk = OmniDaemonSDK()

async def main():
    # SIMPLE: Only topic and content required!
    event = EventEnvelope(
        topic="greet.user",              # REQUIRED
        payload=PayloadBase(
            content={"name": "Alice"}     # REQUIRED
        ),
    )

    task_id = await sdk.publish_task(event_envelope=event)
    print(f"📨 Task ID: {task_id}")

    # Wait and get result
    await asyncio.sleep(2)
    result = await sdk.get_result(task_id)
    print(f"✅ Result: {result}")

asyncio.run(main())

FULL VERSION (With all optional parameters)

# publisher.py - FULL VERSION
import asyncio
from omnidaemon import OmniDaemonSDK, EventEnvelope, PayloadBase

sdk = OmniDaemonSDK()

async def main():
    event = EventEnvelope(
        topic="greet.user",                     # REQUIRED
        payload=PayloadBase(
            content={"name": "Alice"},           # REQUIRED
            webhook="https://api.example.com/callback",  # Optional: HTTP callback
            reply_to="greet.response",           # Optional: Response topic
            correlation_id="req-12345",          # Optional: Track requests
            causation_id="event-67890",          # Optional: Event chain
            source="web-app",                    # Optional: Event origin
            tenant_id="tenant-abc",              # Optional: Multi-tenancy
        ),
    )

    task_id = await sdk.publish_task(event_envelope=event)
    print(f"📨 Task ID: {task_id}")
    print(f"   Correlation ID: {event.payload.correlation_id}")
    print(f"   Source: {event.payload.source}")

    # Results auto-expire after 24 hours
    await asyncio.sleep(2)
    result = await sdk.get_result(task_id)
    print(f"✅ Result: {result}")

asyncio.run(main())

Parameter Defaults:

webhook: None (no HTTP callback)
reply_to: None (no response topic)
correlation_id: Auto-generated UUID
causation_id: None
source: “unknown”
tenant_id: “default”

Run it:

python publisher.py

✅ Expected output:

📨 Task ID: 550e8400-e29b-41d4-a716-446655440000
✅ Result: {'reply': 'Hello, Alice! 👋', 'status': 'success', 'task_id': '...'}

In the agent terminal, you should see:

[Agent greeter-agent] Received message on topic greet.user
[Agent greeter-agent] Processed task in 0.05s

Step 6: Check System Health

In a new terminal:

omnidaemon health

✅ Expected output:

🏥 System Health Check
==================================================
Status: running
Registered Agents: 1
Subscribed Topics: ['greet.user']
Event Bus: RedisStreamEventBus (Pluggable - using Redis Streams)
Storage: Healthy (Pluggable - using Redis)
==================================================
✅ All systems operational!

🎉 Success! What Just Happened?

You now have a fully functional event-driven AI agent runtime:

✅ Event Bus - Running and handling message distribution (using Redis Streams)
✅ Storage Backend - Persisting agents, results, and metrics (using Redis)
✅ OmniDaemon - Installed and operational
✅ Agent - Registered and listening for events
✅ Event Flow - Published task → Agent processed → Result stored
✅ Health Check - All systems verified

The Event Flow:

Publisher (you)
   │
   ├─► Publishes to topic "greet.user"
   │
   ▼
Event Bus (Redis Streams)
   │   (Pluggable: Kafka, RabbitMQ, NATS coming soon)
   │
   ├─► Notifies all subscribers
   │
   ▼
Your Agent (greeter)
   │
   ├─► Processes message
   ├─► Generates response
   │
   ▼
Storage Backend (Redis)
   │   (Pluggable: PostgreSQL, MongoDB, S3 coming soon)
   │
   └─► Stores result for retrieval

⚙️ Configuration (Optional)

The Quick Start uses smart defaults - you don’t need to configure anything! Defaults:

Storage Backend: JSON files in .omnidaemon_data/ (pluggable)
Event Bus: Redis Streams at localhost:6379 (pluggable)
API: Disabled (use SDK/CLI only)

To customize, create a .env file:

# .env
# Storage Backend (pluggable: json, redis, postgresql*, mongodb*, s3*)
STORAGE_BACKEND=redis              # Production: Use Redis for distributed storage
REDIS_URL=redis://localhost:6379   # Connection string for Redis backend

# Event Bus (pluggable: redis_stream, kafka*, rabbitmq*, nats*)
EVENT_BUS_TYPE=redis_stream        # Production-ready option (more coming soon)
REDIS_URL=redis://localhost:6379   # Connection string for Redis Streams

# API Server
OMNIDAEMON_API_ENABLED=true        # Enable HTTP API server
OMNIDAEMON_API_PORT=8765           # API port

# Logging
LOG_LEVEL=INFO                     # DEBUG for troubleshooting

# * = Coming soon

When to change defaults:

Setting	Change When…
`STORAGE_BACKEND=redis`	Production deployment, need distributed storage
`REDIS_URL=...`	Event bus or storage on different host/port
`OMNIDAEMON_API_ENABLED=true`	Want HTTP API access
`LOG_LEVEL=DEBUG`	Troubleshooting issues

For Quick Start: Stick with defaults! 👍

🐛 Quick Troubleshooting

Problem: “Event Bus connection keeps failing”

# For Redis Streams backend (default):
# Check if Redis is running
redis-cli ping

# Check Redis is on default port
redis-cli -p 6379 ping

# If using custom port, set it
export REDIS_URL=redis://localhost:6380
python agent_runner.py

# For other event bus backends (when available):
# Check EVENT_BUS_TYPE in your .env matches your running backend

Problem: “Agent runs but doesn’t process tasks”

# Verify agent registered
python -c "
import asyncio
from omnidaemon import OmniDaemonSDK
agents = asyncio.run(OmniDaemonSDK().list_agents())
print(f'Registered agents: {agents}')
"

# Check event bus streams (if using Redis Streams)
redis-cli XLEN omni-stream:greet.user

Problem: “No output when running agent”

This is normal! Agent runs in background. Look for:

✅ “Registered agent” message
✅ “Listening for topics” message
✅ No error messages

Problem: “Can’t import OmniDaemonSDK”

# Try alternative import
from omnidaemon import OmniDaemonSDK

# Or check if installed
pip list | grep omnidaemon

Still stuck? See Support & Community for help.

🚀 What’s Next?

Congratulations! You’ve built your first AI agent with OmniDaemon. Here’s where to go next:

Learn More

Core Concepts - Understand EDA deeply
Agent Lifecycle - Registration, subscription, deletion
Callback Pattern - Master the callback

Build Real Agents

Use OmniCore Agent - AI agent with MCP tools
Use Google ADK - Google’s Agent Development Kit
Common Patterns - 7 production-ready recipes

Go to Production

Production Setup - Deploy for real
Monitoring - Metrics, health, DLQ

Explore the API

Python SDK Reference - Complete API docs
CLI Reference - All CLI commands

📖 Need Help?

Happy building! 🎉

Get Started

Core Concepts

How-To Guides

Architecture

Enterprise

Community

Quick start

Quick Start: Build Your First Agent in 10 Minutes

Step 1: Install Event Bus & Storage Backend

macOS

Ubuntu/Debian

Windows

Docker (All Platforms - Easiest!)

Step 2: Install OmniDaemon

Using uv (Recommended - Fast!)

Using pip (Traditional)

Step 3: Create Your First Agent (Production-Ready)

Create Agent Directory

📚 Working Examples

Multiple Supervised Agents

Development Alternative (Simple)

Step 4: Run Your Agent

Step 5: Publish an Event

SIMPLE VERSION (Required parameters only)

FULL VERSION (With all optional parameters)

Step 6: Check System Health

🎉 Success! What Just Happened?

⚙️ Configuration (Optional)

🐛 Quick Troubleshooting

Problem: “Event Bus connection keeps failing”

Problem: “Agent runs but doesn’t process tasks”

Problem: “No output when running agent”

Problem: “Can’t import OmniDaemonSDK”

🚀 What’s Next?

Learn More

Build Real Agents

Go to Production

Explore the API

📖 Need Help?

Get Started

Core Concepts

How-To Guides

Architecture

Enterprise

Community

​Quick Start: Build Your First Agent in 10 Minutes

​Step 1: Install Event Bus & Storage Backend

​macOS

​Ubuntu/Debian

​Windows

​Docker (All Platforms - Easiest!)

​Step 2: Install OmniDaemon

​Using uv (Recommended - Fast!)

​Using pip (Traditional)

​Step 3: Create Your First Agent (Production-Ready)

​Create Agent Directory

​📚 Working Examples

​Multiple Supervised Agents

​Development Alternative (Simple)

​Step 4: Run Your Agent

​Step 5: Publish an Event

​SIMPLE VERSION (Required parameters only)

​FULL VERSION (With all optional parameters)

​Step 6: Check System Health

​🎉 Success! What Just Happened?

​⚙️ Configuration (Optional)

​🐛 Quick Troubleshooting

​Problem: “Event Bus connection keeps failing”

​Problem: “Agent runs but doesn’t process tasks”

​Problem: “No output when running agent”

​Problem: “Can’t import OmniDaemonSDK”

​🚀 What’s Next?

​Learn More

​Build Real Agents

​Go to Production

​Explore the API

​📖 Need Help?

Quick Start: Build Your First Agent in 10 Minutes

Step 1: Install Event Bus & Storage Backend

macOS

Ubuntu/Debian

Windows

Docker (All Platforms - Easiest!)

Step 2: Install OmniDaemon

Using uv (Recommended - Fast!)

Using pip (Traditional)

Step 3: Create Your First Agent (Production-Ready)

Create Agent Directory

📚 Working Examples

Multiple Supervised Agents

Development Alternative (Simple)

Step 4: Run Your Agent

Step 5: Publish an Event

SIMPLE VERSION (Required parameters only)

FULL VERSION (With all optional parameters)

Step 6: Check System Health

🎉 Success! What Just Happened?

⚙️ Configuration (Optional)

🐛 Quick Troubleshooting

Problem: “Event Bus connection keeps failing”

Problem: “Agent runs but doesn’t process tasks”

Problem: “No output when running agent”

Problem: “Can’t import OmniDaemonSDK”

🚀 What’s Next?

Learn More

Build Real Agents

Go to Production

Explore the API

📖 Need Help?