Base

Base memory-store interface and common data structures.

A memory store holds what an agent has learned and wants to carry into later conversations: short text notes addressed by path. It is deliberately not a FileStore. That interface carries list_files_bytes, extracted-text caching, content types and BytesIO — all of which exist for agent file plumbing memory does not take part in, and each of which caused a bug when memory was built on top of it.

MemoryEntry

Bases: BaseModel

One memory, without its content.

Source code in dynamiq/storages/memory/base.py

class MemoryEntry(BaseModel):
    """One memory, without its content."""

    path: str
    size: int = 0
    updated_at: datetime | None = None

MemoryNotFoundError

Bases: MemoryStoreError

Raised when a memory does not exist.

Source code in dynamiq/storages/memory/base.py

class MemoryNotFoundError(MemoryStoreError):
    """Raised when a memory does not exist."""

MemoryPermissionError

Bases: MemoryStoreError

Raised when a memory operation is not permitted.

Source code in dynamiq/storages/memory/base.py

class MemoryPermissionError(MemoryStoreError):
    """Raised when a memory operation is not permitted."""

MemoryStore

Bases: ABC, BaseModel

Abstract base class for memory backends.

Four operations over text addressed by path. There is no exists: read raises when a path is absent and delete reports whether it removed anything, which covers every caller and keeps one endpoint out of the API.

_clone_shared tells Node.clone() to share this instance by reference rather than deep-copying it, so parallel cloned tools reach the same store.

Source code in dynamiq/storages/memory/base.py

class MemoryStore(abc.ABC, BaseModel):
    """Abstract base class for memory backends.

    Four operations over text addressed by path. There is no ``exists``: ``read`` raises when a path
    is absent and ``delete`` reports whether it removed anything, which covers every caller and
    keeps one endpoint out of the API.

    ``_clone_shared`` tells ``Node.clone()`` to share this instance by reference rather than
    deep-copying it, so parallel cloned tools reach the same store.
    """

    description: str | None = Field(
        default=None,
        description="What this memory holds. Shown to the model so it can tell memories apart.",
    )

    _clone_shared: ClassVar[bool] = True

    model_config = ConfigDict(arbitrary_types_allowed=True)

    @computed_field
    @property
    def type(self) -> str:
        """Dotted path used to rebuild this store from a serialized workflow."""
        return f"{self.__module__.rsplit('.', 1)[0]}.{self.__class__.__name__}"

    @property
    def to_dict_exclude_params(self) -> dict[str, bool]:
        """Parameters to drop when serializing."""
        return {}

    def to_dict(self, **kwargs) -> dict[str, Any]:
        """Convert the store to a dictionary."""
        kwargs.pop("for_tracing", None)
        kwargs.pop("include_secure_params", None)
        exclude = kwargs.pop("exclude", self.to_dict_exclude_params)
        data = self.model_dump(exclude=exclude, **kwargs)
        data["type"] = self.type
        return data

    def describe_namespaces(self) -> dict[str, str]:
        """Path prefix -> what lives there.

        The empty key means everything this store holds. A composite overrides this to compose its
        routes' descriptions, so the caller never assembles a parallel structure by hand.
        """
        return {"": self.description} if self.description else {}

    def identity(self) -> Hashable:
        """What underlying store this addresses; two stores with the same identity hold the same data.

        A composite strips its route prefix before calling a store, so two routes pointing at one
        store would map different agent paths onto the same key and silently overwrite each other.
        This is how the composite tells them apart and refuses. Subclasses that address something
        remote should key on what identifies it there, not on the Python object.
        """
        return id(self)

    @abc.abstractmethod
    def list(self, prefix: str = "", user_id: str | None = None) -> list[MemoryEntry]:
        """List memories under ``prefix``. An empty prefix lists everything.

        Returns metadata only — read a path to get its content. ``user_id`` is the end user this
        call is for; it overrides whatever default the store was configured with.
        """

    @abc.abstractmethod
    def read(self, path: str, user_id: str | None = None) -> str:
        """Return the content of one memory.

        Raises:
            MemoryNotFoundError: If the path does not exist.
        """

    @abc.abstractmethod
    def write(self, path: str, content: str, user_id: str | None = None) -> MemoryEntry:
        """Create or replace one memory, returning its entry."""

    @abc.abstractmethod
    def delete(self, path: str, user_id: str | None = None) -> bool:
        """Delete one memory. Returns False when it was not there."""

to_dict_exclude_params: dict[str, bool] property

Parameters to drop when serializing.

type: str property

Dotted path used to rebuild this store from a serialized workflow.

delete(path, user_id=None) abstractmethod

Delete one memory. Returns False when it was not there.

Source code in dynamiq/storages/memory/base.py

@abc.abstractmethod
def delete(self, path: str, user_id: str | None = None) -> bool:
    """Delete one memory. Returns False when it was not there."""

describe_namespaces()

Path prefix -> what lives there.

The empty key means everything this store holds. A composite overrides this to compose its routes' descriptions, so the caller never assembles a parallel structure by hand.

Source code in dynamiq/storages/memory/base.py

def describe_namespaces(self) -> dict[str, str]:
    """Path prefix -> what lives there.

    The empty key means everything this store holds. A composite overrides this to compose its
    routes' descriptions, so the caller never assembles a parallel structure by hand.
    """
    return {"": self.description} if self.description else {}

identity()

What underlying store this addresses; two stores with the same identity hold the same data.

A composite strips its route prefix before calling a store, so two routes pointing at one store would map different agent paths onto the same key and silently overwrite each other. This is how the composite tells them apart and refuses. Subclasses that address something remote should key on what identifies it there, not on the Python object.

Source code in dynamiq/storages/memory/base.py

def identity(self) -> Hashable:
    """What underlying store this addresses; two stores with the same identity hold the same data.

    A composite strips its route prefix before calling a store, so two routes pointing at one
    store would map different agent paths onto the same key and silently overwrite each other.
    This is how the composite tells them apart and refuses. Subclasses that address something
    remote should key on what identifies it there, not on the Python object.
    """
    return id(self)

list(prefix='', user_id=None) abstractmethod

List memories under prefix. An empty prefix lists everything.

Returns metadata only — read a path to get its content. user_id is the end user this call is for; it overrides whatever default the store was configured with.

Source code in dynamiq/storages/memory/base.py

@abc.abstractmethod
def list(self, prefix: str = "", user_id: str | None = None) -> list[MemoryEntry]:
    """List memories under ``prefix``. An empty prefix lists everything.

    Returns metadata only — read a path to get its content. ``user_id`` is the end user this
    call is for; it overrides whatever default the store was configured with.
    """

read(path, user_id=None) abstractmethod

Return the content of one memory.

Raises:

Type	Description
MemoryNotFoundError	If the path does not exist.

Source code in dynamiq/storages/memory/base.py

@abc.abstractmethod
def read(self, path: str, user_id: str | None = None) -> str:
    """Return the content of one memory.

    Raises:
        MemoryNotFoundError: If the path does not exist.
    """

to_dict(**kwargs)

Convert the store to a dictionary.

Source code in dynamiq/storages/memory/base.py

def to_dict(self, **kwargs) -> dict[str, Any]:
    """Convert the store to a dictionary."""
    kwargs.pop("for_tracing", None)
    kwargs.pop("include_secure_params", None)
    exclude = kwargs.pop("exclude", self.to_dict_exclude_params)
    data = self.model_dump(exclude=exclude, **kwargs)
    data["type"] = self.type
    return data

write(path, content, user_id=None) abstractmethod

Create or replace one memory, returning its entry.

Source code in dynamiq/storages/memory/base.py

@abc.abstractmethod
def write(self, path: str, content: str, user_id: str | None = None) -> MemoryEntry:
    """Create or replace one memory, returning its entry."""

MemoryStoreConfig

Bases: BaseModel

Configuration for an agent's memory store.

Attributes:

Name	Type	Description
enabled	bool	Whether the agent gets a memory tool at all.
backend	MemoryStore	The store holding memories. A CompositeMemoryStore when there are several.
write_enabled	bool	Whether the agent may create, edit and delete memories.

Source code in dynamiq/storages/memory/base.py

class MemoryStoreConfig(BaseModel):
    """Configuration for an agent's memory store.

    Attributes:
        enabled: Whether the agent gets a memory tool at all.
        backend: The store holding memories. A ``CompositeMemoryStore`` when there are several.
        write_enabled: Whether the agent may create, edit and delete memories.
    """

    enabled: bool = False
    backend: MemoryStore = Field(..., description="Store holding the agent's memories.")
    write_enabled: bool = Field(
        default=True,
        description="Whether the agent is permitted to write memories.",
    )

    model_config = ConfigDict(arbitrary_types_allowed=True)

    @property
    def to_dict_exclude_params(self) -> dict[str, bool]:
        """The backend serializes itself, so exclude it from the plain dump."""
        return {"backend": True}

    def to_dict(self, **kwargs) -> dict[str, Any]:
        """Convert the config to a dictionary, delegating the backend to its own ``to_dict``."""
        for_tracing = kwargs.pop("for_tracing", False)
        if for_tracing and not self.enabled:
            return {"enabled": False}
        include_secure_params = kwargs.pop("include_secure_params", False)
        exclude = kwargs.pop("exclude", self.to_dict_exclude_params)
        data = self.model_dump(exclude=exclude, **kwargs)
        data["backend"] = self.backend.to_dict(for_tracing=for_tracing, include_secure_params=include_secure_params)
        return data

to_dict_exclude_params: dict[str, bool] property

The backend serializes itself, so exclude it from the plain dump.

to_dict(**kwargs)

Convert the config to a dictionary, delegating the backend to its own to_dict.

Source code in dynamiq/storages/memory/base.py

def to_dict(self, **kwargs) -> dict[str, Any]:
    """Convert the config to a dictionary, delegating the backend to its own ``to_dict``."""
    for_tracing = kwargs.pop("for_tracing", False)
    if for_tracing and not self.enabled:
        return {"enabled": False}
    include_secure_params = kwargs.pop("include_secure_params", False)
    exclude = kwargs.pop("exclude", self.to_dict_exclude_params)
    data = self.model_dump(exclude=exclude, **kwargs)
    data["backend"] = self.backend.to_dict(for_tracing=for_tracing, include_secure_params=include_secure_params)
    return data

MemoryStoreError

Bases: Exception

Base exception for memory-store operations.

Source code in dynamiq/storages/memory/base.py

class MemoryStoreError(Exception):
    """Base exception for memory-store operations."""

    def __init__(self, message: str, operation: str | None = None, path: str | None = None):
        self.message = message
        self.operation = operation
        self.path = path
        super().__init__(self.message)

render_namespaces(namespaces)

One line per memory, for the tool description and the prompt.

Empty when nothing is described, in which case the surrounding wording already covers a single unnamed memory.

Source code in dynamiq/storages/memory/base.py

def render_namespaces(namespaces: dict[str, str]) -> str:
    """One line per memory, for the tool description and the prompt.

    Empty when nothing is described, in which case the surrounding wording already covers a single
    unnamed memory.
    """
    if not namespaces:
        return ""
    return "\n".join(f"- {prefix} - {text}" if prefix else f"- {text}" for prefix, text in namespaces.items() if text)