Sessions | Smallest AI Docs

The Session (CrewSession) 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 CrewSession 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: CrewSession):
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.crew.session import CrewSession
2 from smallestai.atoms.crew.server import AtomsCrewApp
3 
4 async def setup(session: CrewSession):
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 = AtomsCrewApp(setup_handler=setup)
25     app.run()