`mnesis.session`¶

session ¶

Mnesis Session — the primary public API entry point.

MnesisSession ¶

MnesisSession(
    session_id: str,
    model: str,
    model_info: ModelInfo,
    config: MnesisConfig,
    system_prompt: str,
    agent: str,
    store: ImmutableStore,
    dag_store: SummaryDAGStore,
    context_builder: ContextBuilder,
    compaction_engine: CompactionEngine,
    token_estimator: TokenEstimator,
    event_bus: EventBus,
)

A single Mnesis session managing a conversation with an LLM.

Handles the complete lifecycle: message persistence, context assembly, streaming, compaction, and cleanup.

Usage::

# Preferred: open() returns an async context manager directly
async with MnesisSession.open(model="anthropic/claude-opus-4-6") as session:
    result = await session.send("Hello!")
    print(result.text)

# Alternative: create() + async with (equivalent, for explicit lifecycle)
async with await MnesisSession.create(model="anthropic/claude-opus-4-6") as session:
    result = await session.send("Hello!")

# Manual lifecycle (no context manager)
session = await MnesisSession.create(model="anthropic/claude-opus-4-6")
result = await session.send("Hello!")
await session.close()

BYO-LLM (bring-your-own LLM client)

If you manage LLM calls yourself (Anthropic SDK, OpenAI SDK, Gemini, etc.) and only want Mnesis for memory management, use :meth:record together with :meth:context_for_next_turn::

session = await MnesisSession.create(model="anthropic/claude-opus-4-6")

# Get the compaction-aware context window for your LLM call
messages = await session.context_for_next_turn()
response = your_client.chat(messages=messages, system=your_system_prompt)

# Persist the completed turn so Mnesis tracks tokens and compaction
await session.record(
    user_message="Explain quantum entanglement.",
    assistant_response=response.text,
    tokens=TokenUsage(input=response.usage.input, output=response.usage.output),
)

See docs/byo-llm.md and examples/06_byo_llm.py for a full walkthrough.

id `property` ¶

id: str

The session ID.

model `property` ¶

model: str

The model string for this session.

token_usage `property` ¶

token_usage: TokenUsage

Cumulative token usage across all turns in this session.

compaction_in_progress `property` ¶

compaction_in_progress: bool

True while a background compaction task is running.

event_bus `property` ¶

event_bus: EventBus

The event bus for this session. Subscribe to monitor events.

create `async` `classmethod` ¶

create(
    *,
    model: str,
    agent: str = "default",
    parent_id: str | None = None,
    config: MnesisConfig | None = None,
    system_prompt: str = "You are a helpful assistant.",
    db_path: str | None = None,
    pool: StorePool | None = None,
) -> MnesisSession

Create a new Mnesis session.

Parameters:

Name	Type	Description	Default
`model`	`str`	LLM model string in litellm format (e.g. `"anthropic/claude-opus-4-6"`).	required
`agent`	`str`	Agent role name for multi-agent setups.	`'default'`
`parent_id`	`str \| None`	Parent session ID for sub-sessions (AgenticMap).	`None`
`config`	`MnesisConfig \| None`	Mnesis configuration. Defaults to `MnesisConfig()`.	`None`
`system_prompt`	`str`	System prompt for all turns in this session.	`'You are a helpful assistant.'`
`db_path`	`str \| None`	Override database path (useful for testing). Raises `ValueError` if both `db_path` and `config.store.db_path` are supplied.	`None`
`pool`	`StorePool \| None`	Optional shared connection pool. When provided, all sessions pointing at the same `db_path` share a single connection, avoiding SQLite write-lock contention between concurrent sub-agents. The caller is responsible for calling `pool.close_all()` at shutdown.	`None`

Returns:

Type	Description
`MnesisSession`	An initialized MnesisSession ready to receive messages.

Raises:

Type	Description
`ValueError`	If both `db_path` and `config.store.db_path` are supplied.
`Error`	If the database cannot be initialized.

open `async` `classmethod` ¶

open(
    *,
    model: str,
    agent: str = "default",
    parent_id: str | None = None,
    config: MnesisConfig | None = None,
    system_prompt: str = "You are a helpful assistant.",
    db_path: str | None = None,
    pool: StorePool | None = None,
) -> AsyncGenerator[MnesisSession, None]

Create a new session and use it as an async context manager.

This is the preferred idiom for most callers — it avoids the await ... async with double-ceremony of :meth:create::

async with MnesisSession.open(model="anthropic/claude-opus-4-6") as session:
    result = await session.send("Hello!")

All parameters are identical to :meth:create. The session is automatically closed (pending compaction awaited, DB connection released) when the async with block exits, even on exception.

Parameters:

Name	Type	Description	Default
`model`	`str`	LLM model string in litellm format.	required
`agent`	`str`	Agent role name for multi-agent setups.	`'default'`
`parent_id`	`str \| None`	Parent session ID for sub-sessions (AgenticMap).	`None`
`config`	`MnesisConfig \| None`	Mnesis configuration. Defaults to `MnesisConfig()`.	`None`
`system_prompt`	`str`	System prompt for all turns in this session.	`'You are a helpful assistant.'`
`db_path`	`str \| None`	Override database path.	`None`
`pool`	`StorePool \| None`	Optional shared connection pool (see :meth:`create`).	`None`

Yields:

Type	Description
`AsyncGenerator[MnesisSession, None]`	An initialized MnesisSession.

Note

To receive operator events (LLMMap, AgenticMap), pass event_bus=session.event_bus when constructing operators.

load `async` `classmethod` ¶

load(
    session_id: str,
    config: MnesisConfig | None = None,
    db_path: str | None = None,
    pool: StorePool | None = None,
) -> MnesisSession

Load an existing session from the store.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session ID to load.	required
`config`	`MnesisConfig \| None`	Optional config override.	`None`
`db_path`	`str \| None`	Override database path.	`None`
`pool`	`StorePool \| None`	Optional shared connection pool (see `create()`).	`None`

Returns:

Type	Description
`MnesisSession`	An MnesisSession wrapping the existing session.

Raises:

Type	Description
`SessionNotFoundError`	If the session does not exist.

send `async` ¶

send(
    message: str | list[MessagePart],
    *,
    tools: list[Any] | None = None,
    on_part: Callable[[MessagePart], None | Awaitable[None]]
    | None = None,
    system_prompt: str | None = None,
) -> TurnResult

Send a user message and receive a streaming assistant response.

Parameters:

Name	Type	Description	Default
`message`	`str \| list[MessagePart]`	User message text or list of MessagePart objects.	required
`tools`	`list[Any] \| None`	Optional list of tool definitions (litellm format dicts).	`None`
`on_part`	`Callable[[MessagePart], None \| Awaitable[None]] \| None`	Optional callback invoked for each streamed `MessagePart`. During streaming, `on_part` receives :class:`TextPart` chunks only. Note: :class:`ToolPart` objects are never delivered via `on_part` during `send()`; they are only available when passing pre-built parts to :meth:`record`.	`None`
`system_prompt`	`str \| None`	Override the session system prompt for this turn only. Note: not all providers support per-turn system prompt overrides — check your provider's documentation before relying on this.	`None`

Returns:

Type	Description
`TurnResult`	TurnResult with the assistant's text, token usage, and status.

Raises:

Type	Description
`MnesisStoreError`	If message persistence fails.

record `async` ¶

record(
    user_message: str | list[MessagePart],
    assistant_response: str | list[MessagePart],
    *,
    tokens: TokenUsage | None = None,
    finish_reason: str = "stop",
) -> RecordResult

Persist a completed user/assistant turn without making an LLM call.

Use this when you manage LLM calls yourself (e.g. using the Anthropic, OpenAI, or Gemini SDKs directly) and only want mnesis to handle memory, context assembly, and compaction.

Parameters:

Name	Type	Description	Default
`user_message`	`str \| list[MessagePart]`	The user message text or parts to record.	required
`assistant_response`	`str \| list[MessagePart]`	The assistant reply text or parts to record.	required
`tokens`	`TokenUsage \| None`	Token usage for the turn. Estimated from text if omitted.	`None`
`finish_reason`	`str`	The finish reason from your LLM response (e.g. `"stop"`, `"max_tokens"`). Defaults to `"stop"`.	`'stop'`

Returns:

Type	Description
`RecordResult`	RecordResult with the persisted message IDs and token usage.

Example::

import anthropic
from mnesis import MnesisSession
from mnesis.models.message import TokenUsage

client = anthropic.Anthropic()
session = await MnesisSession.create(model="anthropic/claude-opus-4-6")

user_text = "Explain quantum entanglement."
response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": user_text}],
)
result = await session.record(
    user_message=user_text,
    assistant_response=response.content[0].text,
    tokens=TokenUsage(
        input=response.usage.input_tokens,
        output=response.usage.output_tokens,
    ),
)

messages `async` ¶

messages() -> list[MessageWithParts]

Return the full message history for this session.

The returned list includes all stored rows:

Regular user and assistant turns.
Compaction summary messages (m.is_summary == True).
Messages whose parts contain :class:~mnesis.models.message.CompactionMarkerPart tombstones (tool outputs pruned by the compaction engine).

To get only the conversational turns, use :meth:conversation_messages or filter manually with [m for m in msgs if not m.is_summary].

Returns:

Type	Description
`list[MessageWithParts]`	List of MessageWithParts in chronological order.

conversation_messages `async` ¶

conversation_messages() -> list[MessageWithParts]

Return only the conversational turns, excluding compaction summaries.

Convenience wrapper around :meth:messages that filters out summary messages produced by the compaction engine (is_summary == True). Tombstoned tool outputs within regular messages are still included — only top-level summary rows are removed.

Returns:

Type	Description
`list[MessageWithParts]`	List of non-summary MessageWithParts in chronological order.

context_for_next_turn `async` ¶

context_for_next_turn(
    system_prompt: str | None = None,
) -> list[dict[str, Any]]

Return the compaction-aware context window for the next LLM call.

Intended for BYO-LLM callers who manage their own LLM client but want Mnesis to handle context assembly and compaction. The returned list is in the format expected by most chat completion APIs ([{"role": "user"|"assistant", "content": "..."}, ...]).

Summaries injected by the compaction engine are included as assistant messages at the correct position, so the context is always accurate even after one or more compaction cycles.

After calling your LLM, persist the completed turn with :meth:record so Mnesis can track token usage and trigger compaction when needed.

Parameters:

Name	Type	Description	Default
`system_prompt`	`str \| None`	Override the session system prompt for this turn. Defaults to the system prompt set at :meth:`create` / :meth:`load`.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	Ordered list of `{"role": str, "content": str \| list}` dicts
`list[dict[str, Any]]`	ready to pass to any chat completion API.

Example::

messages = await session.context_for_next_turn()
response = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "system", "content": my_system_prompt}, *messages],
)
await session.record(
    user_message=my_user_text,
    assistant_response=response.choices[0].message.content,
)

compact `async` ¶

compact() -> CompactionResult

Manually trigger synchronous compaction.

Blocks until compaction completes. Useful for checkpointing before complex operations or for testing.

The compaction result is stored internally so that the next :meth:send or :meth:record call will include it in its :class:~mnesis.models.snapshot.TurnSnapshot via the compact_result field.

Returns:

Type	Description
`CompactionResult`	CompactionResult describing the compaction outcome.

close `async` ¶

close() -> None

Clean up session resources.

If background compaction is in progress, close() waits for it to finish before releasing the database connection. This guarantees that no in-flight write is interrupted and that the compaction summary is fully persisted before the DB handle is closed.

Publishes :attr:~mnesis.events.bus.MnesisEvent.SESSION_CLOSED after cleanup.

subscribe ¶

subscribe(event: MnesisEvent, handler: Any) -> None

Register an event handler on this session's event bus.

Convenience wrapper for session.event_bus.subscribe().

history ¶

history() -> list[TurnSnapshot]

Return the accumulated per-turn context snapshots.

Each entry corresponds to one completed turn — one :meth:send call or one :meth:record call. Entries are appended in turn order and the list grows monotonically; it is never reordered or truncated.

Typical use-cases:

Debugging — inspect the exact context composition at each turn to diagnose unexpected compaction or budget overflows.
Research — plot sawtooth token-usage curves, compute compaction level distributions, or analyse information retention across compaction boundaries.

Returns:

Type	Description
`list[TurnSnapshot]`	A list of :class:`~mnesis.models.snapshot.TurnSnapshot` objects in
`list[TurnSnapshot]`	chronological order. Returns an empty list if no turns have
`list[TurnSnapshot]`	completed yet.

Example::

async with MnesisSession.open(model="anthropic/claude-opus-4-6") as session:
    await session.send("Hello!")
    await session.send("Tell me about Paris.")
    for snap in session.history():
        print(
            f"Turn {snap.turn_index}: "
            f"{snap.context_tokens.total} tokens "
            f"(summary={snap.context_tokens.summary})"
        )

make_id ¶

make_id(prefix: str) -> str

Generate a ULID-based sortable identifier.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	Short prefix for readability (e.g. `"msg"`, `"sess"`, `"part"`).	required

Returns:

Type	Description
`str`	ID string in the format `"{prefix}_{ulid}"`.

mnesis.session¶

session ¶

MnesisSession ¶

id property ¶

model property ¶

token_usage property ¶

compaction_in_progress property ¶

event_bus property ¶

create async classmethod ¶

open async classmethod ¶

load async classmethod ¶

send async ¶

record async ¶

messages async ¶

conversation_messages async ¶

context_for_next_turn async ¶

compact async ¶

close async ¶

subscribe ¶

history ¶

make_id ¶

`mnesis.session`¶

id `property` ¶

model `property` ¶

token_usage `property` ¶

compaction_in_progress `property` ¶

event_bus `property` ¶

create `async` `classmethod` ¶

open `async` `classmethod` ¶

load `async` `classmethod` ¶

send `async` ¶

record `async` ¶

messages `async` ¶

conversation_messages `async` ¶

context_for_next_turn `async` ¶

compact `async` ¶

close `async` ¶