Skip to main content
The SessionStoreProtocol defines the methods any session backend must implement — swap JSON files, Redis, or your own store without changing agent code.
from praisonaiagents import Agent

agent = Agent(
    name="Assistant",
    memory={"session_id": "chat-123"},
)
agent.start("Hello!")
The user plugs in a custom session backend; agents call the same protocol methods regardless of storage.

How It Works

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
set_chat_history(session_id, messages)boolRecommended. Atomically replace chat history for idempotent saves
update_session_metadata(session_id, **fields)boolMerge metadata fields without touching messages
get_working_history(session_id, max_messages=None)list[dict]Optional. Compacted resume path — summary + tail when a checkpoint exists, else raw history
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.
Backward compatible resume. On resume the agent prefers get_working_history(...) when the store implements it (checked via hasattr), so it can replay a compacted summary + tail. Stores that implement only get_chat_history continue to work unchanged — they simply resume from the raw transcript. See Compacted Session Resume.
The protocol uses Python’s typing.Protocol with @runtime_checkable, so any class with matching method signatures automatically satisfies it — no inheritance needed.
set_chat_history(session_id, messages) (recommended). Atomically replaces the chat history for a session. If your store omits this method, Session.save_state() falls back to add_message() per message and may produce duplicates when called repeatedly. The built-in DefaultSessionStore implements this with reload-under-lock atomic-write semantics.
class MyStore:
    def set_chat_history(self, session_id: str, messages: list[dict]) -> bool:
        """Replace the chat history for a session atomically.
        
        Args:
            session_id: Session identifier
            messages: list of {"role": str, "content": str}
            
        Returns:
            True if successful
        """
        with self._lock:
            self._data[session_id] = list(messages)
            self._flush()
        return True
If your custom store backs onto shared storage (Redis, Postgres, S3), reload from that backing store on every read — do not rely on in-process caches. DefaultSessionStore and BotSessionManager._load_history assume this.

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() reloads under a cross-process file lock so concurrent metadata updates and message appends are both preserved.

Built-in Implementations

DefaultSessionStore

JSON file-based persistence with atomic writes and file locking. Mutating and read methods reload from disk under lock so multiple processes can share one directory.
from praisonaiagents.session import DefaultSessionStore

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

HierarchicalSessionStore

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

store = get_hierarchical_session_store()
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 get_hierarchical_session_store

store = get_hierarchical_session_store()

snapshot_id = store.create_snapshot("session-123", label="Before complex task")
store.revert_to_snapshot("session-123", snapshot_id)

Best Practices

Without it, repeated Session.save_state() calls may duplicate messages via the add_message fallback.
Redis, Postgres, and multi-worker file stores must not serve stale in-process caches.
SessionStoreProtocol is @runtime_checkable — verify custom stores with isinstance(store, SessionStoreProtocol).

Session Persistence

Automatic session save and restore

Sessions Overview

Sessions and remote agents