***

title: Events
description: 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

<Steps>
  <Step title="Define the Event">
    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"
    ```
  </Step>

  <Step title="Emit the Event">
    Use `self.send_event()` to broadcast the event to downstream nodes.

    ```python
    class SupportBot(OutputAgentNode):
        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"
                ))
    ```
  </Step>

  <Step title="Handle the Event">
    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)
    ```
  </Step>
</Steps>

***

## Best Practices

<AccordionGroup>
  <Accordion title="Enforce type safety (Pydantic)" icon="shield-check">
    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!
    ```
  </Accordion>

  <Accordion title="Don't break the chain" icon="link-slash">
    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)
    ```
  </Accordion>

  <Accordion title="Use unique event types" icon="fingerprint">
    When defining custom events, use a namespaced type string like `myapp.order.created` to avoid collisions with future system events.
  </Accordion>
</AccordionGroup>