IntroductionSDK

Graph

View as MarkdownOpen in Claude

A Graph defines how nodes are connected. Events flow through the graph from parent nodes to child nodes, creating a processing pipeline for your agent logic.

The Atoms graph is a DAG (Directed Acyclic Graph). Events can flow through multiple branches, but never in circles.

How Graphs Work

When you call session.add_edge(parent, child), you’re creating a connection. Events emitted by the parent via send_event() are automatically queued for the child.

User Input → [Root] → [Your Nodes] → [Sink] → User Output

The session automatically creates two special nodes:

  • Root: Entry point—receives events from the WebSocket
  • Sink: Exit point—sends events back to the WebSocket

Building a Graph

Step 1: Add Nodes

1async def setup(session: AgentSession):
2    logger = LoggerNode()
3    agent = SalesAgent()
4    analytics = AnalyticsNode()
5    
6    session.add_node(logger)
7    session.add_node(agent)
8    session.add_node(analytics)

session.add_node(node): Registers a Node instance with the session. The node must inherit from the base Node class.

Step 2: Connect with Edges

1    # Define the flow: Logger → Agent → Analytics
2    session.add_edge(logger, agent)
3    session.add_edge(agent, analytics)
4    
5    await session.start()

The Resulting Graph

[Root] → [Logger] → [Agent] → [Analytics] → [Sink]

Graph Patterns

1# [Root] → [A] → [B] → [C] → [Sink]
2# The simplest pattern—events flow sequentially.
3
4session.add_edge(node_a, node_b)
5session.add_edge(node_b, node_c)

Automatic Connections

Nodes without explicit parents connect to Root. Nodes without explicit children connect to Sink.

1# Just add one node with no edges:
2session.add_node(my_agent)
3
4# Automatically becomes:
5# [Root] -> [my_agent] -> [Sink]

This means a minimal agent only needs:

1async def setup(session: AgentSession):
2    session.add_node(SalesAgent())
3    await session.start()
4    await session.wait_until_complete()

Cycle Detection

Graphs must not contain cycles. The session validates this at startup:

1# This will FAIL
2session.add_edge(node_a, node_b)
3session.add_edge(node_b, node_c)
4session.add_edge(node_c, node_a)  # Creates a cycle!
5
6await session.start()  
7# Raises: ValueError("Graph contains cycles")

Cycles would cause infinite event loops. The framework prevents this at startup, but design your graphs as DAGs from the start.

Event Flow in Detail

When a node calls send_event():

  1. The event is queued for each child node
  2. Each child’s process_event() is called asynchronously
  3. Children can further propagate via their own send_event()
1class ParentNode(Node):
2    async def process_event(self, event):
3        # Modify the event if needed
4        event.metadata["processed_by"] = self.name
5        
6        # Queue for all children
7        await self.send_event(event)

Each node has its own event queue. Multiple events can be queued while a node is processing, and they’ll be handled in order.

Custom Routing

For dynamic routing, don’t use send_event()—directly queue to specific children:

1class RouterNode(Node):
2    def __init__(self, sales_node, support_node):
3        super().__init__(name="router")
4        self.sales = sales_node
5        self.support = support_node
6    
7    async def process_event(self, event):
8        intent = self.classify(event)
9        
10        if intent == "sales":
11            await self.sales.queue_event(event)
12        elif intent == "support":
13            await self.support.queue_event(event)
14        else:
15            # Fallback: send to all children normally
16            await self.send_event(event)

Best Practices

Deeply nested graphs increase latency. Events have to hop through every node.

Bad: A -> B -> C -> D -> E (5 hops) Good: Router -> [A, B, C, D, E] (2 hops)

You will thank yourself when reading logs.

1# Good
2LoggerNode(name="input_logger")
3
4# Bad
5LoggerNode(name="log1")

If a node is sending to more than 3 children, it’s usually better to have a dedicated Router node that decides where the event goes, rather than broadcasting to everyone.

Graphs can get complex. Sketching the flow on paper (or Excalidraw) before coding saves a lot of headaches.