Events

Events are typed objects that flow through the graph. They represent everything that happens in the system, from a user saying “Hello” to an agent deciding to call a function.

Flow

Understanding how events move is key to understanding Atoms:

1. Transport Layer: Receives WebSocket message → Converts to SDKEvent.
2. Session: Routes event to Root Nodes.
3. Nodes: Process event → Emit new events to Children.
4. Sink: Catches output events → Converts to WebSocket message → Sends to Client.

Structure

All events inherit from SDKEvent, which is a Pydantic model (TypedModel). This ensures type safety and easy serialization.

1
class SDKEvent(TypedModel):
2
    id: str         # unique identifier for the event
3
    type: str       # specific event type string (e.g., "agent.speak")
4
    timestamp: datetime = Field(default_factory=datetime.now)
5
    metadata: Dict  # optional payload for custom data

Common Events

Custom Events

Signaling between agents

While standard system events handle core mechanics like text and audio, you often need to signal custom business logic states—like PaymentFailed or EscalationRequired.

Why Custom Events? Unlike passing untyped dictionaries, defining SDKEvent subclasses gives you type safety, validation, and clearer intent in your code.

Implementation

1
# 1. Define the Event
2
class EscalationEvent(SDKEvent, type="escalation_event"):
3
    reason: str
4
    severity: Optional[str] = "medium"

1
class SupportBot(OutputAgentNode):
2
    async def generate_response(self, messages):
3
        # ... analysis logic ...
4
        if "angry" in user_sentiment:
5
            # Emit the custom event
6
            await self.send_event(EscalationEvent(
7
                reason="User is frustrated",
8
                severity="high"
9
            ))

1
class SupervisorNode(Node):
2
    async def process_event(self, event: SDKEvent):
3
        # Check for your custom event type
4
        if isinstance(event, EscalationEvent):
5
            print(f"Escalation received: {event.reason}")
6
            # Take action...
7
        
8
        # Important: Propagate event to children!
9
        await self.send_event(event)

Best Practices

1
# Bad: Accessing dict keys blindly
2
if event.get("type") == "message":
3
    print(event["content"])
4
    
5
# Good: Using the model
6
if isinstance(event, SDKSystemLLMRequestEvent):
7
    print(event.content)  # IDE autocompletes this!

1
async def process_event(self, event):
2
    if isinstance(event, MyType):
3
        await handle(event)
4
    
5
    # ALWAYS do this
6
    await self.send_event(event)