Skip to main content
The SessionStoreProtocol defines five methods that any session backend must implement. This lets you swap between JSON files, Redis, MongoDB, or your own custom store — without changing any agent or bot code.

Quick Start

1

Use the built-in JSON store (zero config)

from praisonaiagents import Agent

agent = Agent(
    name="Assistant",
    memory={"session_id": "chat-123"}
)
agent.start("Hello!")
Sessions are automatically persisted to ~/.praisonai/sessions/chat-123.json.
2

Swap to a custom store

from praisonaiagents.session import SessionStoreProtocol

class RedisSessionStore:
    """Your custom Redis-backed session store."""

    def add_message(self, session_id, role, content, metadata=None):
        # Save to Redis
        ...
        return True

    def get_chat_history(self, session_id, max_messages=None):
        # Load from Redis
        return [{"role": "user", "content": "Hi"}]

    def clear_session(self, session_id):
        return True

    def delete_session(self, session_id):
        return True

    def session_exists(self, session_id):
        return False

# Verify at runtime
store = RedisSessionStore()
assert isinstance(store, SessionStoreProtocol)  # ✅ True

Protocol Methods

The protocol requires exactly five methods:
MethodReturnsPurpose
add_message(session_id, role, content, metadata)boolStore a single message
get_chat_history(session_id, max_messages)list[dict]Retrieve messages in LLM format
clear_session(session_id)boolRemove all messages (keep metadata)
delete_session(session_id)boolDelete session completely
session_exists(session_id)boolCheck if a session exists
update_session_metadata(session_id, **fields)boolMerge metadata fields without touching messages
For persistent backends (DB / JSON), clear_session() and delete_session() remove the underlying stored messages too, not just in-memory data. This ensures cleared history does not reappear after a reload or restart.
The protocol uses Python’s typing.Protocol with @runtime_checkable, so any class with matching method signatures automatically satisfies it — no inheritance needed.

update_session_metadata() API

The update_session_metadata() method enables safe metadata updates across processes without touching messages:

Parameters

ParameterTypeDescription
session_idstrUnique session identifier
**fieldsAnyMetadata fields to merge (e.g., model, total_tokens, cost)

Reserved Metadata Fields

The SDK uses these reserved fields internally:
FieldTypeDescriptionAuto-populated
modelstrLLM model name (e.g., “gpt-4o-mini”)
total_tokensintCumulative input + output tokens
costfloatEstimated USD cost
agent_idstrGateway or registry agent ID
sourcestrOrigin: “chat”, “gateway”, “cli”, “api”

Example Usage

from praisonaiagents.session import get_default_session_store

store = get_default_session_store()

# Merge metadata into an existing session without touching messages
store.update_session_metadata(
    "user-123-chat",
    model="gpt-4o-mini",
    total_tokens=125,
    cost=0.0032,
    source="chat",
    agent_id="assistant-001",
)

# None values are skipped
store.update_session_metadata(
    "user-123-chat",
    cost=None,  # Ignored, doesn't overwrite existing cost
    custom_field="my value"
)

Concurrency Safety

In DefaultSessionStore, update_session_metadata() uses a cross-process file lock with locked reload-from-disk to ensure concurrent metadata updates and message appends are both preserved. This fixed a silent message-loss bug that affected multi-worker deployments before PR #1709.
If your custom store backs onto shared storage (Redis, Postgres, S3, another file system), make get_chat_history and get_session read from that backing store on every call rather than caching in process memory. DefaultSessionStore guarantees this; downstream code (e.g. BotSessionManager._load_history) relies on it.

Built-in Implementations

DefaultSessionStore

JSON file-based persistence with atomic writes and file locking. Zero dependencies. All mutating methods reload from disk inside the file lock, so multiple processes can safely share one session directory.
from praisonaiagents.session import DefaultSessionStore

store = DefaultSessionStore(
    session_dir="/custom/path",
    max_messages=200,
)

HierarchicalSessionStore

Extends DefaultSessionStore with session forking, snapshots, and revert.
from praisonaiagents.session import HierarchicalSessionStore

store = HierarchicalSessionStore()
parent = store.create_session(title="Main")
child = store.fork_session(parent, from_message_index=5)

Using with Bots

Bots use the same SessionStoreProtocol for persistent per-user sessions:
from praisonaiagents.session import get_default_session_store

# Bot sessions persist automatically
from praisonai.bots import Bot

bot = Bot(
    "telegram",
    agent=my_agent,
    session_store=get_default_session_store(),
)
Each user gets a deterministic session key like bot_telegram_12345, stored in the same ~/.praisonai/sessions/ directory as agent sessions.

Building a Custom Store

from typing import Any, Dict, List, Optional

class InMemorySessionStore:
    """Fast in-memory store for unit tests."""

    def __init__(self):
        self._data: Dict[str, List[Dict]] = {}

    def add_message(
        self, session_id: str, role: str, content: str,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> bool:
        self._data.setdefault(session_id, []).append(
            {"role": role, "content": content}
        )
        return True

    def get_chat_history(
        self, session_id: str, max_messages: Optional[int] = None,
    ) -> List[Dict[str, str]]:
        msgs = self._data.get(session_id, [])
        return msgs[-max_messages:] if max_messages else list(msgs)

    def clear_session(self, session_id: str) -> bool:
        self._data[session_id] = []
        return True

    def delete_session(self, session_id: str) -> bool:
        self._data.pop(session_id, None)
        return True

    def session_exists(self, session_id: str) -> bool:
        return session_id in self._data
from praisonaiagents.session import SessionStoreProtocol

store = InMemorySessionStore()

# Runtime check — no registration needed
assert isinstance(store, SessionStoreProtocol)

# Use it anywhere a SessionStoreProtocol is expected
store.add_message("test", "user", "Hello")
history = store.get_chat_history("test")
assert history == [{"role": "user", "content": "Hello"}]

Checkpoint Query Protocol

For session stores that support checkpoints and rollback functionality:
from praisonaiagents.session.protocols import CheckpointQueryProtocol

class CheckpointEnabledStore:
    """Session store with checkpoint support."""
    
    def list_checkpoints(self, session_id: str) -> List[Dict[str, Any]]:
        """Return checkpoint metadata for a session."""
        return [
            {
                "checkpoint_id": "checkpoint_001",
                "created_at": "2026-01-01T12:00:00Z",
                "message_count": 10,
                "description": "Before complex task"
            }
        ]
    
    def get_checkpoint(self, session_id: str, checkpoint_id: str) -> Optional[Dict[str, Any]]:
        """Return full checkpoint data."""
        return {
            "checkpoint_id": checkpoint_id,
            "session_id": session_id,
            "messages": [...],  # Full message history at checkpoint
            "metadata": {...}
        }

# Verify checkpoint protocol
store = CheckpointEnabledStore()
assert isinstance(store, CheckpointQueryProtocol)

Checkpoint Usage

from praisonaiagents.session import HierarchicalSessionStore

# Built-in checkpoint support
store = HierarchicalSessionStore()

# Create snapshot before risky operation
checkpoint_id = store.snapshot_session("session-123", "Before complex task")

# List available checkpoints
checkpoints = store.list_checkpoints("session-123")
for cp in checkpoints:
    print(f"Checkpoint {cp['checkpoint_id']}: {cp['description']}")

# Restore from checkpoint if needed
store.restore_session("session-123", checkpoint_id)

Architecture Overview

Next Steps

Learn about session persistence details

Sessions & remote agents overview