Event System
Real-time event streaming with runtime switching. Track every tool call, agent thought, and message in real-time.
Quick Start
from omnicoreagent import OmniCoreAgent, EventRouter
# Start with in-memory
agent = OmniCoreAgent(
event_router=EventRouter("in_memory"),
...
)
# Switch to Redis Streams for production
await agent.switch_event_store("redis_stream")
# Check current event store type
await agent.get_event_store_type()
# Stream events in real-time
async for event in agent.stream_events(session_id):
print(f"{event.type}: {event.payload}")
Event Types
| Event | Description |
|---|
user_message | User input received |
agent_message | Agent response generated |
agent_thought | Agent’s internal reasoning step |
tool_call_started | Tool execution begins |
tool_call_result | Tool execution completes |
final_answer | Agent produces final response |
sub_agent_started | Sub-agent delegation begins |
sub_agent_result | Sub-agent returns result |
sub_agent_error | Sub-agent encountered an error |
Retrieving Events
# Get event history for a session
events = await agent.get_events(session_id="user_1")
for event in events:
print(f"[{event.timestamp}] {event.type}: {event.payload}")
Enable events when you need real-time monitoring, debugging, or building UIs that show agent progress. Essential for production observability.