ant_ai.memory.protocol

Memory pydantic-model

Bases: BaseModel

Pluggable memory interface for agents.

Implement retrieve and update to connect any memory backend. Both methods take ctx explicitly — the same InvocationContext Tool/ToolStep inject into any tool method that declares a ctx parameter (see MemoryTool) — plus **kwargs for backend-specific parameters (e.g. mem0's user_id, filters, run_id) callers can pass when not going through ctx.

Show JSON schema:

{
  "description": "Pluggable memory interface for agents.\n\nImplement `retrieve` and `update` to connect any memory backend.\nBoth methods take `ctx` explicitly \u2014 the same `InvocationContext`\n`Tool`/`ToolStep` inject into any tool method that declares a `ctx`\nparameter (see `MemoryTool`) \u2014 plus `**kwargs` for backend-specific\nparameters (e.g. mem0's `user_id`, `filters`, `run_id`) callers can pass\nwhen not going through `ctx`.",
  "properties": {},
  "title": "Memory",
  "type": "object"
}

Source code in src/ant_ai/memory/protocol.py

class Memory(BaseModel):
    """
    Pluggable memory interface for agents.

    Implement `retrieve` and `update` to connect any memory backend.
    Both methods take `ctx` explicitly — the same `InvocationContext`
    `Tool`/`ToolStep` inject into any tool method that declares a `ctx`
    parameter (see `MemoryTool`) — plus `**kwargs` for backend-specific
    parameters (e.g. mem0's `user_id`, `filters`, `run_id`) callers can pass
    when not going through `ctx`.
    """

    async def retrieve(
        self,
        query: str,
        *,
        top_k: int = 5,
        ctx: InvocationContext | None = None,
        **kwargs: Any,
    ) -> list[Message]:
        """Return relevant memories as Messages, ready to inject into agent state."""
        raise NotImplementedError

    async def update(
        self,
        messages: list[Message],
        *,
        ctx: InvocationContext | None = None,
        **kwargs: Any,
    ) -> None:
        """Persist new knowledge from the current session."""
        raise NotImplementedError

retrieve async

retrieve(
    query: str,
    *,
    top_k: int = 5,
    ctx: InvocationContext | None = None,
    **kwargs: Any,
) -> list[Message]

Return relevant memories as Messages, ready to inject into agent state.

Source code in src/ant_ai/memory/protocol.py

async def retrieve(
    self,
    query: str,
    *,
    top_k: int = 5,
    ctx: InvocationContext | None = None,
    **kwargs: Any,
) -> list[Message]:
    """Return relevant memories as Messages, ready to inject into agent state."""
    raise NotImplementedError

update async

update(
    messages: list[Message],
    *,
    ctx: InvocationContext | None = None,
    **kwargs: Any,
) -> None

Persist new knowledge from the current session.

Source code in src/ant_ai/memory/protocol.py

async def update(
    self,
    messages: list[Message],
    *,
    ctx: InvocationContext | None = None,
    **kwargs: Any,
) -> None:
    """Persist new knowledge from the current session."""
    raise NotImplementedError