Sessions | Smallest AI Docs

The Session (AgentSession) is the container that holds your active conversation. It creates a sandbox for every user connection, managing state, resources, and the event loop.

One Session = One User

Every time a user connects via WebSocket (or phone), a new AgentSession instance is spawned. This ensures total isolation—variables set in one session never leak to another.

Core Capabilities

The Session is your primary interface for controlling the agent’s environment.

1. Graph Construction

You don’t just add nodes; you build a graph. The session provides methods to define exactly how events flow.

1 async def on_start(session: AgentSession):
2     # Add nodes to the sandbox
3     session.add_node(agent)
4     session.add_node(logger)
5     
6     # Define custom flow: Logger -> Agent
7     session.add_edge(logger, agent)
8     
9     await session.start()

2. Event Hooks

Power Feature: You can listen to events globally!

Use the @session.on_event decorator to inject logic without creating a dedicated node. This is perfect for simple side-effects like logging analytics or sending welcome messages.

1 @session.on_event("system.user_joined")
2 async def welcome(session, event):
3     print(f"User {event.session_id} connected!")
4 
5 @session.on_event("agent.speak")
6 async def track_speech(session, event):
7     metrics.increment("agent_speech_count")

3. Automatic Plumbing

The session automatically manages the entry and exit points of your graph.

Root Node: Automatically created. Receives events from the client and forwards them to your nodes.
Sink Node: Automatically created. Catches events from your nodes and sends them back to the client.

SDK Reference

Method	Description
`session.add_node(node)`	Registers a node in the graph.
`session.add_edge(parent, child)`	Connect two nodes explicitly.
`session.start()`	Locks the graph, validates no cycles exist, and starts the event loop.
`session.wait_until_complete()`	Keeps the process alive until the WebSocket disconnects.
`session.context`	A dict-like object for storing session-scoped state.

Cycle Detection

When you call start(), the session runs a validation algorithm. If your graph contains a cycle (A → B → A), it will raise a ValueError immediately to prevent infinite loops.

Example: Complete Setup

A full example showing graph building, hooks, and lifecycle management.

1 from smallestai.atoms.agent.session import AgentSession
2 from smallestai.atoms.agent.server import AtomsApp
3 
4 async def setup(session: AgentSession):
5     # 1. Initialize Nodes
6     agent = SalesAgent()
7     tracker = StatsTracker()
8     
9     # 2. Build Graph
10     session.add_node(tracker)
11     session.add_node(agent)
12     session.add_edge(tracker, agent)  # Events go Tracker -> Agent
13     
14     # 3. Register Hooks
15     @session.on_event("system.control.interrupt")
16     async def on_interrupt(session, event):
17         print("User interrupted the agent!")
18 
19     # 4. Launch
20     await session.start()
21     await session.wait_until_complete()
22 
23 if __name__ == "__main__":
24     app = AtomsApp(setup_handler=setup)
25     app.run()