Sessions

Sessions let you persist an agent's conversation history and state across process restarts, serverless invocations, or separate user requests.

File-based sessions

Store sessions on the local filesystem:

from elsai import Agent
from elsai.session import FileSessionManager

session_manager = FileSessionManager(storage_dir="./sessions")

agent = Agent(
    agent_id="user-alice",   # Unique identifier for this session
    session_manager=session_manager,
)

agent("My name is Alice.")
# session saved to ./sessions/user-alice/

# Later (even in a new process):
agent2 = Agent(
    agent_id="user-alice",
    session_manager=session_manager,
)
result = agent2("What is my name?")
print(result)  # → "Your name is Alice."

S3-based sessions

Store sessions in Amazon S3 for serverless or distributed deployments:

from elsai import Agent
from elsai.session import S3SessionManager

session_manager = S3SessionManager(
    bucket_name="my-agent-sessions",
    prefix="sessions/",         # Optional prefix
    region_name="us-west-2",
)

agent = Agent(
    agent_id="user-bob",
    session_manager=session_manager,
)

Repository-based sessions

For database-backed sessions, implement SessionRepository:

from elsai.session import RepositorySessionManager, SessionRepository

class PostgresSessionRepository(SessionRepository):
    def save(self, session_id: str, data: dict) -> None:
        # Write to PostgreSQL
        ...

    def load(self, session_id: str) -> dict | None:
        # Read from PostgreSQL
        ...

    def delete(self, session_id: str) -> None:
        ...

manager = RepositorySessionManager(repository=PostgresSessionRepository())
agent = Agent(agent_id="user-123", session_manager=manager)

File storage structure

When using FileSessionManager, sessions are stored in this directory layout:

<storage_dir>/
└── session_<session_id>/
    ├── session.json                  ← session metadata & timestamps
    ├── agents/
    │   └── agent_<agent_id>/
    │       ├── agent.json            ← agent state & configuration
    │       └── messages/
    │           ├── message_<id>.json ← individual conversation messages
    │           └── message_<id>.json
    └── multi_agents/
        └── multi_agent_<id>/
            └── multi_agent.json      ← orchestrator state & node history

Each message_<id>.json holds a single conversation turn:

{
  "role": "user",
  "content": [{ "text": "What is 2 + 2?" }],
  "created_at": "2025-06-01T12:00:00Z"
}

S3 storage structure

S3SessionManager mirrors the same hierarchy in S3:

<prefix>/
└── session_<session_id>/
    ├── session.json
    ├── agents/
    │   └── agent_<agent_id>/
    │       ├── agent.json
    │       └── messages/
    │           └── message_<id>.json
    └── multi_agents/
        └── multi_agent_<id>/
            └── multi_agent.json

Required IAM permissions:

{
  "Effect": "Allow",
  "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject", "s3:ListBucket"],
  "Resource": ["arn:aws:s3:::my-bucket/*", "arn:aws:s3:::my-bucket"]
}

SessionManager interface

The SessionManager is a plugin — it implements the Plugin interface for single agents and MultiAgentPlugin for orchestrators.

Property	Description
session_id	Unique identifier for the session
storage_dir / bucket	Storage backend location
save_latest_on	When to persist: "message" (after each turn) or "invocation" (after full run)

What gets persisted

By default sessions save:

• Conversation messages (agent.messages)
• Agent state (agent.state)
• Conversation manager state

Snapshots (in-memory)

Take a point-in-time snapshot without a session manager:

# Save state
snapshot = agent.take_snapshot(preset="session")

# Serialise if needed
import json
snapshot_dict = snapshot.model_dump()

# Restore later
from elsai.types._snapshot import Snapshot

agent2 = Agent(agent_id="user-alice")
agent2.load_snapshot(Snapshot(**snapshot_dict))

Snapshot fields

Field	Description
messages	Full conversation history
state	Agent state key-value pairs
conversation_manager_state	Conversation manager internals
interrupt_state	Interrupt/resume state
system_prompt	System prompt content blocks

# Selective snapshot
snapshot = agent.take_snapshot(include=["messages", "state"])
snapshot = agent.take_snapshot(exclude=["interrupt_state"])

Custom data in snapshots

snapshot = agent.take_snapshot(
    preset="session",
    app_data={"user_tier": "premium", "feature_flags": ["beta_tools"]},
)
print(snapshot.app_data["user_tier"])  # "premium"

Multi-user example

from elsai import Agent
from elsai.session import FileSessionManager

session_manager = FileSessionManager("./sessions")

def get_agent_for_user(user_id: str) -> Agent:
    return Agent(
        agent_id=user_id,
        session_manager=session_manager,
        system_prompt="You are a personal assistant for this user.",
    )

# Each user gets their own isolated conversation
alice_agent = get_agent_for_user("alice")
bob_agent = get_agent_for_user("bob")

alice_agent("My favourite colour is blue.")
bob_agent("I love Python programming.")

# Later:
alice_agent2 = get_agent_for_user("alice")
print(alice_agent2("What is my favourite colour?"))  # blue