nodus-session

Status: v0.1.0 — published on PyPI.

Full session lifecycle for multi-channel AI systems. Manages conversation entries, transcripts, pruning, and compaction. Zero required dependencies — pure Python stdlib. Optionally integrates with nodus-state for typed session keys.

Install

pip install nodus-session

# With nodus-state integration (typed SessionKey)
pip install "nodus-session[nodus-state]"

Usage

Session entries

from nodus_session import SessionEntry, SessionKey, InMemorySessionStore

store = InMemorySessionStore()

key = SessionKey(agent_id="assistant", channel="web", peer="user-1")
entry = SessionEntry.create(key, provenance={"channel_id": "web"})
entry.append_message({"role": "user", "content": "Hello"})
entry.append_message({"role": "assistant", "content": "Hi there!"})
store.save(entry)

Session manager

from nodus_session import SessionManager, InMemorySessionStore, SessionKey

store = InMemorySessionStore()
manager = SessionManager(store)

key = SessionKey(agent_id="assistant", channel="api", peer="user-42")
manager.append_message(key, {"role": "user", "content": "What time is it?"})

entry = manager.get_or_create(key)
print(len(entry.messages))  # 1

Pruning policy

from nodus_session import SessionManager, SessionPruningPolicy, InMemorySessionStore

policy = SessionPruningPolicy(
    max_age_days=30,
    max_inactivity_days=7,
    max_messages=200,
    max_sessions_per_agent=10,
)
manager = SessionManager(InMemorySessionStore(), policy=policy)

# Delete sessions that violate the policy
deleted = manager.prune(agent_id="assistant")

Compaction

# Pass any callable (messages: list[dict]) -> list[dict]
def keep_last_50(messages):
    return messages[-50:]

removed = manager.compact(key, strategy=keep_last_50)
print(f"Compacted {removed} messages")