Memory

Three orthogonal layers, each with its own contract:

1. Conversation management — what to do with the message history within a single run (keep, window, summarize) when it grows past the model's context window.
2. Cross-thread store — durable key-value scoped by namespace, used for long-term memory entries the agent reads at session start and writes at session end.
3. Long-term memory manager — the policy layer that decides what to extract from a run and when to retrieve, sitting on top of any BaseStore backend.

For checkpointing (state persistence between runs), see Checkpointers — those backends live in tulip.memory.backends, with S3-compatible object storage as a production target. Note those are checkpointer backends (per-thread state); the cross-thread store layer below is a separate contract.

Conversation management

ConversationManager

Bases: ABC

Base class for conversation management strategies.

Conversation managers handle how message history is maintained, including trimming, summarization, and other memory management.

apply abstractmethod

apply(messages: list[Message]) -> list[Message]

Apply conversation management to messages.

Parameters:

Name	Type	Description	Default
messages	list[Message]	Full message history	required

Returns:

Type	Description
list[Message]	Managed message list (potentially trimmed/summarized)

Source code in .sdk/src/tulip/memory/conversation.py

@abstractmethod
def apply(self, messages: list[Message]) -> list[Message]:
    """
    Apply conversation management to messages.

    Args:
        messages: Full message history

    Returns:
        Managed message list (potentially trimmed/summarized)
    """
    ...

async_apply async

async_apply(messages: list[Message]) -> list[Message]

Async version of apply. Supports async summarization functions.

Default implementation delegates to the synchronous apply(). Override in subclasses that need async operations (e.g., LLM summarization).

Source code in .sdk/src/tulip/memory/conversation.py

async def async_apply(self, messages: list[Message]) -> list[Message]:
    """
    Async version of apply. Supports async summarization functions.

    Default implementation delegates to the synchronous apply().
    Override in subclasses that need async operations (e.g., LLM summarization).
    """
    return self.apply(messages)

NullManager

Bases: ConversationManager

Null conversation manager - keeps all messages unchanged.

Use this when you want to preserve the entire conversation history without any modifications. Suitable for short conversations or when full history is required.

apply

apply(messages: list[Message]) -> list[Message]

Return messages unchanged.

Source code in .sdk/src/tulip/memory/conversation.py

def apply(self, messages: list[Message]) -> list[Message]:
    """Return messages unchanged."""
    return messages.copy()

async_apply async

async_apply(messages: list[Message]) -> list[Message]

Async version of apply. Supports async summarization functions.

Default implementation delegates to the synchronous apply(). Override in subclasses that need async operations (e.g., LLM summarization).

Source code in .sdk/src/tulip/memory/conversation.py

async def async_apply(self, messages: list[Message]) -> list[Message]:
    """
    Async version of apply. Supports async summarization functions.

    Default implementation delegates to the synchronous apply().
    Override in subclasses that need async operations (e.g., LLM summarization).
    """
    return self.apply(messages)

SlidingWindowManager

SlidingWindowManager(window_size: int = 20, preserve_system: bool = True, preserve_first_user: bool = True)

Bases: ConversationManager

Sliding window conversation manager - keeps last N messages.

Preserves the system message (if present) and the last N messages from the conversation. This is a simple and effective strategy for managing conversation length.

When the window would retain no user turn at all, the opening request is re-attached. That case is an agent loop — assistant/tool all the way down — where the first user message is the task, and losing it leaves the model working from a role description and a wall of tool output with no statement of what it was asked to do. It reads as drift rather than as an error, and on Qwen-family chat templates it fails outright with No user query found in messages.. Twenty tool round-trips reach the default window, so this is the ordinary case for a tool-using agent, not an edge one.

A chat is left alone: while any user turn survives in the window, the newest ones are the live context and the first is not privileged.

Parameters:

Name	Type	Description	Default
window_size	int	Maximum number of messages to keep (excluding system message)	20
preserve_system	bool	Whether to preserve the system message at the start	True
preserve_first_user	bool	Whether to re-attach the opening user turn when the window would otherwise contain none	True

Source code in .sdk/src/tulip/memory/conversation.py

def __init__(
    self,
    window_size: int = 20,
    preserve_system: bool = True,
    preserve_first_user: bool = True,
):
    if window_size < 1:
        raise ValueError("window_size must be at least 1")
    self.window_size = window_size
    self.preserve_system = preserve_system
    self.preserve_first_user = preserve_first_user

apply

apply(messages: list[Message]) -> list[Message]

Apply sliding window to messages.

Keeps the system message (if preserve_system is True) and the last window_size messages, re-attaching the opening user turn when the window would otherwise retain none (see the class docstring).

Source code in .sdk/src/tulip/memory/conversation.py

def apply(self, messages: list[Message]) -> list[Message]:
    """
    Apply sliding window to messages.

    Keeps the system message (if preserve_system is True) and the last
    window_size messages, re-attaching the opening user turn when the window
    would otherwise retain none (see the class docstring).
    """
    if not messages:
        return []

    from tulip.core.messages import Role

    result: list[Message] = []
    non_system_messages: list[Message] = []

    for msg in messages:
        if msg.role == Role.SYSTEM and self.preserve_system:
            result.append(msg)
        else:
            non_system_messages.append(msg)

    # Keep only the last window_size messages
    before_count = len(non_system_messages)
    if before_count > self.window_size:
        anchor: Message | None = None
        if self.preserve_first_user:
            anchor = next(
                (m for m in non_system_messages if m.role == Role.USER),
                None,
            )

        non_system_messages = non_system_messages[-self.window_size :]

        # The cut lands at a fixed offset, not a turn boundary, so it can
        # fall between an assistant's tool_calls and the tool messages
        # answering them. Those results are orphans the provider will
        # reject, so drop them here rather than leaving the agent loop to
        # strip them later.
        def _drop_leading_orphans(window: list[Message]) -> list[Message]:
            kept = {tc.id for m in window if m.role == Role.ASSISTANT for tc in m.tool_calls}
            start = 0
            while (
                start < len(window)
                and window[start].role == Role.TOOL
                and window[start].tool_call_id not in kept
            ):
                start += 1
            return window[start:]

        non_system_messages = _drop_leading_orphans(non_system_messages)

        # Re-attach the opening request only when the window kept no user
        # turn at all. In a chat the newest user turns are the live context
        # and the first one is not privileged; it is an agent loop —
        # assistant/tool all the way down — that ends up with no user turn
        # and therefore no statement of the task.
        #
        # Trimming orphans usually frees the slot; if it didn't, drop one
        # more turn from the front and re-normalise, since that drop can
        # itself orphan the results that followed it.
        if anchor is not None and not any(m.role == Role.USER for m in non_system_messages):
            if len(non_system_messages) >= self.window_size:
                non_system_messages = _drop_leading_orphans(non_system_messages[1:])
            non_system_messages.insert(0, anchor)
        from tulip.observability.emit import (  # noqa: PLC0415
            EV_MEMORY_CONVERSATION_PRUNED,
            emit_sync,
        )

        emit_sync(
            EV_MEMORY_CONVERSATION_PRUNED,
            strategy="sliding_window",
            window_size=self.window_size,
            # Count what actually went, not what the raw slice implies:
            # orphan trimming and the re-attached anchor both move this.
            removed_count=before_count - len(non_system_messages),
        )

    result.extend(non_system_messages)
    return result

async_apply async

async_apply(messages: list[Message]) -> list[Message]

Async version of apply. Supports async summarization functions.

Default implementation delegates to the synchronous apply(). Override in subclasses that need async operations (e.g., LLM summarization).

Source code in .sdk/src/tulip/memory/conversation.py

async def async_apply(self, messages: list[Message]) -> list[Message]:
    """
    Async version of apply. Supports async summarization functions.

    Default implementation delegates to the synchronous apply().
    Override in subclasses that need async operations (e.g., LLM summarization).
    """
    return self.apply(messages)

SummarizingManager

SummarizingManager(threshold: int = 30, keep_recent: int = 10, summarize_fn: Any | None = None, preserve_system: bool = True)

Bases: ConversationManager

Summarizing conversation manager - summarizes older messages.

When the conversation exceeds a threshold, older messages are summarized into a single summary message, preserving recent context.

This manager requires a summarization function to be provided, which can use an LLM to generate summaries.

Parameters:

Name	Type	Description	Default
threshold	int	Number of messages before summarization kicks in	30
keep_recent	int	Number of recent messages to always preserve	10
summarize_fn	Any | None	Async function that summarizes a list of messages	None
preserve_system	bool	Whether to preserve the system message	True

Source code in .sdk/src/tulip/memory/conversation.py

def __init__(
    self,
    threshold: int = 30,
    keep_recent: int = 10,
    summarize_fn: Any | None = None,
    preserve_system: bool = True,
):
    if threshold < 1:
        raise ValueError("threshold must be at least 1")
    if keep_recent < 1:
        raise ValueError("keep_recent must be at least 1")
    if keep_recent >= threshold:
        raise ValueError("keep_recent must be less than threshold")

    self.threshold = threshold
    self.keep_recent = keep_recent
    self.summarize_fn = summarize_fn
    self.preserve_system = preserve_system
    self._summary_cache: dict[int, str] = {}

apply

apply(messages: list[Message]) -> list[Message]

Apply summarization to messages.

If total messages exceed threshold, older messages are summarized. Note: If no summarize_fn is provided, falls back to a simple summary.

Source code in .sdk/src/tulip/memory/conversation.py

def apply(self, messages: list[Message]) -> list[Message]:
    """
    Apply summarization to messages.

    If total messages exceed threshold, older messages are summarized.
    Note: If no summarize_fn is provided, falls back to a simple summary.
    """
    if not messages:
        return []

    from tulip.core.messages import Message as Msg
    from tulip.core.messages import Role

    result: list[Msg] = []
    non_system_messages: list[Msg] = []
    system_message: Msg | None = None

    for msg in messages:
        if msg.role == Role.SYSTEM and self.preserve_system:
            system_message = msg
        else:
            non_system_messages.append(msg)

    # If under threshold, no summarization needed
    if len(non_system_messages) <= self.threshold:
        if system_message:
            result.append(system_message)
        result.extend(non_system_messages)
        return result

    # Split into messages to summarize and recent messages
    to_summarize = non_system_messages[: -self.keep_recent]
    recent = non_system_messages[-self.keep_recent :]

    # Generate summary
    summary_text = self._generate_summary(to_summarize)

    # Build result
    if system_message:
        result.append(system_message)

    # Add summary as a system message
    summary_message = Msg(
        role=Role.SYSTEM,
        content=f"[Summary of previous conversation ({len(to_summarize)} messages)]:\n{summary_text}",
    )
    result.append(summary_message)
    result.extend(recent)

    return result

async_apply async

async_apply(messages: list[Message]) -> list[Message]

Async apply with LLM summarization support.

If summarize_fn is an async function, it will be properly awaited. Falls back to synchronous _generate_summary() otherwise.

Source code in .sdk/src/tulip/memory/conversation.py

async def async_apply(self, messages: list[Message]) -> list[Message]:
    """
    Async apply with LLM summarization support.

    If summarize_fn is an async function, it will be properly awaited.
    Falls back to synchronous _generate_summary() otherwise.
    """
    if not messages:
        return []

    from tulip.core.messages import Message as Msg
    from tulip.core.messages import Role

    result: list[Msg] = []
    non_system_messages: list[Msg] = []
    system_message: Msg | None = None

    for msg in messages:
        if msg.role == Role.SYSTEM and self.preserve_system:
            system_message = msg
        else:
            non_system_messages.append(msg)

    if len(non_system_messages) <= self.threshold:
        if system_message:
            result.append(system_message)
        result.extend(non_system_messages)
        return result

    to_summarize = non_system_messages[: -self.keep_recent]
    recent = non_system_messages[-self.keep_recent :]

    # Use async summarize_fn if available
    import asyncio

    from tulip.observability.emit import (  # noqa: PLC0415
        EV_MEMORY_COMPACTOR_COMPLETED,
        EV_MEMORY_COMPACTOR_TRIGGERED,
        emit,
    )

    await emit(
        EV_MEMORY_COMPACTOR_TRIGGERED,
        strategy="summarizing",
        messages_before=len(non_system_messages),
        threshold=self.threshold,
    )
    import time as _time  # noqa: PLC0415

    _started = _time.perf_counter()

    if self.summarize_fn is not None and asyncio.iscoroutinefunction(self.summarize_fn):
        summary_text = await self.summarize_fn(to_summarize)
    else:
        summary_text = self._generate_summary(to_summarize)

    if system_message:
        result.append(system_message)

    summary_message = Msg(
        role=Role.SYSTEM,
        content=f"[Summary of previous conversation ({len(to_summarize)} messages)]:\n{summary_text}",
    )
    result.append(summary_message)
    result.extend(recent)
    await emit(
        EV_MEMORY_COMPACTOR_COMPLETED,
        strategy="summarizing",
        messages_before=len(non_system_messages),
        messages_after=len(result),
        summarized_count=len(to_summarize),
        duration_ms=(_time.perf_counter() - _started) * 1000,
    )
    return result

Cross-thread store

The cross-thread KV the long-term memory manager writes to. Three implementations ship:

Store	Backing	Reach for it when
InMemoryStore	in-process dict	tests, and anything that can lose memory on exit
HolographicStore	SQLite + FTS5 + holographic reduced representations	you want durable memory with no external infrastructure
PgMemory	Postgres + pgvector, per-tenant row-level security	multi-tenant, where isolation has to hold at the database

The drivers in tulip.memory.backends referenced by Checkpointers implement BaseCheckpointer (per-thread state), not BaseStore. The two layers are separate contracts and a backend for one is not a backend for the other.

BaseStore

Bases: ABC

Abstract base class for store implementations.

Provides common functionality and defines the interface.

capabilities property

capabilities: StoreCapabilities

Return capabilities. Override in subclasses.

put abstractmethod async

put(namespace: tuple[str, ...], key: str, value: Any, metadata: dict[str, Any] | None = None) -> None

Store a value.

Source code in .sdk/src/tulip/memory/store.py

@abstractmethod
async def put(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: Any,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Store a value."""
    ...

get abstractmethod async

get(namespace: tuple[str, ...], key: str) -> Any | None

Retrieve a value.

Source code in .sdk/src/tulip/memory/store.py

@abstractmethod
async def get(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> Any | None:
    """Retrieve a value."""
    ...

delete abstractmethod async

delete(namespace: tuple[str, ...], key: str) -> bool

Delete a value.

Source code in .sdk/src/tulip/memory/store.py

@abstractmethod
async def delete(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> bool:
    """Delete a value."""
    ...

list_keys abstractmethod async

list_keys(namespace: tuple[str, ...], limit: int = 100) -> list[str]

List keys in namespace.

Source code in .sdk/src/tulip/memory/store.py

@abstractmethod
async def list_keys(
    self,
    namespace: tuple[str, ...],
    limit: int = 100,
) -> list[str]:
    """List keys in namespace."""
    ...

exists async

exists(namespace: tuple[str, ...], key: str) -> bool

Check if a key exists.

Source code in .sdk/src/tulip/memory/store.py

async def exists(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> bool:
    """Check if a key exists."""
    value = await self.get(namespace, key)
    return value is not None

get_item async

get_item(namespace: tuple[str, ...], key: str) -> StoreItem | None

Get full item with metadata.

Source code in .sdk/src/tulip/memory/store.py

async def get_item(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> StoreItem | None:
    """Get full item with metadata."""
    # Default implementation - subclasses can override
    value = await self.get(namespace, key)
    if value is None:
        return None
    now = datetime.now(UTC)
    return StoreItem(
        namespace=namespace,
        key=key,
        value=value,
        metadata={},
        created_at=now,
        updated_at=now,
    )

search async

search(namespace: tuple[str, ...], query: str | None = None, limit: int = 10) -> list[StoreItem]

Search for items in namespace.

Optional method — backends advertise support via capabilities.search. Subclasses that support search must override this method; the default raises :class:StoreCapabilityError.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to search in	required
query	str | None	Search query (implementation-specific)	None
limit	int	Maximum results	10

Returns:

Type	Description
list[StoreItem]	List of matching items

Raises:

Type	Description
StoreCapabilityError	If the backend does not support search.

Source code in .sdk/src/tulip/memory/store.py

async def search(
    self,
    namespace: tuple[str, ...],
    query: str | None = None,
    limit: int = 10,
) -> list[StoreItem]:
    """
    Search for items in namespace.

    Optional method — backends advertise support via
    ``capabilities.search``. Subclasses that support search must
    override this method; the default raises
    :class:`StoreCapabilityError`.

    Args:
        namespace: Namespace to search in
        query: Search query (implementation-specific)
        limit: Maximum results

    Returns:
        List of matching items

    Raises:
        StoreCapabilityError: If the backend does not support search.
    """
    raise StoreCapabilityError(
        capability="search",
        store_class=type(self).__name__,
        hint=(
            "Choose a backend that advertises capabilities.search "
            "(e.g., InMemoryStore, OpenSearch, pgvector)."
        ),
    )

put_with_embedding async

put_with_embedding(namespace: tuple[str, ...], key: str, value: Any, embedding: list[float], metadata: dict[str, Any] | None = None) -> None

Store a value with its embedding vector for semantic search.

Optional method — backends advertise support via capabilities.semantic_search. Subclasses that support semantic search must override this method.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required
value	Any	Value to store (must be JSON-serializable)	required
embedding	list[float]	Vector embedding (e.g., from OpenAI, Cohere, etc.)	required
metadata	dict[str, Any] | None	Optional metadata for filtering	None

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Example

Get embedding from your provider

embedding = await embedder.embed("User prefers dark theme")

await store.put_with_embedding( ("users", user_id, "memories"), "theme_preference", {"theme": "dark", "reason": "easier on eyes"}, embedding=embedding, metadata={"category": "preferences"} )

Source code in .sdk/src/tulip/memory/store.py

async def put_with_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: Any,
    embedding: list[float],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Store a value with its embedding vector for semantic search.

    Optional method — backends advertise support via
    ``capabilities.semantic_search``. Subclasses that support
    semantic search must override this method.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace
        value: Value to store (must be JSON-serializable)
        embedding: Vector embedding (e.g., from OpenAI, Cohere, etc.)
        metadata: Optional metadata for filtering

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.

    Example:
        # Get embedding from your provider
        embedding = await embedder.embed("User prefers dark theme")

        await store.put_with_embedding(
            ("users", user_id, "memories"),
            "theme_preference",
            {"theme": "dark", "reason": "easier on eyes"},
            embedding=embedding,
            metadata={"category": "preferences"}
        )
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Use put() for plain storage, or pick a vector-capable backend.",
    )

search_by_embedding async

search_by_embedding(namespace: tuple[str, ...], query_embedding: list[float], limit: int = 10, threshold: float | None = None, metadata_filter: dict[str, Any] | None = None) -> list[SemanticSearchResult]

Search for similar items using vector similarity.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to search in	required
query_embedding	list[float]	Vector to compare against stored embeddings	required
limit	int	Maximum results to return	10
threshold	float | None	Minimum similarity score (0.0-1.0), or None for no threshold	None
metadata_filter	dict[str, Any] | None	Optional metadata constraints	None

Returns:

Type	Description
list[SemanticSearchResult]	List of SemanticSearchResult, sorted by similarity (highest first)

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def search_by_embedding(
    self,
    namespace: tuple[str, ...],
    query_embedding: list[float],
    limit: int = 10,
    threshold: float | None = None,
    metadata_filter: dict[str, Any] | None = None,
) -> list[SemanticSearchResult]:
    """
    Search for similar items using vector similarity.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Namespace to search in
        query_embedding: Vector to compare against stored embeddings
        limit: Maximum results to return
        threshold: Minimum similarity score (0.0-1.0), or None for no threshold
        metadata_filter: Optional metadata constraints

    Returns:
        List of SemanticSearchResult, sorted by similarity (highest first)

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Pick a vector-capable backend (e.g., pgvector, Qdrant).",
    )

get_embedding async

get_embedding(namespace: tuple[str, ...], key: str) -> list[float] | None

Get the embedding vector for a stored item.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required

Returns:

Type	Description
list[float] | None	Embedding vector or None if not found / no embedding stored.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def get_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> list[float] | None:
    """
    Get the embedding vector for a stored item.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace

    Returns:
        Embedding vector or None if not found / no embedding stored.

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
    )

put_batch async

put_batch(items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]) -> None

Store multiple items.

Backends that advertise capabilities.batch_operations should override this method to dispatch atomically. The default falls back to sequential put() calls so callers always get functional batching, just without atomicity guarantees.

Parameters:

Name	Type	Description	Default
items	list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]	List of (namespace, key, value, metadata) tuples	required

Source code in .sdk/src/tulip/memory/store.py

async def put_batch(
    self,
    items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]],
) -> None:
    """
    Store multiple items.

    Backends that advertise ``capabilities.batch_operations`` should
    override this method to dispatch atomically. The default falls
    back to sequential ``put()`` calls so callers always get
    functional batching, just without atomicity guarantees.

    Args:
        items: List of (namespace, key, value, metadata) tuples
    """
    # Sequential fallback — works for every backend, even those that
    # don't advertise batch_operations. Backends that *do* advertise
    # the capability override this method to dispatch in one
    # transaction / round-trip.
    for namespace, key, value, metadata in items:
        await self.put(namespace, key, value, metadata)

get_batch async

get_batch(keys: list[tuple[tuple[str, ...], str]]) -> dict[tuple[tuple[str, ...], str], Any]

Retrieve multiple items.

Parameters:

Name	Type	Description	Default
keys	list[tuple[tuple[str, ...], str]]	List of (namespace, key) tuples	required

Returns:

Type	Description
dict[tuple[tuple[str, ...], str], Any]	Dict mapping (namespace, key) to value

Source code in .sdk/src/tulip/memory/store.py

async def get_batch(
    self,
    keys: list[tuple[tuple[str, ...], str]],
) -> dict[tuple[tuple[str, ...], str], Any]:
    """
    Retrieve multiple items.

    Args:
        keys: List of (namespace, key) tuples

    Returns:
        Dict mapping (namespace, key) to value
    """
    results = {}
    for namespace, key in keys:
        value = await self.get(namespace, key)
        if value is not None:
            results[(namespace, key)] = value
    return results

list_namespaces async

list_namespaces(prefix: tuple[str, ...] | None = None, limit: int = 100) -> list[tuple[str, ...]]

List all namespaces.

Optional method — see capabilities.list_namespaces.

Parameters:

Name	Type	Description	Default
prefix	tuple[str, ...] | None	Optional prefix to filter namespaces	None
limit	int	Maximum namespaces to return	100

Returns:

Type	Description
list[tuple[str, ...]]	List of namespace tuples

Raises:

Type	Description
StoreCapabilityError	If the backend does not support namespace listing.

Source code in .sdk/src/tulip/memory/store.py

async def list_namespaces(
    self,
    prefix: tuple[str, ...] | None = None,
    limit: int = 100,
) -> list[tuple[str, ...]]:
    """
    List all namespaces.

    Optional method — see ``capabilities.list_namespaces``.

    Args:
        prefix: Optional prefix to filter namespaces
        limit: Maximum namespaces to return

    Returns:
        List of namespace tuples

    Raises:
        StoreCapabilityError: If the backend does not support
            namespace listing.
    """
    raise StoreCapabilityError(
        capability="list_namespaces",
        store_class=type(self).__name__,
    )

clear_namespace async

clear_namespace(namespace: tuple[str, ...]) -> int

Delete all items in a namespace.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to clear	required

Returns:

Type	Description
int	Number of items deleted

Source code in .sdk/src/tulip/memory/store.py

async def clear_namespace(
    self,
    namespace: tuple[str, ...],
) -> int:
    """
    Delete all items in a namespace.

    Args:
        namespace: Namespace to clear

    Returns:
        Number of items deleted
    """
    keys = await self.list_keys(namespace, limit=10000)
    count = 0
    for key in keys:
        if await self.delete(namespace, key):
            count += 1
    return count

close async

close() -> None

Close any resources.

Source code in .sdk/src/tulip/memory/store.py

async def close(self) -> None:
    """Close any resources."""

InMemoryStore

InMemoryStore()

Bases: BaseStore

In-memory store implementation.

Fast but not persistent - data is lost when process exits. Useful for testing and development.

Source code in .sdk/src/tulip/memory/store.py

def __init__(self) -> None:
    self._data: dict[str, StoreItem] = {}
    self._namespaces: set[tuple[str, ...]] = set()
    self._lock = asyncio.Lock()

exists async

exists(namespace: tuple[str, ...], key: str) -> bool

Check if a key exists.

Source code in .sdk/src/tulip/memory/store.py

async def exists(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> bool:
    """Check if a key exists."""
    value = await self.get(namespace, key)
    return value is not None

put_with_embedding async

put_with_embedding(namespace: tuple[str, ...], key: str, value: Any, embedding: list[float], metadata: dict[str, Any] | None = None) -> None

Store a value with its embedding vector for semantic search.

Optional method — backends advertise support via capabilities.semantic_search. Subclasses that support semantic search must override this method.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required
value	Any	Value to store (must be JSON-serializable)	required
embedding	list[float]	Vector embedding (e.g., from OpenAI, Cohere, etc.)	required
metadata	dict[str, Any] | None	Optional metadata for filtering	None

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Example

Get embedding from your provider

embedding = await embedder.embed("User prefers dark theme")

await store.put_with_embedding( ("users", user_id, "memories"), "theme_preference", {"theme": "dark", "reason": "easier on eyes"}, embedding=embedding, metadata={"category": "preferences"} )

Source code in .sdk/src/tulip/memory/store.py

async def put_with_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: Any,
    embedding: list[float],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Store a value with its embedding vector for semantic search.

    Optional method — backends advertise support via
    ``capabilities.semantic_search``. Subclasses that support
    semantic search must override this method.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace
        value: Value to store (must be JSON-serializable)
        embedding: Vector embedding (e.g., from OpenAI, Cohere, etc.)
        metadata: Optional metadata for filtering

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.

    Example:
        # Get embedding from your provider
        embedding = await embedder.embed("User prefers dark theme")

        await store.put_with_embedding(
            ("users", user_id, "memories"),
            "theme_preference",
            {"theme": "dark", "reason": "easier on eyes"},
            embedding=embedding,
            metadata={"category": "preferences"}
        )
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Use put() for plain storage, or pick a vector-capable backend.",
    )

search_by_embedding async

search_by_embedding(namespace: tuple[str, ...], query_embedding: list[float], limit: int = 10, threshold: float | None = None, metadata_filter: dict[str, Any] | None = None) -> list[SemanticSearchResult]

Search for similar items using vector similarity.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to search in	required
query_embedding	list[float]	Vector to compare against stored embeddings	required
limit	int	Maximum results to return	10
threshold	float | None	Minimum similarity score (0.0-1.0), or None for no threshold	None
metadata_filter	dict[str, Any] | None	Optional metadata constraints	None

Returns:

Type	Description
list[SemanticSearchResult]	List of SemanticSearchResult, sorted by similarity (highest first)

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def search_by_embedding(
    self,
    namespace: tuple[str, ...],
    query_embedding: list[float],
    limit: int = 10,
    threshold: float | None = None,
    metadata_filter: dict[str, Any] | None = None,
) -> list[SemanticSearchResult]:
    """
    Search for similar items using vector similarity.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Namespace to search in
        query_embedding: Vector to compare against stored embeddings
        limit: Maximum results to return
        threshold: Minimum similarity score (0.0-1.0), or None for no threshold
        metadata_filter: Optional metadata constraints

    Returns:
        List of SemanticSearchResult, sorted by similarity (highest first)

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Pick a vector-capable backend (e.g., pgvector, Qdrant).",
    )

get_embedding async

get_embedding(namespace: tuple[str, ...], key: str) -> list[float] | None

Get the embedding vector for a stored item.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required

Returns:

Type	Description
list[float] | None	Embedding vector or None if not found / no embedding stored.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def get_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> list[float] | None:
    """
    Get the embedding vector for a stored item.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace

    Returns:
        Embedding vector or None if not found / no embedding stored.

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
    )

get_batch async

get_batch(keys: list[tuple[tuple[str, ...], str]]) -> dict[tuple[tuple[str, ...], str], Any]

Retrieve multiple items.

Parameters:

Name	Type	Description	Default
keys	list[tuple[tuple[str, ...], str]]	List of (namespace, key) tuples	required

Returns:

Type	Description
dict[tuple[tuple[str, ...], str], Any]	Dict mapping (namespace, key) to value

Source code in .sdk/src/tulip/memory/store.py

async def get_batch(
    self,
    keys: list[tuple[tuple[str, ...], str]],
) -> dict[tuple[tuple[str, ...], str], Any]:
    """
    Retrieve multiple items.

    Args:
        keys: List of (namespace, key) tuples

    Returns:
        Dict mapping (namespace, key) to value
    """
    results = {}
    for namespace, key in keys:
        value = await self.get(namespace, key)
        if value is not None:
            results[(namespace, key)] = value
    return results

clear_namespace async

clear_namespace(namespace: tuple[str, ...]) -> int

Delete all items in a namespace.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to clear	required

Returns:

Type	Description
int	Number of items deleted

Source code in .sdk/src/tulip/memory/store.py

async def clear_namespace(
    self,
    namespace: tuple[str, ...],
) -> int:
    """
    Delete all items in a namespace.

    Args:
        namespace: Namespace to clear

    Returns:
        Number of items deleted
    """
    keys = await self.list_keys(namespace, limit=10000)
    count = 0
    for key in keys:
        if await self.delete(namespace, key):
            count += 1
    return count

close async

close() -> None

Close any resources.

Source code in .sdk/src/tulip/memory/store.py

async def close(self) -> None:
    """Close any resources."""

NamespacedStore

NamespacedStore(store: BaseStore, namespace: tuple[str, ...])

Store wrapper with a fixed namespace prefix.

Makes it easier to work with a specific scope.

Example

user_store = NamespacedStore(store, ("users", user_id)) await user_store.put("preferences", {"theme": "dark"}) prefs = await user_store.get("preferences")

Source code in .sdk/src/tulip/memory/store.py

def __init__(
    self,
    store: BaseStore,
    namespace: tuple[str, ...],
):
    self._store = store
    self._namespace = namespace

scoped

scoped(*suffix: str) -> NamespacedStore

Create new wrapper with extended namespace.

Source code in .sdk/src/tulip/memory/store.py

def scoped(self, *suffix: str) -> NamespacedStore:
    """Create new wrapper with extended namespace."""
    return NamespacedStore(self._store, (*self._namespace, *suffix))

StoreContext

StoreContext(store: BaseStore, user_id: str | None = None, session_id: str | None = None)

Context object passed to nodes for store access.

Provides convenient methods for common memory operations.

Example

async def my_node(inputs, *, store: StoreContext): # Get user memory user_prefs = await store.get_user_memory("preferences")

# Remember something
await store.remember(
    "discussed_topic",
    {"topic": "python", "timestamp": now}
)

# Search memories
related = await store.search_memories("python")

Source code in .sdk/src/tulip/memory/store.py

def __init__(
    self,
    store: BaseStore,
    user_id: str | None = None,
    session_id: str | None = None,
):
    self._store = store
    self._user_id = user_id
    self._session_id = session_id

store property

store: BaseStore

Access the underlying store.

for_user

for_user(user_id: str | None = None) -> NamespacedStore

Get namespaced store for a user.

If user_id is not provided, uses the user_id from the context.

Source code in .sdk/src/tulip/memory/store.py

def for_user(self, user_id: str | None = None) -> NamespacedStore:
    """Get namespaced store for a user.

    If user_id is not provided, uses the user_id from the context.
    """
    uid = user_id or self._user_id
    if not uid:
        raise ValueError("user_id must be provided or set in context")
    return NamespacedStore(self._store, ("users", uid))

for_session

for_session(session_id: str | None = None) -> NamespacedStore

Get namespaced store for a session.

If session_id is not provided, uses the session_id from the context.

Source code in .sdk/src/tulip/memory/store.py

def for_session(self, session_id: str | None = None) -> NamespacedStore:
    """Get namespaced store for a session.

    If session_id is not provided, uses the session_id from the context.
    """
    sid = session_id or self._session_id
    if not sid:
        raise ValueError("session_id must be provided or set in context")
    return NamespacedStore(self._store, ("sessions", sid))

get_user_memory async

get_user_memory(key: str) -> Any | None

Get a memory for the current user.

Source code in .sdk/src/tulip/memory/store.py

async def get_user_memory(self, key: str) -> Any | None:
    """Get a memory for the current user."""
    if not self._user_id:
        return None
    return await self._store.get(("users", self._user_id, "memories"), key)

remember async

remember(key: str, value: Any, metadata: dict[str, Any] | None = None) -> None

Store a memory for the current user.

Source code in .sdk/src/tulip/memory/store.py

async def remember(
    self,
    key: str,
    value: Any,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Store a memory for the current user."""
    if not self._user_id:
        raise ValueError("No user_id set in store context")
    await self._store.put(
        ("users", self._user_id, "memories"),
        key,
        value,
        metadata,
    )

forget async

forget(key: str) -> bool

Delete a memory for the current user.

Source code in .sdk/src/tulip/memory/store.py

async def forget(self, key: str) -> bool:
    """Delete a memory for the current user."""
    if not self._user_id:
        return False
    return await self._store.delete(("users", self._user_id, "memories"), key)

search_memories async

search_memories(query: str, limit: int = 10) -> list[StoreItem]

Search user memories (full-text).

Source code in .sdk/src/tulip/memory/store.py

async def search_memories(
    self,
    query: str,
    limit: int = 10,
) -> list[StoreItem]:
    """Search user memories (full-text)."""
    if not self._user_id:
        return []
    return await self._store.search(
        ("users", self._user_id, "memories"),
        query,
        limit,
    )

remember_with_embedding async

remember_with_embedding(key: str, value: Any, embedding: list[float], metadata: dict[str, Any] | None = None) -> None

Store a memory with its embedding for semantic search.

Requires: capabilities.semantic_search = True

Parameters:

Name	Type	Description	Default
key	str	Memory key	required
value	Any	Value to store	required
embedding	list[float]	Vector embedding for semantic search	required
metadata	dict[str, Any] | None	Optional metadata	None

Example

embedding = await embedder.embed("User likes dark theme") await store.remember_with_embedding( "theme_pref", {"theme": "dark"}, embedding=embedding, )

Source code in .sdk/src/tulip/memory/store.py

async def remember_with_embedding(
    self,
    key: str,
    value: Any,
    embedding: list[float],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Store a memory with its embedding for semantic search.

    Requires: capabilities.semantic_search = True

    Args:
        key: Memory key
        value: Value to store
        embedding: Vector embedding for semantic search
        metadata: Optional metadata

    Example:
        embedding = await embedder.embed("User likes dark theme")
        await store.remember_with_embedding(
            "theme_pref",
            {"theme": "dark"},
            embedding=embedding,
        )
    """
    if not self._user_id:
        raise ValueError("No user_id set in store context")
    await self._store.put_with_embedding(
        ("users", self._user_id, "memories"),
        key,
        value,
        embedding,
        metadata,
    )

search_memories_semantic async

search_memories_semantic(query_embedding: list[float], limit: int = 10, threshold: float | None = None) -> list[SemanticSearchResult]

Search user memories by semantic similarity.

Requires: capabilities.semantic_search = True

Parameters:

Name	Type	Description	Default
query_embedding	list[float]	Vector to search with	required
limit	int	Maximum results	10
threshold	float | None	Minimum similarity score (0.0-1.0)	None

Returns:

Type	Description
list[SemanticSearchResult]	List of SemanticSearchResult sorted by similarity

Example

query_vec = await embedder.embed("user preferences") results = await store.search_memories_semantic( query_embedding=query_vec, limit=5, threshold=0.7, ) for r in results: print(f"{r.item.key}: {r.score:.2f} similar")

Source code in .sdk/src/tulip/memory/store.py

async def search_memories_semantic(
    self,
    query_embedding: list[float],
    limit: int = 10,
    threshold: float | None = None,
) -> list[SemanticSearchResult]:
    """
    Search user memories by semantic similarity.

    Requires: capabilities.semantic_search = True

    Args:
        query_embedding: Vector to search with
        limit: Maximum results
        threshold: Minimum similarity score (0.0-1.0)

    Returns:
        List of SemanticSearchResult sorted by similarity

    Example:
        query_vec = await embedder.embed("user preferences")
        results = await store.search_memories_semantic(
            query_embedding=query_vec,
            limit=5,
            threshold=0.7,
        )
        for r in results:
            print(f"{r.item.key}: {r.score:.2f} similar")
    """
    if not self._user_id:
        return []
    return await self._store.search_by_embedding(
        ("users", self._user_id, "memories"),
        query_embedding,
        limit,
        threshold,
    )

get_global async

get_global(key: str) -> Any | None

Get a global value.

Source code in .sdk/src/tulip/memory/store.py

async def get_global(self, key: str) -> Any | None:
    """Get a global value."""
    return await self._store.get(("global",), key)

set_global async

set_global(key: str, value: Any, metadata: dict[str, Any] | None = None) -> None

Set a global value.

Source code in .sdk/src/tulip/memory/store.py

async def set_global(
    self,
    key: str,
    value: Any,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Set a global value."""
    await self._store.put(("global",), key, value, metadata)

StoreItem dataclass

StoreItem(namespace: tuple[str, ...], key: str, value: Any, metadata: dict[str, Any], created_at: datetime, updated_at: datetime, version: int = 1)

An item stored in the store.

Attributes:

Name	Type	Description
namespace	tuple[str, ...]	The namespace tuple
key	str	The key within namespace
value	Any	The stored value
metadata	dict[str, Any]	Optional metadata
created_at	datetime	Creation timestamp
updated_at	datetime	Last update timestamp
version	int	Version counter for optimistic locking

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

Source code in .sdk/src/tulip/memory/store.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary."""
    return {
        "namespace": list(self.namespace),
        "key": self.key,
        "value": self.value,
        "metadata": self.metadata,
        "created_at": self.created_at.isoformat(),
        "updated_at": self.updated_at.isoformat(),
        "version": self.version,
    }

from_dict classmethod

from_dict(d: dict[str, Any]) -> StoreItem

Create from dictionary.

Source code in .sdk/src/tulip/memory/store.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> StoreItem:
    """Create from dictionary."""
    return cls(
        namespace=tuple(d["namespace"]),
        key=d["key"],
        value=d["value"],
        metadata=d.get("metadata", {}),
        created_at=datetime.fromisoformat(d["created_at"]),
        updated_at=datetime.fromisoformat(d["updated_at"]),
        version=d.get("version", 1),
    )

StoreCapabilities dataclass

StoreCapabilities(search: bool = False, semantic_search: bool = False, embedding_dimension: int | None = None, ttl: bool = False, list_namespaces: bool = False, batch_operations: bool = False, transactions: bool = False)

Capabilities supported by a store implementation.

Use this to discover what features a store supports.

StoreCapabilityError

StoreCapabilityError(capability: str, store_class: str, hint: str | None = None)

Bases: NotImplementedError

Raised when a store operation is requested that the backend does not support.

Carries structured context (capability, store_class, hint) so callers can branch on the capability or surface a clean message.

Subclasses :class:NotImplementedError so legacy except NotImplementedError blocks continue to catch it — this is strictly a richer payload, not a behaviour change.

Example::

try:
    results = await store.search(("users", uid), query="...")
except StoreCapabilityError as exc:
    log.info(
        "store %s does not support %s; falling back",
        exc.store_class,
        exc.capability,
    )
    results = await store.list_keys(("users", uid))

Source code in .sdk/src/tulip/memory/store.py

def __init__(
    self,
    capability: str,
    store_class: str,
    hint: str | None = None,
) -> None:
    self.capability = capability
    self.store_class = store_class
    self.hint = hint
    msg = f"{store_class} does not support {capability!r}"
    if hint:
        msg = f"{msg}. {hint}"
    super().__init__(msg)

SemanticSearchResult dataclass

SemanticSearchResult(item: StoreItem, score: float, distance: float | None = None)

Result from semantic (vector similarity) search.

Attributes:

Name	Type	Description
item	StoreItem	The matching store item
score	float	Similarity score (0.0 to 1.0, higher is more similar)
distance	float | None	Raw distance metric (interpretation depends on distance type)

Durable store backends

Both shipped in 2.2.0.

HolographicStore

HolographicStore(path: str = ':memory:', *, dim: int = _DEFAULT_DIM)

Bases: BaseStore

Local SQLite + FTS5 + HRR store — zero external infra.

Every value is kept as a row plus an FTS5 index entry and (when numpy is present) an HRR phase vector. search ranks by HRR associative similarity, fused with FTS5 lexical match; without numpy it is FTS5-only. A single connection guarded by a lock keeps it safe across the store's threads.

Source code in .sdk/src/tulip/memory/store_backends/holographic.py

def __init__(self, path: str = ":memory:", *, dim: int = _DEFAULT_DIM) -> None:
    self._dim = dim
    self._lock = threading.Lock()
    self._conn = sqlite3.connect(path, check_same_thread=False)
    self._conn.execute("PRAGMA journal_mode=WAL")
    self._conn.execute(
        "CREATE TABLE IF NOT EXISTS items ("
        "ns TEXT, key TEXT, value TEXT, metadata TEXT, created_at TEXT, "
        "updated_at TEXT, version INTEGER, content TEXT, hrr BLOB, "
        "PRIMARY KEY (ns, key))"
    )
    self._conn.execute(
        "CREATE VIRTUAL TABLE IF NOT EXISTS items_fts "
        "USING fts5(ns UNINDEXED, key UNINDEXED, content)"
    )
    self._conn.commit()

exists async

exists(namespace: tuple[str, ...], key: str) -> bool

Check if a key exists.

Source code in .sdk/src/tulip/memory/store.py

async def exists(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> bool:
    """Check if a key exists."""
    value = await self.get(namespace, key)
    return value is not None

get_item async

get_item(namespace: tuple[str, ...], key: str) -> StoreItem | None

Get full item with metadata.

Source code in .sdk/src/tulip/memory/store.py

async def get_item(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> StoreItem | None:
    """Get full item with metadata."""
    # Default implementation - subclasses can override
    value = await self.get(namespace, key)
    if value is None:
        return None
    now = datetime.now(UTC)
    return StoreItem(
        namespace=namespace,
        key=key,
        value=value,
        metadata={},
        created_at=now,
        updated_at=now,
    )

put_with_embedding async

put_with_embedding(namespace: tuple[str, ...], key: str, value: Any, embedding: list[float], metadata: dict[str, Any] | None = None) -> None

Store a value with its embedding vector for semantic search.

Optional method — backends advertise support via capabilities.semantic_search. Subclasses that support semantic search must override this method.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required
value	Any	Value to store (must be JSON-serializable)	required
embedding	list[float]	Vector embedding (e.g., from OpenAI, Cohere, etc.)	required
metadata	dict[str, Any] | None	Optional metadata for filtering	None

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Example

Get embedding from your provider

embedding = await embedder.embed("User prefers dark theme")

await store.put_with_embedding( ("users", user_id, "memories"), "theme_preference", {"theme": "dark", "reason": "easier on eyes"}, embedding=embedding, metadata={"category": "preferences"} )

Source code in .sdk/src/tulip/memory/store.py

async def put_with_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: Any,
    embedding: list[float],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Store a value with its embedding vector for semantic search.

    Optional method — backends advertise support via
    ``capabilities.semantic_search``. Subclasses that support
    semantic search must override this method.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace
        value: Value to store (must be JSON-serializable)
        embedding: Vector embedding (e.g., from OpenAI, Cohere, etc.)
        metadata: Optional metadata for filtering

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.

    Example:
        # Get embedding from your provider
        embedding = await embedder.embed("User prefers dark theme")

        await store.put_with_embedding(
            ("users", user_id, "memories"),
            "theme_preference",
            {"theme": "dark", "reason": "easier on eyes"},
            embedding=embedding,
            metadata={"category": "preferences"}
        )
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Use put() for plain storage, or pick a vector-capable backend.",
    )

search_by_embedding async

search_by_embedding(namespace: tuple[str, ...], query_embedding: list[float], limit: int = 10, threshold: float | None = None, metadata_filter: dict[str, Any] | None = None) -> list[SemanticSearchResult]

Search for similar items using vector similarity.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to search in	required
query_embedding	list[float]	Vector to compare against stored embeddings	required
limit	int	Maximum results to return	10
threshold	float | None	Minimum similarity score (0.0-1.0), or None for no threshold	None
metadata_filter	dict[str, Any] | None	Optional metadata constraints	None

Returns:

Type	Description
list[SemanticSearchResult]	List of SemanticSearchResult, sorted by similarity (highest first)

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def search_by_embedding(
    self,
    namespace: tuple[str, ...],
    query_embedding: list[float],
    limit: int = 10,
    threshold: float | None = None,
    metadata_filter: dict[str, Any] | None = None,
) -> list[SemanticSearchResult]:
    """
    Search for similar items using vector similarity.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Namespace to search in
        query_embedding: Vector to compare against stored embeddings
        limit: Maximum results to return
        threshold: Minimum similarity score (0.0-1.0), or None for no threshold
        metadata_filter: Optional metadata constraints

    Returns:
        List of SemanticSearchResult, sorted by similarity (highest first)

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Pick a vector-capable backend (e.g., pgvector, Qdrant).",
    )

get_embedding async

get_embedding(namespace: tuple[str, ...], key: str) -> list[float] | None

Get the embedding vector for a stored item.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required

Returns:

Type	Description
list[float] | None	Embedding vector or None if not found / no embedding stored.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def get_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> list[float] | None:
    """
    Get the embedding vector for a stored item.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace

    Returns:
        Embedding vector or None if not found / no embedding stored.

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
    )

put_batch async

put_batch(items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]) -> None

Store multiple items.

Backends that advertise capabilities.batch_operations should override this method to dispatch atomically. The default falls back to sequential put() calls so callers always get functional batching, just without atomicity guarantees.

Parameters:

Name	Type	Description	Default
items	list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]	List of (namespace, key, value, metadata) tuples	required

Source code in .sdk/src/tulip/memory/store.py

async def put_batch(
    self,
    items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]],
) -> None:
    """
    Store multiple items.

    Backends that advertise ``capabilities.batch_operations`` should
    override this method to dispatch atomically. The default falls
    back to sequential ``put()`` calls so callers always get
    functional batching, just without atomicity guarantees.

    Args:
        items: List of (namespace, key, value, metadata) tuples
    """
    # Sequential fallback — works for every backend, even those that
    # don't advertise batch_operations. Backends that *do* advertise
    # the capability override this method to dispatch in one
    # transaction / round-trip.
    for namespace, key, value, metadata in items:
        await self.put(namespace, key, value, metadata)

get_batch async

get_batch(keys: list[tuple[tuple[str, ...], str]]) -> dict[tuple[tuple[str, ...], str], Any]

Retrieve multiple items.

Parameters:

Name	Type	Description	Default
keys	list[tuple[tuple[str, ...], str]]	List of (namespace, key) tuples	required

Returns:

Type	Description
dict[tuple[tuple[str, ...], str], Any]	Dict mapping (namespace, key) to value

Source code in .sdk/src/tulip/memory/store.py

async def get_batch(
    self,
    keys: list[tuple[tuple[str, ...], str]],
) -> dict[tuple[tuple[str, ...], str], Any]:
    """
    Retrieve multiple items.

    Args:
        keys: List of (namespace, key) tuples

    Returns:
        Dict mapping (namespace, key) to value
    """
    results = {}
    for namespace, key in keys:
        value = await self.get(namespace, key)
        if value is not None:
            results[(namespace, key)] = value
    return results

list_namespaces async

list_namespaces(prefix: tuple[str, ...] | None = None, limit: int = 100) -> list[tuple[str, ...]]

List all namespaces.

Optional method — see capabilities.list_namespaces.

Parameters:

Name	Type	Description	Default
prefix	tuple[str, ...] | None	Optional prefix to filter namespaces	None
limit	int	Maximum namespaces to return	100

Returns:

Type	Description
list[tuple[str, ...]]	List of namespace tuples

Raises:

Type	Description
StoreCapabilityError	If the backend does not support namespace listing.

Source code in .sdk/src/tulip/memory/store.py

async def list_namespaces(
    self,
    prefix: tuple[str, ...] | None = None,
    limit: int = 100,
) -> list[tuple[str, ...]]:
    """
    List all namespaces.

    Optional method — see ``capabilities.list_namespaces``.

    Args:
        prefix: Optional prefix to filter namespaces
        limit: Maximum namespaces to return

    Returns:
        List of namespace tuples

    Raises:
        StoreCapabilityError: If the backend does not support
            namespace listing.
    """
    raise StoreCapabilityError(
        capability="list_namespaces",
        store_class=type(self).__name__,
    )

clear_namespace async

clear_namespace(namespace: tuple[str, ...]) -> int

Delete all items in a namespace.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to clear	required

Returns:

Type	Description
int	Number of items deleted

Source code in .sdk/src/tulip/memory/store.py

async def clear_namespace(
    self,
    namespace: tuple[str, ...],
) -> int:
    """
    Delete all items in a namespace.

    Args:
        namespace: Namespace to clear

    Returns:
        Number of items deleted
    """
    keys = await self.list_keys(namespace, limit=10000)
    count = 0
    for key in keys:
        if await self.delete(namespace, key):
            count += 1
    return count

PgMemory

PgMemory(dsn: str, *, table: str = 'tulip_memories', dim: int = _DEFAULT_PG_DIM, embedder: BaseEmbedding | None = None)

Bases: BaseStore

Postgres + pgvector memory store with per-tenant Row-Level Security.

namespace[0] is the tenant (the isolation boundary); the full namespace tuple scopes within a tenant (e.g. (tenant, "user", user_id)). The schema, RLS policy, and HNSW index are created on first use, so it works against a bare Postgres locally and a managed Aurora cluster identically.

Recall quality depends on the vector. Pass an embedder (any :class:~tulip.rag.embeddings.base.BaseEmbedding, e.g. brokered OpenAI text-embedding-3-small) for true semantic recall — the enterprise default. Without one it falls back to the HRR [cos φ, sin φ] encoding, which is lexical/associative (matches shared or hashed tokens), not trained semantics — fine offline, but it will not match paraphrases the way an embedding model does. Choose the embedder to match the value of recall.

Width and the ANN index. Without an embedder the pgvector column is 2·dim wide (the [cos φ, sin φ] pair), and pgvector will not build an HNSW index over more than :data:PGVECTOR_ANN_MAX_DIM dimensions. dim therefore defaults to 512 (a 1024-wide column) and a larger explicit dim is rejected at construction rather than at CREATE INDEX. An embedder wider than the limit is allowed — its width is the model's, not ours — but the table is then created without an ANN index and the constructor emits a RuntimeWarning saying so.

Source code in .sdk/src/tulip/memory/store_backends/postgresql.py

def __init__(
    self,
    dsn: str,
    *,
    table: str = "tulip_memories",
    dim: int = _DEFAULT_PG_DIM,
    embedder: BaseEmbedding | None = None,
) -> None:
    if dim < 1:
        raise ValueError(f"dim must be >= 1, got {dim}")
    self._dsn = dsn
    self._table = _validate_ident(table, "table")
    self._dim = dim
    self._embedder = embedder
    # A real embedder fixes the column width; HRR [cos, sin] doubles `dim`.
    self._vdim = embedder.dimension if embedder is not None else 2 * dim
    # Fail here, not thousands of lines into a run with a raw driver error:
    # a `dim` we control that cannot carry an ANN index is a caller mistake.
    if embedder is None and self._vdim > PGVECTOR_ANN_MAX_DIM:
        raise ValueError(
            f"PgMemory(dim={dim}) is not indexable: the HRR encoding stores "
            f"[cos φ, sin φ], so the pgvector column would be 2·{dim} = "
            f"{self._vdim} dimensions wide, over pgvector's "
            f"{PGVECTOR_ANN_MAX_DIM}-dimension limit for an HNSW index "
            f"(CREATE INDEX would raise ProgramLimitExceededError). Pass "
            f"dim <= {PGVECTOR_ANN_MAX_DIM // 2} (the default is "
            f"{_DEFAULT_PG_DIM}), or pass an embedder."
        )
    # An embedder's width is *not* ours to choose (text-embedding-3-large is
    # 3072). Rather than reject a legitimate model we keep the column and skip
    # the index — but say so loudly, because a silent sequential scan is
    # exactly the failure this store had.
    self._ann = self._vdim <= PGVECTOR_ANN_MAX_DIM
    if not self._ann:
        warnings.warn(
            f"PgMemory: embedder dimension {self._vdim} exceeds pgvector's "
            f"{PGVECTOR_ANN_MAX_DIM}-dimension limit for an HNSW index, so "
            f"table {self._table!r} will be created WITHOUT an ANN index — "
            f"every search is a sequential scan. Use an embedder with "
            f"<= {PGVECTOR_ANN_MAX_DIM} dimensions (or a model that supports "
            f"shortening its output) for indexed recall.",
            RuntimeWarning,
            stacklevel=2,
        )
    self._pool: Pool | None = None
    # Serialises first use so two concurrent calls cannot each build a pool
    # (and each run schema creation).
    self._pool_lock = asyncio.Lock()

exists async

exists(namespace: tuple[str, ...], key: str) -> bool

Check if a key exists.

Source code in .sdk/src/tulip/memory/store.py

async def exists(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> bool:
    """Check if a key exists."""
    value = await self.get(namespace, key)
    return value is not None

get_item async

get_item(namespace: tuple[str, ...], key: str) -> StoreItem | None

Get full item with metadata.

Source code in .sdk/src/tulip/memory/store.py

async def get_item(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> StoreItem | None:
    """Get full item with metadata."""
    # Default implementation - subclasses can override
    value = await self.get(namespace, key)
    if value is None:
        return None
    now = datetime.now(UTC)
    return StoreItem(
        namespace=namespace,
        key=key,
        value=value,
        metadata={},
        created_at=now,
        updated_at=now,
    )

put_with_embedding async

put_with_embedding(namespace: tuple[str, ...], key: str, value: Any, embedding: list[float], metadata: dict[str, Any] | None = None) -> None

Store a value with its embedding vector for semantic search.

Optional method — backends advertise support via capabilities.semantic_search. Subclasses that support semantic search must override this method.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required
value	Any	Value to store (must be JSON-serializable)	required
embedding	list[float]	Vector embedding (e.g., from OpenAI, Cohere, etc.)	required
metadata	dict[str, Any] | None	Optional metadata for filtering	None

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Example

Get embedding from your provider

embedding = await embedder.embed("User prefers dark theme")

await store.put_with_embedding( ("users", user_id, "memories"), "theme_preference", {"theme": "dark", "reason": "easier on eyes"}, embedding=embedding, metadata={"category": "preferences"} )

Source code in .sdk/src/tulip/memory/store.py

async def put_with_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: Any,
    embedding: list[float],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Store a value with its embedding vector for semantic search.

    Optional method — backends advertise support via
    ``capabilities.semantic_search``. Subclasses that support
    semantic search must override this method.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace
        value: Value to store (must be JSON-serializable)
        embedding: Vector embedding (e.g., from OpenAI, Cohere, etc.)
        metadata: Optional metadata for filtering

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.

    Example:
        # Get embedding from your provider
        embedding = await embedder.embed("User prefers dark theme")

        await store.put_with_embedding(
            ("users", user_id, "memories"),
            "theme_preference",
            {"theme": "dark", "reason": "easier on eyes"},
            embedding=embedding,
            metadata={"category": "preferences"}
        )
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Use put() for plain storage, or pick a vector-capable backend.",
    )

search_by_embedding async

search_by_embedding(namespace: tuple[str, ...], query_embedding: list[float], limit: int = 10, threshold: float | None = None, metadata_filter: dict[str, Any] | None = None) -> list[SemanticSearchResult]

Search for similar items using vector similarity.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to search in	required
query_embedding	list[float]	Vector to compare against stored embeddings	required
limit	int	Maximum results to return	10
threshold	float | None	Minimum similarity score (0.0-1.0), or None for no threshold	None
metadata_filter	dict[str, Any] | None	Optional metadata constraints	None

Returns:

Type	Description
list[SemanticSearchResult]	List of SemanticSearchResult, sorted by similarity (highest first)

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def search_by_embedding(
    self,
    namespace: tuple[str, ...],
    query_embedding: list[float],
    limit: int = 10,
    threshold: float | None = None,
    metadata_filter: dict[str, Any] | None = None,
) -> list[SemanticSearchResult]:
    """
    Search for similar items using vector similarity.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Namespace to search in
        query_embedding: Vector to compare against stored embeddings
        limit: Maximum results to return
        threshold: Minimum similarity score (0.0-1.0), or None for no threshold
        metadata_filter: Optional metadata constraints

    Returns:
        List of SemanticSearchResult, sorted by similarity (highest first)

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Pick a vector-capable backend (e.g., pgvector, Qdrant).",
    )

get_embedding async

get_embedding(namespace: tuple[str, ...], key: str) -> list[float] | None

Get the embedding vector for a stored item.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required

Returns:

Type	Description
list[float] | None	Embedding vector or None if not found / no embedding stored.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def get_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> list[float] | None:
    """
    Get the embedding vector for a stored item.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace

    Returns:
        Embedding vector or None if not found / no embedding stored.

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
    )

put_batch async

put_batch(items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]) -> None

Store multiple items.

Backends that advertise capabilities.batch_operations should override this method to dispatch atomically. The default falls back to sequential put() calls so callers always get functional batching, just without atomicity guarantees.

Parameters:

Name	Type	Description	Default
items	list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]	List of (namespace, key, value, metadata) tuples	required

Source code in .sdk/src/tulip/memory/store.py

async def put_batch(
    self,
    items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]],
) -> None:
    """
    Store multiple items.

    Backends that advertise ``capabilities.batch_operations`` should
    override this method to dispatch atomically. The default falls
    back to sequential ``put()`` calls so callers always get
    functional batching, just without atomicity guarantees.

    Args:
        items: List of (namespace, key, value, metadata) tuples
    """
    # Sequential fallback — works for every backend, even those that
    # don't advertise batch_operations. Backends that *do* advertise
    # the capability override this method to dispatch in one
    # transaction / round-trip.
    for namespace, key, value, metadata in items:
        await self.put(namespace, key, value, metadata)

get_batch async

get_batch(keys: list[tuple[tuple[str, ...], str]]) -> dict[tuple[tuple[str, ...], str], Any]

Retrieve multiple items.

Parameters:

Name	Type	Description	Default
keys	list[tuple[tuple[str, ...], str]]	List of (namespace, key) tuples	required

Returns:

Type	Description
dict[tuple[tuple[str, ...], str], Any]	Dict mapping (namespace, key) to value

Source code in .sdk/src/tulip/memory/store.py

async def get_batch(
    self,
    keys: list[tuple[tuple[str, ...], str]],
) -> dict[tuple[tuple[str, ...], str], Any]:
    """
    Retrieve multiple items.

    Args:
        keys: List of (namespace, key) tuples

    Returns:
        Dict mapping (namespace, key) to value
    """
    results = {}
    for namespace, key in keys:
        value = await self.get(namespace, key)
        if value is not None:
            results[(namespace, key)] = value
    return results

list_namespaces async

list_namespaces(prefix: tuple[str, ...] | None = None, limit: int = 100) -> list[tuple[str, ...]]

List all namespaces.

Optional method — see capabilities.list_namespaces.

Parameters:

Name	Type	Description	Default
prefix	tuple[str, ...] | None	Optional prefix to filter namespaces	None
limit	int	Maximum namespaces to return	100

Returns:

Type	Description
list[tuple[str, ...]]	List of namespace tuples

Raises:

Type	Description
StoreCapabilityError	If the backend does not support namespace listing.

Source code in .sdk/src/tulip/memory/store.py

async def list_namespaces(
    self,
    prefix: tuple[str, ...] | None = None,
    limit: int = 100,
) -> list[tuple[str, ...]]:
    """
    List all namespaces.

    Optional method — see ``capabilities.list_namespaces``.

    Args:
        prefix: Optional prefix to filter namespaces
        limit: Maximum namespaces to return

    Returns:
        List of namespace tuples

    Raises:
        StoreCapabilityError: If the backend does not support
            namespace listing.
    """
    raise StoreCapabilityError(
        capability="list_namespaces",
        store_class=type(self).__name__,
    )

clear_namespace async

clear_namespace(namespace: tuple[str, ...]) -> int

Delete all items in a namespace.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to clear	required

Returns:

Type	Description
int	Number of items deleted

Source code in .sdk/src/tulip/memory/store.py

async def clear_namespace(
    self,
    namespace: tuple[str, ...],
) -> int:
    """
    Delete all items in a namespace.

    Args:
        namespace: Namespace to clear

    Returns:
        Number of items deleted
    """
    keys = await self.list_keys(namespace, limit=10000)
    count = 0
    for key in keys:
        if await self.delete(namespace, key):
            count += 1
    return count

Long-term memory manager

LLMMemoryManager is the default — it uses an auxiliary model to extract and categorize memories at session end and retrieves the top-k relevant entries at session start. NoopMemoryManager is the pass-through used in tests.

BaseMemoryManager

Bases: ABC

Abstract base for long-term memory managers.

Subclasses implement :meth:extract to decide what is worth remembering; the base class handles retrieval, injection, and the session-start / session-end lifecycle hooks that the agent calls automatically.

Two concrete implementations are provided:

• :class:NoopMemoryManager — no-op, useful for testing.
• :class:LLMMemoryManager — persists to any :class:BaseStore backend; accepts an optional LLM-backed extraction function.

extract abstractmethod async

extract(messages: list[Message]) -> list[Memory]

Extract durable memories from a finished conversation.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The full message history for the completed session.	required

Returns:

Type	Description
list[Memory]	List of :class:Memory objects to persist. May be empty.

Source code in .sdk/src/tulip/memory/manager.py

@abstractmethod
async def extract(self, messages: list[Message]) -> list[Memory]:
    """Extract durable memories from a finished conversation.

    Args:
        messages: The full message history for the completed session.

    Returns:
        List of :class:`Memory` objects to persist.  May be empty.
    """
    ...

retrieve abstractmethod async

retrieve(limit: int = 20) -> list[Memory]

Retrieve stored memories for injection at session start.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of memories to return.	20

Returns:

Type	Description
list[Memory]	List of :class:Memory objects, most recently updated first.

Source code in .sdk/src/tulip/memory/manager.py

@abstractmethod
async def retrieve(self, limit: int = 20) -> list[Memory]:
    """Retrieve stored memories for injection at session start.

    Args:
        limit: Maximum number of memories to return.

    Returns:
        List of :class:`Memory` objects, most recently updated first.
    """
    ...

save abstractmethod async

save(memories: list[Memory]) -> None

Persist a list of memories to the backing store.

Implementations should upsert by key — re-extracting the same fact updates the record rather than creating a duplicate.

Parameters:

Name	Type	Description	Default
memories	list[Memory]	Memories to persist.	required

Source code in .sdk/src/tulip/memory/manager.py

@abstractmethod
async def save(self, memories: list[Memory]) -> None:
    """Persist a list of memories to the backing store.

    Implementations should upsert by key — re-extracting the same
    fact updates the record rather than creating a duplicate.

    Args:
        memories: Memories to persist.
    """
    ...

on_session_start async

on_session_start(state: AgentState) -> AgentState

Retrieve memories and inject them into the agent state.

Called by the agent runtime at the start of every invocation, after on_before_invocation hooks but before the first model call. The default implementation retrieves all stored memories and prepends a formatted system message to state.messages.

Parameters:

Name	Type	Description	Default
state	AgentState	Current agent state (just-created or loaded from checkpointer).	required

Returns:

Type	Description
AgentState	Possibly-modified state with memory context injected.

Source code in .sdk/src/tulip/memory/manager.py

async def on_session_start(self, state: AgentState) -> AgentState:
    """Retrieve memories and inject them into the agent state.

    Called by the agent runtime at the start of every invocation,
    after ``on_before_invocation`` hooks but before the first model
    call.  The default implementation retrieves all stored memories
    and prepends a formatted system message to ``state.messages``.

    Args:
        state: Current agent state (just-created or loaded from
            checkpointer).

    Returns:
        Possibly-modified state with memory context injected.
    """
    memories = await self.retrieve()
    if not memories:
        return state

    from tulip.observability.emit import emit  # noqa: PLC0415

    injected_state = _inject_memories_into_state(state, memories)

    await emit(
        "memory.manager.injected",
        memory_count=len(memories),
        types=[m.type.value for m in memories],
    )

    return injected_state

on_session_end async

on_session_end(state: AgentState) -> None

Extract memories from the finished session and save them.

Called by the agent runtime in the finally block of every invocation, after on_after_invocation hooks but before the final checkpoint.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state with the complete message history.	required

Source code in .sdk/src/tulip/memory/manager.py

async def on_session_end(self, state: AgentState) -> None:
    """Extract memories from the finished session and save them.

    Called by the agent runtime in the ``finally`` block of every
    invocation, after ``on_after_invocation`` hooks but before the
    final checkpoint.

    Args:
        state: Final agent state with the complete message history.
    """
    memories = await self.extract(list(state.messages))
    if not memories:
        return

    await self.save(memories)

    from tulip.observability.emit import emit  # noqa: PLC0415

    await emit(
        "memory.manager.extracted",
        memory_count=len(memories),
        types=[m.type.value for m in memories],
        keys=[m.key for m in memories],
    )

LLMMemoryManager

LLMMemoryManager(store: BaseStore, *, extract_fn: ExtractFn | None = None, namespace_prefix: tuple[str, ...] = ('tulip_memory',), max_memories: int = 50, retrieve_limit: int = 20)

Bases: BaseMemoryManager

Memory manager backed by any :class:~tulip.memory.store.BaseStore.

Extraction uses either a caller-supplied async function (extract_fn) or a built-in heuristic that pattern-matches common conversational signals (corrections, confirmations, role disclosures). Supply extract_fn for production use; the heuristic is adequate for demos and tests.

Parameters:

Name	Type	Description	Default
store	BaseStore	Any :class:~tulip.memory.store.BaseStore implementation. InMemoryStore works for development; use RedisBackend, PostgreSQLBackend, etc. for production.	required
extract_fn	ExtractFn | None	Optional async callable (messages: list[Message]) -> list[Memory]. When provided, the heuristic is bypassed.	None
namespace_prefix	tuple[str, ...]	Store namespace prefix. Scope per user or tenant by passing ("tenants", tenant_id). Default ("tulip_memory",).	('tulip_memory',)
max_memories	int	Hard cap on the total number of memories kept per type. Oldest entries are pruned when the limit is reached.	50
retrieve_limit	int	Maximum memories returned by :meth:retrieve.	20

Example::

from tulip.memory.manager import LLMMemoryManager
from tulip.memory.store import InMemoryStore

manager = LLMMemoryManager(
    store=InMemoryStore(),
    extract_fn=my_llm_extractor,
    namespace_prefix=("users", user_id),
)

agent = Agent(model="anthropic:claude-sonnet-4-6", memory_manager=manager)

Source code in .sdk/src/tulip/memory/manager.py

def __init__(
    self,
    store: BaseStore,
    *,
    extract_fn: ExtractFn | None = None,
    namespace_prefix: tuple[str, ...] = ("tulip_memory",),
    max_memories: int = 50,
    retrieve_limit: int = 20,
) -> None:
    self.store = store
    self.extract_fn = extract_fn
    self.namespace_prefix = namespace_prefix
    self.max_memories = max_memories
    self.retrieve_limit = retrieve_limit

extract async

extract(messages: list[Message]) -> list[Memory]

Extract memories from a message list.

Uses extract_fn when provided; otherwise applies the built-in heuristic.

Source code in .sdk/src/tulip/memory/manager.py

async def extract(self, messages: list[Message]) -> list[Memory]:
    """Extract memories from a message list.

    Uses ``extract_fn`` when provided; otherwise applies the
    built-in heuristic.
    """
    if self.extract_fn is not None:
        return await self.extract_fn(messages)
    return _heuristic_extract(messages)

retrieve async

retrieve(limit: int = 20) -> list[Memory]

Retrieve all stored memories across every type.

Returns memories sorted by most-recently-updated first, up to limit entries total.

Source code in .sdk/src/tulip/memory/manager.py

async def retrieve(self, limit: int = 20) -> list[Memory]:
    """Retrieve all stored memories across every type.

    Returns memories sorted by most-recently-updated first, up to
    ``limit`` entries total.
    """
    memories: list[Memory] = []

    for memory_type in MemoryType:
        ns = self._ns(memory_type)
        try:
            items = await self.store.search(ns, query=None, limit=limit)
        except Exception:  # noqa: BLE001
            # Gracefully fall back for backends that don't support search.
            keys = await self.store.list_keys(ns, limit=limit)
            items = []
            for k in keys:
                raw = await self.store.get(ns, k)
                if raw is not None:
                    from datetime import UTC, datetime  # noqa: PLC0415

                    from tulip.memory.store import StoreItem  # noqa: PLC0415

                    now = datetime.now(UTC)
                    items.append(
                        StoreItem(
                            namespace=ns,
                            key=k,
                            value=raw,
                            metadata={},
                            created_at=now,
                            updated_at=now,
                        )
                    )

        for item in items:
            try:
                memories.append(Memory.from_store_value(item.value))
            except (KeyError, ValueError):
                pass

    # Sort newest first by updated_at (best-effort — not all items carry it).
    memories.sort(
        key=lambda m: m.metadata.get("updated_at", ""),
        reverse=True,
    )
    return memories[:limit]

save async

save(memories: list[Memory]) -> None

Upsert memories into the backing store.

Memories with the same key and type overwrite the previous entry — no duplicates accumulate.

Source code in .sdk/src/tulip/memory/manager.py

async def save(self, memories: list[Memory]) -> None:
    """Upsert memories into the backing store.

    Memories with the same ``key`` and ``type`` overwrite the
    previous entry — no duplicates accumulate.
    """
    from datetime import UTC, datetime  # noqa: PLC0415

    now = datetime.now(UTC).isoformat()

    for memory in memories:
        ns = self._ns(memory.type)
        value = memory.to_store_value()
        value["metadata"]["updated_at"] = now

        await self.store.put(
            ns,
            memory.key,
            value,
            metadata={"type": memory.type.value, "updated_at": now},
        )

on_session_start async

on_session_start(state: AgentState) -> AgentState

Retrieve memories and inject them into the agent state.

Called by the agent runtime at the start of every invocation, after on_before_invocation hooks but before the first model call. The default implementation retrieves all stored memories and prepends a formatted system message to state.messages.

Parameters:

Name	Type	Description	Default
state	AgentState	Current agent state (just-created or loaded from checkpointer).	required

Returns:

Type	Description
AgentState	Possibly-modified state with memory context injected.

Source code in .sdk/src/tulip/memory/manager.py

async def on_session_start(self, state: AgentState) -> AgentState:
    """Retrieve memories and inject them into the agent state.

    Called by the agent runtime at the start of every invocation,
    after ``on_before_invocation`` hooks but before the first model
    call.  The default implementation retrieves all stored memories
    and prepends a formatted system message to ``state.messages``.

    Args:
        state: Current agent state (just-created or loaded from
            checkpointer).

    Returns:
        Possibly-modified state with memory context injected.
    """
    memories = await self.retrieve()
    if not memories:
        return state

    from tulip.observability.emit import emit  # noqa: PLC0415

    injected_state = _inject_memories_into_state(state, memories)

    await emit(
        "memory.manager.injected",
        memory_count=len(memories),
        types=[m.type.value for m in memories],
    )

    return injected_state

on_session_end async

on_session_end(state: AgentState) -> None

Extract memories from the finished session and save them.

Called by the agent runtime in the finally block of every invocation, after on_after_invocation hooks but before the final checkpoint.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state with the complete message history.	required

Source code in .sdk/src/tulip/memory/manager.py

async def on_session_end(self, state: AgentState) -> None:
    """Extract memories from the finished session and save them.

    Called by the agent runtime in the ``finally`` block of every
    invocation, after ``on_after_invocation`` hooks but before the
    final checkpoint.

    Args:
        state: Final agent state with the complete message history.
    """
    memories = await self.extract(list(state.messages))
    if not memories:
        return

    await self.save(memories)

    from tulip.observability.emit import emit  # noqa: PLC0415

    await emit(
        "memory.manager.extracted",
        memory_count=len(memories),
        types=[m.type.value for m in memories],
        keys=[m.key for m in memories],
    )

NoopMemoryManager

Bases: BaseMemoryManager

Pass-through memory manager — stores and retrieves nothing.

Useful as a test double or as a placeholder when you want the agent-wiring (memory_manager= kwarg) without actual persistence.

Memory dataclass

Memory(type: MemoryType, key: str, content: str, metadata: dict[str, Any] = dict())

A single durable memory entry.

Attributes:

Name	Type	Description
type	MemoryType	Semantic category (user / feedback / project / reference).
key	str	Stable logical name within the category. The store uses this as the key, so re-extracting the same fact under the same key updates the record rather than creating a duplicate.
content	str	The actual fact or rule, as a human-readable string.
metadata	dict[str, Any]	Arbitrary extra fields — confidence, source, why, how_to_apply, ISO timestamps, etc.

to_store_value

to_store_value() -> dict[str, Any]

Serialise to a JSON-compatible dict for the store.

Source code in .sdk/src/tulip/memory/manager.py

def to_store_value(self) -> dict[str, Any]:
    """Serialise to a JSON-compatible dict for the store."""
    return {
        "type": self.type.value,
        "key": self.key,
        "content": self.content,
        "metadata": self.metadata,
    }

from_store_value classmethod

from_store_value(value: dict[str, Any]) -> Memory

Deserialise from a store value dict.

Source code in .sdk/src/tulip/memory/manager.py

@classmethod
def from_store_value(cls, value: dict[str, Any]) -> Memory:
    """Deserialise from a store value dict."""
    return cls(
        type=MemoryType(value["type"]),
        key=value["key"],
        content=value["content"],
        metadata=value.get("metadata", {}),
    )

MemoryType

Bases: StrEnum

Semantic category for a stored memory.

Delta checkpointing

Storage-efficient checkpointer that persists only the diff between consecutive states (the module targets ~77% storage reduction on long conversations; actual savings depend on workload). Layered on top of any DeltaStorage backend.

DeltaCheckpointer

DeltaCheckpointer(storage: DeltaStorage | None = None, max_chain_depth: int = 5, compression_level: int = 6)

Delta-based checkpointer for efficient state persistence.

This checkpointer stores only the differences between states, achieving significant storage savings. It maintains a chain of deltas pointing back to a full checkpoint.

Features: - Delta compression: Only store changes from parent state - Chain depth limiting: Force full checkpoint after N deltas - zlib compression: Additional compression of stored data - ~77% storage reduction in typical usage

Parameters:

Name	Type	Description	Default
storage	DeltaStorage | None	Storage backend for checkpoints	None
max_chain_depth	int	Maximum number of deltas before forcing full checkpoint	5
compression_level	int	zlib compression level (0-9, higher = better compression)	6

Source code in .sdk/src/tulip/memory/delta.py

def __init__(
    self,
    storage: DeltaStorage | None = None,
    max_chain_depth: int = 5,
    compression_level: int = 6,
):
    self.storage = storage or InMemoryDeltaStorage()
    self.max_chain_depth = max_chain_depth
    self.compression_level = compression_level
    self._state_cache: dict[str, dict[str, Any]] = {}  # Thread -> checkpoint_id -> state

save async

save(state: AgentState, thread_id: str, checkpoint_id: str | None = None) -> str

Save agent state with delta compression.

If a parent checkpoint exists and chain depth allows, only the delta from parent is stored. Otherwise, a full checkpoint is created.

Parameters:

Name	Type	Description	Default
state	AgentState	Current agent state	required
thread_id	str	Thread identifier	required
checkpoint_id	str | None	Optional specific checkpoint ID	None

Returns:

Type	Description
str	Checkpoint ID for the saved state

Source code in .sdk/src/tulip/memory/delta.py

async def save(
    self,
    state: AgentState,
    thread_id: str,
    checkpoint_id: str | None = None,
) -> str:
    """
    Save agent state with delta compression.

    If a parent checkpoint exists and chain depth allows,
    only the delta from parent is stored. Otherwise, a full
    checkpoint is created.

    Args:
        state: Current agent state
        thread_id: Thread identifier
        checkpoint_id: Optional specific checkpoint ID

    Returns:
        Checkpoint ID for the saved state
    """
    checkpoint_id = checkpoint_id or uuid4().hex

    # Get current state as dict
    current_data = state.to_checkpoint()

    # Check for existing checkpoints
    existing = await self.storage.list_checkpoints(thread_id, limit=1)

    if not existing:
        # No parent - create full checkpoint
        checkpoint = await self._create_full_checkpoint(current_data, thread_id, checkpoint_id)
    else:
        parent_meta = existing[0]

        # Check if we should create a full checkpoint due to chain depth
        if parent_meta.chain_depth >= self.max_chain_depth:
            checkpoint = await self._create_full_checkpoint(
                current_data, thread_id, checkpoint_id
            )
        else:
            # Create delta checkpoint
            parent_checkpoint = await self.storage.retrieve(
                thread_id, parent_meta.checkpoint_id
            )
            if parent_checkpoint is None:
                # Parent not found, create full
                checkpoint = await self._create_full_checkpoint(
                    current_data, thread_id, checkpoint_id
                )
            else:
                # Load parent state and compute delta
                parent_data = await self._load_full_state(thread_id, parent_meta.checkpoint_id)
                if parent_data is None:
                    checkpoint = await self._create_full_checkpoint(
                        current_data, thread_id, checkpoint_id
                    )
                else:
                    checkpoint = await self._create_delta_checkpoint(
                        current_data,
                        parent_data,
                        thread_id,
                        checkpoint_id,
                        parent_meta.checkpoint_id,
                        parent_meta.chain_depth + 1,
                    )

    await self.storage.store(thread_id, checkpoint_id, checkpoint)

    # Update cache
    if thread_id not in self._state_cache:
        self._state_cache[thread_id] = {}
    self._state_cache[thread_id][checkpoint_id] = current_data

    return checkpoint_id

load async

load(thread_id: str, checkpoint_id: str | None = None) -> AgentState | None

Load agent state from checkpoint.

If loading a delta checkpoint, reconstructs full state by walking the delta chain back to a full checkpoint.

Parameters:

Name	Type	Description	Default
thread_id	str	Thread identifier	required
checkpoint_id	str | None	Specific checkpoint ID (latest if None)	None

Returns:

Type	Description
AgentState | None	Restored AgentState or None if not found

Source code in .sdk/src/tulip/memory/delta.py

async def load(
    self,
    thread_id: str,
    checkpoint_id: str | None = None,
) -> AgentState | None:
    """
    Load agent state from checkpoint.

    If loading a delta checkpoint, reconstructs full state
    by walking the delta chain back to a full checkpoint.

    Args:
        thread_id: Thread identifier
        checkpoint_id: Specific checkpoint ID (latest if None)

    Returns:
        Restored AgentState or None if not found
    """
    from tulip.core.state import AgentState

    if checkpoint_id is None:
        # Get latest checkpoint
        checkpoints = await self.storage.list_checkpoints(thread_id, limit=1)
        if not checkpoints:
            return None
        checkpoint_id = checkpoints[0].checkpoint_id

    # Try cache first
    if thread_id in self._state_cache:
        if checkpoint_id in self._state_cache[thread_id]:
            return AgentState.from_checkpoint(self._state_cache[thread_id][checkpoint_id])

    # Load and reconstruct state
    state_data = await self._load_full_state(thread_id, checkpoint_id)
    if state_data is None:
        return None

    return AgentState.from_checkpoint(state_data)

list_checkpoints async

list_checkpoints(thread_id: str, limit: int = 10) -> list[str]

List available checkpoint IDs.

Parameters:

Name	Type	Description	Default
thread_id	str	Thread identifier	required
limit	int	Maximum number to return	10

Returns:

Type	Description
list[str]	List of checkpoint IDs, newest first

Source code in .sdk/src/tulip/memory/delta.py

async def list_checkpoints(
    self,
    thread_id: str,
    limit: int = 10,
) -> list[str]:
    """
    List available checkpoint IDs.

    Args:
        thread_id: Thread identifier
        limit: Maximum number to return

    Returns:
        List of checkpoint IDs, newest first
    """
    metadata_list = await self.storage.list_checkpoints(thread_id, limit)
    return [m.checkpoint_id for m in metadata_list]

get_metadata async

get_metadata(thread_id: str, checkpoint_id: str) -> CheckpointMetadata | None

Get metadata for a specific checkpoint.

Source code in .sdk/src/tulip/memory/delta.py

async def get_metadata(
    self,
    thread_id: str,
    checkpoint_id: str,
) -> CheckpointMetadata | None:
    """Get metadata for a specific checkpoint."""
    checkpoint = await self.storage.retrieve(thread_id, checkpoint_id)
    return checkpoint.metadata if checkpoint else None

delete async

delete(thread_id: str, checkpoint_id: str | None = None) -> bool

Delete checkpoint(s).

Source code in .sdk/src/tulip/memory/delta.py

async def delete(
    self,
    thread_id: str,
    checkpoint_id: str | None = None,
) -> bool:
    """Delete checkpoint(s)."""
    # Clear cache
    if thread_id in self._state_cache:
        if checkpoint_id:
            self._state_cache[thread_id].pop(checkpoint_id, None)
        else:
            del self._state_cache[thread_id]

    return await self.storage.delete(thread_id, checkpoint_id)

get_storage_stats async

get_storage_stats(thread_id: str) -> dict[str, Any]

Get storage statistics for a thread.

Returns dict with: - total_checkpoints: Number of checkpoints - total_size: Uncompressed size - compressed_size: Compressed size - compression_ratio: Overall compression ratio - full_checkpoints: Number of full checkpoints - delta_checkpoints: Number of delta checkpoints

Source code in .sdk/src/tulip/memory/delta.py

async def get_storage_stats(
    self,
    thread_id: str,
) -> dict[str, Any]:
    """
    Get storage statistics for a thread.

    Returns dict with:
    - total_checkpoints: Number of checkpoints
    - total_size: Uncompressed size
    - compressed_size: Compressed size
    - compression_ratio: Overall compression ratio
    - full_checkpoints: Number of full checkpoints
    - delta_checkpoints: Number of delta checkpoints
    """
    metadata_list = await self.storage.list_checkpoints(thread_id, limit=1000)

    if not metadata_list:
        return {
            "total_checkpoints": 0,
            "total_size": 0,
            "compressed_size": 0,
            "compression_ratio": 1.0,
            "full_checkpoints": 0,
            "delta_checkpoints": 0,
        }

    total_size = sum(m.size_bytes for m in metadata_list)
    compressed_size = sum(m.compressed_size_bytes for m in metadata_list)
    full_count = sum(1 for m in metadata_list if m.is_full)
    delta_count = sum(1 for m in metadata_list if not m.is_full)

    return {
        "total_checkpoints": len(metadata_list),
        "total_size": total_size,
        "compressed_size": compressed_size,
        "compression_ratio": total_size / compressed_size if compressed_size > 0 else 1.0,
        "full_checkpoints": full_count,
        "delta_checkpoints": delta_count,
    }

DeltaCheckpoint dataclass

DeltaCheckpoint(metadata: CheckpointMetadata, data: bytes, is_delta: bool = True)

A delta checkpoint containing changes from parent.

compression_ratio property

compression_ratio: float

Calculate compression ratio (uncompressed / compressed).

CheckpointMetadata dataclass

CheckpointMetadata(checkpoint_id: str, thread_id: str, parent_id: str | None = None, is_full: bool = True, chain_depth: int = 0, created_at: datetime = (lambda: datetime.now(UTC))(), size_bytes: int = 0, compressed_size_bytes: int = 0)

Metadata for a checkpoint.

DeltaStorage

Bases: ABC

Abstract storage backend for delta checkpoints.

store abstractmethod async

store(thread_id: str, checkpoint_id: str, checkpoint: DeltaCheckpoint) -> None

Store a checkpoint.

Source code in .sdk/src/tulip/memory/delta.py

@abstractmethod
async def store(
    self,
    thread_id: str,
    checkpoint_id: str,
    checkpoint: DeltaCheckpoint,
) -> None:
    """Store a checkpoint."""
    ...

retrieve abstractmethod async

retrieve(thread_id: str, checkpoint_id: str) -> DeltaCheckpoint | None

Retrieve a checkpoint.

Source code in .sdk/src/tulip/memory/delta.py

@abstractmethod
async def retrieve(
    self,
    thread_id: str,
    checkpoint_id: str,
) -> DeltaCheckpoint | None:
    """Retrieve a checkpoint."""
    ...

list_checkpoints abstractmethod async

list_checkpoints(thread_id: str, limit: int = 10) -> list[CheckpointMetadata]

List checkpoint metadata, newest first.

Source code in .sdk/src/tulip/memory/delta.py

@abstractmethod
async def list_checkpoints(
    self,
    thread_id: str,
    limit: int = 10,
) -> list[CheckpointMetadata]:
    """List checkpoint metadata, newest first."""
    ...

delete abstractmethod async

delete(thread_id: str, checkpoint_id: str | None = None) -> bool

Delete checkpoint(s).

Source code in .sdk/src/tulip/memory/delta.py

@abstractmethod
async def delete(
    self,
    thread_id: str,
    checkpoint_id: str | None = None,
) -> bool:
    """Delete checkpoint(s)."""
    ...

InMemoryDeltaStorage

InMemoryDeltaStorage()

Bases: DeltaStorage

In-memory storage for delta checkpoints (for testing).

Source code in .sdk/src/tulip/memory/delta.py

def __init__(self) -> None:
    self._storage: dict[str, dict[str, DeltaCheckpoint]] = {}

Registry

String-based checkpointer lookup — used when configuration passes a provider name (e.g. "s3", "redis") instead of an instance. Custom backends register themselves via register_checkpointer.

get_checkpointer

get_checkpointer(checkpointer_string: str, **kwargs: Any) -> BaseCheckpointer

Get a checkpointer from a string identifier.

Format: "provider" or "provider:config_hint"

The config_hint is provider-specific and is passed as a keyword argument.

Examples:

• "memory" -> MemoryCheckpointer
• "file:./checkpoints" -> FileCheckpointer(base_dir="./checkpoints")
• "redis:localhost:6379" -> RedisCheckpointer(url="redis://localhost:6379")
• "postgresql:mydb" -> PostgreSQLCheckpointer(database="mydb")
• "mysql:mydb" -> MySQLCheckpointer(database="mydb")
• "opensearch" -> OpenSearchCheckpointer()
• "s3:my-bucket" -> S3Backend(bucket="my-bucket")

Parameters:

Name	Type	Description	Default
checkpointer_string	str	Checkpointer identifier	required
**kwargs	Any	Provider-specific configuration	{}

Returns:

Type	Description
BaseCheckpointer	Checkpointer instance

Raises:

Type	Description
ValueError	If provider is unknown

Source code in .sdk/src/tulip/memory/registry.py

def get_checkpointer(checkpointer_string: str, **kwargs: Any) -> BaseCheckpointer:
    """
    Get a checkpointer from a string identifier.

    Format: "provider" or "provider:config_hint"

    The config_hint is provider-specific and is passed as a keyword argument.

    Examples:
        - "memory" -> MemoryCheckpointer
        - "file:./checkpoints" -> FileCheckpointer(base_dir="./checkpoints")
        - "redis:localhost:6379" -> RedisCheckpointer(url="redis://localhost:6379")
        - "postgresql:mydb" -> PostgreSQLCheckpointer(database="mydb")
        - "mysql:mydb" -> MySQLCheckpointer(database="mydb")
        - "opensearch" -> OpenSearchCheckpointer()
        - "s3:my-bucket" -> S3Backend(bucket="my-bucket")

    Args:
        checkpointer_string: Checkpointer identifier
        **kwargs: Provider-specific configuration

    Returns:
        Checkpointer instance

    Raises:
        ValueError: If provider is unknown
    """
    if ":" in checkpointer_string:
        provider, config_hint = checkpointer_string.split(":", 1)
    else:
        provider = checkpointer_string
        config_hint = None

    if provider not in _CHECKPOINTERS:
        available = list(_CHECKPOINTERS.keys())
        raise ValueError(
            f"Unknown checkpointer provider: '{provider}'. "
            f"Available providers: {available}. "
            f"Install optional dependencies or register a custom provider."
        )

    # Pass config_hint if provided
    if config_hint:
        kwargs["config_hint"] = config_hint

    return _CHECKPOINTERS[provider](**kwargs)

register_checkpointer

register_checkpointer(name: str, factory: Callable[..., BaseCheckpointer]) -> None

Register a checkpointer provider.

Parameters:

Name	Type	Description	Default
name	str	Provider name (e.g., "redis", "postgresql", "s3")	required
factory	Callable[..., BaseCheckpointer]	Factory function that takes kwargs and returns a checkpointer	required

Example

def my_factory(kwargs) -> BaseCheckpointer: ... return MyCustomCheckpointer(kwargs) register_checkpointer("custom", my_factory)

Source code in .sdk/src/tulip/memory/registry.py

def register_checkpointer(
    name: str,
    factory: Callable[..., BaseCheckpointer],
) -> None:
    """
    Register a checkpointer provider.

    Args:
        name: Provider name (e.g., "redis", "postgresql", "s3")
        factory: Factory function that takes kwargs and returns a checkpointer

    Example:
        >>> def my_factory(**kwargs) -> BaseCheckpointer:
        ...     return MyCustomCheckpointer(**kwargs)
        >>> register_checkpointer("custom", my_factory)
    """
    _CHECKPOINTERS[name] = factory

list_checkpointers

list_checkpointers() -> list[str]

List available checkpointer providers.

Returns:

Type	Description
list[str]	List of registered provider names

Source code in .sdk/src/tulip/memory/registry.py

def list_checkpointers() -> list[str]:
    """
    List available checkpointer providers.

    Returns:
        List of registered provider names
    """
    return list(_CHECKPOINTERS.keys())