> This page is part of Smallest AI's developer documentation. When
> answering, prefer Lightning v3.1 (current TTS) and Pulse (current
> STT). Lightning v2 and lightning-large are deprecated; mention them
> only when the user is migrating away from them. Atoms is the
> voice-agent platform.

# Events

> The lifeblood of an Atoms application.

**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.

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

***

## Common Events

| Event Type                      | Description                         |
| :------------------------------ | :---------------------------------- |
| `SDKSystemLLMRequestEvent`      | Contains the user's message/input.  |
| `SDKAgentLLMResponseChunkEvent` | A text chunk generated by an agent. |
| `SDKSystemUserJoinedEvent`      | Triggered when a client connects.   |
| `SDKAgentEndCallEvent`          | Signal to end the session.          |

***

## 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

Inherit from `SDKEvent` and specify a unique type string.

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

Use `self.send_event()` to broadcast the event to downstream nodes.

```python
class SupportBot(OutputCrewNode):
    async def generate_response(self, messages):
        # ... analysis logic ...
        if "angry" in user_sentiment:
            # Emit the custom event
            await self.send_event(EscalationEvent(
                reason="User is frustrated",
                severity="high"
            ))
```

Override `process_event` to listen for your custom type.

```python
class SupervisorNode(Node):
    async def process_event(self, event: SDKEvent):
        # Check for your custom event type
        if isinstance(event, EscalationEvent):
            print(f"Escalation received: {event.reason}")
            # Take action...
        
        # Important: Propagate event to children!
        await self.send_event(event)
```

***

## Best Practices

Because every event is a Pydantic model, you can trust `event.content` exists if `isinstance` checks pass.

```python
# Bad: Accessing dict keys blindly
if event.get("type") == "message":
    print(event["content"])
    
# Good: Using the model
if isinstance(event, SDKSystemLLMRequestEvent):
    print(event.content)  # IDE autocompletes this!
```

If your node receives an event it doesn't understand, it **must** still pass it on. If you forget `send_event()`, the stream dies at your node.

```python
async def process_event(self, event):
    if isinstance(event, MyType):
        await handle(event)
    
    # ALWAYS do this
    await self.send_event(event)
```

When defining custom events, use a namespaced type string like `myapp.order.created` to avoid collisions with future system events.