Session Store Protocol

Swap session backends with a unified protocol interface

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:

set_chat_history (Recommended)

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

update_session_metadata() API

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

Parameters

Reserved Metadata Fields

The SDK uses these reserved fields internally:

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

from praisonaiagents.session import DefaultSessionStore

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

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)

from praisonaiagents.session import SqliteSessionStore

store = SqliteSessionStore(db_path="~/.praisonai/sessions.db")

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)

Portable Store Protocol

PortableSessionStoreProtocol adds store-agnostic backup / migrate / restore alongside SessionStoreProtocol, SearchableSessionStoreProtocol, and SessionMirrorProtocol. DefaultSessionStore and SqliteSessionStore implement it.

from praisonaiagents.session import DefaultSessionStore, PortableSessionStoreProtocol, ImportReport

store = DefaultSessionStore()
assert isinstance(store, PortableSessionStoreProtocol)  # runtime_checkable

payload = store.export_all()
report: ImportReport = store.import_sessions(payload)

ImportReport is JSON-serialisable via as_dict():

{
  "imported": 3,
  "skipped": [{"session_id": "s1", "reason": "already exists (use overwrite)"}],
  "skipped_count": 1,
  "version": 1
}

PortableSessionStoreProtocol is @runtime_checkable, so isinstance(store, PortableSessionStoreProtocol) verifies conformance. See Gateway Session Portability for the operator-facing story.

Incremental Session Store Protocol

IncrementalSessionStoreProtocol is an optional protocol for incremental streamed-turn persistence.

Without it, an interrupted stream loses the whole in-progress turn — praisonai run --continue / --session <id> resumed from the last completed turn. With it, the agent flushes the growing partial assistant turn as tokens arrive, so resume shows whatever was produced before the interrupt. DefaultSessionStore, SqliteSessionStore, and EncryptedSessionStore implement it.

from praisonaiagents.session import (
    DefaultSessionStore,
    IncrementalSessionStoreProtocol,
)

store = DefaultSessionStore()
assert isinstance(store, IncrementalSessionStoreProtocol)  # runtime_checkable

store.upsert_partial_assistant_message("s1", "Hello")
store.upsert_partial_assistant_message("s1", "Hello world")
store.upsert_partial_assistant_message("s1", "Hello world", finalize=True)

Best Practices

Related