Skip to content

Store

Cross-thread durable memory. See the Cross-thread memory guide.

agentflow.store

Store: durable, cross-thread application memory (distinct from checkpoints).

A checkpointer persists one thread's execution state for resume; a store holds durable key-value data shared across threads (profiles, facts, long-term memory). See :mod:agentflow.store.base for the protocol.

Store

Bases: Protocol

Durable, namespaced key-value memory shared across threads.

Source code in agentflow/store/base.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@runtime_checkable
class Store(Protocol):
    """Durable, namespaced key-value memory shared across threads."""

    async def get(self, namespace: Namespace, key: str) -> Item | None:
        """Return the item at ``(namespace, key)``, or None if absent/expired."""
        ...

    async def put(
        self,
        namespace: Namespace,
        key: str,
        value: Any,
        *,
        ttl: float | None = None,
        if_absent: bool = False,
    ) -> Item:
        """Upsert ``value`` at ``(namespace, key)`` and return the stored item.

        ``ttl`` is a lifetime in seconds from now; ``None`` means no expiry. An
        upsert preserves the original ``created_at`` and refreshes
        ``updated_at``.

        When ``if_absent`` is True the write is conditional: it succeeds only if
        no live item exists at ``(namespace, key)``, otherwise it raises
        :class:`~agentflow.errors.StoreConflict`. An expired item counts as
        absent, so its key can be reclaimed. This is an atomic create — two
        writers racing the same key give exactly one winner and one conflict —
        which is how a single-flight claim is built (see
        :class:`~agentflow.prebuilt.IdempotentOp`). ``if_absent`` ignores no
        other semantics: a successful conditional write behaves like a first
        insert (fresh ``created_at``).
        """
        ...

    async def delete(self, namespace: Namespace, key: str) -> bool:
        """Delete the item; return True if something was removed."""
        ...

    async def search(
        self,
        namespace_prefix: Namespace,
        *,
        limit: int = 100,
        offset: int = 0,
    ) -> list[Item]:
        """Return items whose namespace starts with ``namespace_prefix``,
        ordered by namespace then key, skipping expired items. ``limit`` and
        ``offset`` paginate."""
        ...

    async def list_namespaces(
        self, *, prefix: Namespace = (), limit: int = 100, offset: int = 0
    ) -> list[Namespace]:
        """Return the distinct namespaces that currently hold items, filtered
        to those starting with ``prefix`` and ordered lexicographically."""
        ...

get async 

get(namespace: Namespace, key: str) -> Item | None

Return the item at (namespace, key), or None if absent/expired.

Source code in agentflow/store/base.py 
62
63
64
async def get(self, namespace: Namespace, key: str) -> Item | None:
    """Return the item at ``(namespace, key)``, or None if absent/expired."""
    ...

put async 

put(namespace: Namespace, key: str, value: Any, *, ttl: float | None = None, if_absent: bool = False) -> Item

Upsert value at (namespace, key) and return the stored item.

ttl is a lifetime in seconds from now; None means no expiry. An upsert preserves the original created_at and refreshes updated_at.

When if_absent is True the write is conditional: it succeeds only if no live item exists at (namespace, key), otherwise it raises :class:~agentflow.errors.StoreConflict. An expired item counts as absent, so its key can be reclaimed. This is an atomic create — two writers racing the same key give exactly one winner and one conflict — which is how a single-flight claim is built (see :class:~agentflow.prebuilt.IdempotentOp). if_absent ignores no other semantics: a successful conditional write behaves like a first insert (fresh created_at).

Source code in agentflow/store/base.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def put(
    self,
    namespace: Namespace,
    key: str,
    value: Any,
    *,
    ttl: float | None = None,
    if_absent: bool = False,
) -> Item:
    """Upsert ``value`` at ``(namespace, key)`` and return the stored item.

    ``ttl`` is a lifetime in seconds from now; ``None`` means no expiry. An
    upsert preserves the original ``created_at`` and refreshes
    ``updated_at``.

    When ``if_absent`` is True the write is conditional: it succeeds only if
    no live item exists at ``(namespace, key)``, otherwise it raises
    :class:`~agentflow.errors.StoreConflict`. An expired item counts as
    absent, so its key can be reclaimed. This is an atomic create — two
    writers racing the same key give exactly one winner and one conflict —
    which is how a single-flight claim is built (see
    :class:`~agentflow.prebuilt.IdempotentOp`). ``if_absent`` ignores no
    other semantics: a successful conditional write behaves like a first
    insert (fresh ``created_at``).
    """
    ...

delete async 

delete(namespace: Namespace, key: str) -> bool

Delete the item; return True if something was removed.

Source code in agentflow/store/base.py 
93
94
95
async def delete(self, namespace: Namespace, key: str) -> bool:
    """Delete the item; return True if something was removed."""
    ...

search async 

search(namespace_prefix: Namespace, *, limit: int = 100, offset: int = 0) -> list[Item]

Return items whose namespace starts with namespace_prefix, ordered by namespace then key, skipping expired items. limit and offset paginate.

Source code in agentflow/store/base.py 
 97
 98
 99
100
101
102
103
104
105
106
107
async def search(
    self,
    namespace_prefix: Namespace,
    *,
    limit: int = 100,
    offset: int = 0,
) -> list[Item]:
    """Return items whose namespace starts with ``namespace_prefix``,
    ordered by namespace then key, skipping expired items. ``limit`` and
    ``offset`` paginate."""
    ...

list_namespaces async 

list_namespaces(*, prefix: Namespace = (), limit: int = 100, offset: int = 0) -> list[Namespace]

Return the distinct namespaces that currently hold items, filtered to those starting with prefix and ordered lexicographically.

Source code in agentflow/store/base.py 
109
110
111
112
113
114
async def list_namespaces(
    self, *, prefix: Namespace = (), limit: int = 100, offset: int = 0
) -> list[Namespace]:
    """Return the distinct namespaces that currently hold items, filtered
    to those starting with ``prefix`` and ordered lexicographically."""
    ...

Item dataclass

One stored value plus its addressing and timestamps.

created_at / updated_at are ISO-8601 strings (UTC). expires_at, when set, is the ISO time after which the item is considered gone; a store must not return an expired item from get or search.

Source code in agentflow/store/base.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@dataclass(frozen=True, slots=True)
class Item:
    """One stored value plus its addressing and timestamps.

    ``created_at`` / ``updated_at`` are ISO-8601 strings (UTC). ``expires_at``,
    when set, is the ISO time after which the item is considered gone; a store
    must not return an expired item from ``get`` or ``search``.
    """

    namespace: Namespace
    key: str
    value: Any
    created_at: str = ""
    updated_at: str = ""
    expires_at: str | None = None
    extra: dict[str, Any] = field(default_factory=dict)

MemoryStore

Keeps items in a nested dict keyed by namespace then key. Not durable.

Expiry is lazy: an expired item is skipped on read and dropped when next touched, so no background sweeper is needed.

Source code in agentflow/store/memory.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class MemoryStore:
    """Keeps items in a nested dict keyed by namespace then key. Not durable.

    Expiry is lazy: an expired item is skipped on read and dropped when next
    touched, so no background sweeper is needed.
    """

    def __init__(self) -> None:
        self._items: dict[Namespace, dict[str, Item]] = {}

    async def get(self, namespace: Namespace, key: str) -> Item | None:
        bucket = self._items.get(namespace)
        if not bucket:
            return None
        item = bucket.get(key)
        if item is None:
            return None
        if is_expired(item.expires_at):
            del bucket[key]
            return None
        return item

    async def put(
        self,
        namespace: Namespace,
        key: str,
        value: Any,
        *,
        ttl: float | None = None,
        if_absent: bool = False,
    ) -> Item:
        validate_namespace(namespace)
        bucket = self._items.setdefault(namespace, {})
        existing = bucket.get(key)
        live = existing is not None and not is_expired(existing.expires_at)
        # Atomic by construction: this method never awaits between the check and
        # the write, and the event loop is single-threaded, so no other task can
        # interleave. A live item under if_absent is a lost claim.
        if if_absent and live:
            raise StoreConflict(namespace, key)
        created = existing.created_at if (live and existing is not None) else ""
        ts = now_iso()
        item = Item(
            namespace=namespace,
            key=key,
            value=value,
            created_at=created or ts,
            updated_at=ts,
            expires_at=expiry_iso(ttl),
        )
        bucket[key] = item
        return item

    async def delete(self, namespace: Namespace, key: str) -> bool:
        bucket = self._items.get(namespace)
        if not bucket or key not in bucket:
            return False
        del bucket[key]
        if not bucket:
            del self._items[namespace]
        return True

    async def search(
        self,
        namespace_prefix: Namespace,
        *,
        limit: int = 100,
        offset: int = 0,
    ) -> list[Item]:
        matches: list[Item] = []
        for ns in sorted(self._items):
            if ns[: len(namespace_prefix)] != namespace_prefix:
                continue
            for key in sorted(self._items[ns]):
                item = self._items[ns][key]
                if is_expired(item.expires_at):
                    continue
                matches.append(item)
        return matches[offset : offset + limit]

    async def list_namespaces(
        self, *, prefix: Namespace = (), limit: int = 100, offset: int = 0
    ) -> list[Namespace]:
        seen: set[Namespace] = set()
        for ns, bucket in self._items.items():
            if ns[: len(prefix)] != prefix:
                continue
            if any(not is_expired(i.expires_at) for i in bucket.values()):
                seen.add(ns)
        return sorted(seen)[offset : offset + limit]

PostgresStore

Durable namespaced key-value store backed by a PostgreSQL table.

Pass either a dsn or an existing asyncpg pool via pool. table names the store table (created on first use).

Source code in agentflow/store/postgres.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class PostgresStore:
    """Durable namespaced key-value store backed by a PostgreSQL table.

    Pass either a ``dsn`` or an existing ``asyncpg`` pool via ``pool``.
    ``table`` names the store table (created on first use).
    """

    def __init__(
        self,
        dsn: str = "postgresql://postgres@localhost:5432/postgres",
        *,
        pool: asyncpg.Pool | None = None,
        table: str = "store",
    ) -> None:
        if not table.isidentifier():
            raise ValueError(f"unsafe table name {table!r}")
        self._dsn = dsn
        self._pool = pool
        self._owns_pool = pool is None
        self.table = table
        self._init_done = False
        self._schema_lock = asyncio.Lock()
        self._pool_lock = asyncio.Lock()

    # ------------------------------------------------------------------

    async def _get_pool(self) -> asyncpg.Pool:
        if self._pool is None:
            async with self._pool_lock:
                if self._pool is None:
                    self._pool = await asyncpg.create_pool(self._dsn)
        await self._ensure_schema()
        return self._pool

    async def _ensure_schema(self) -> None:
        if self._init_done:
            return
        assert self._pool is not None
        async with self._schema_lock:
            if self._init_done:  # another coroutine created it while we waited
                return

            async def create(conn: asyncpg.Connection) -> None:
                await conn.execute(
                    f"""
                    CREATE TABLE IF NOT EXISTS {self.table} (
                        namespace  TEXT[]      NOT NULL,
                        key        TEXT        NOT NULL,
                        value      JSONB       NOT NULL,
                        created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                        updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                        expires_at TIMESTAMPTZ,
                        PRIMARY KEY (namespace, key)
                    )
                    """
                )
                await conn.execute(
                    f"CREATE INDEX IF NOT EXISTS idx_{self.table}_ns ON {self.table} (namespace)"
                )

            await ensure_schema(self._pool, self.table, create)
            self._init_done = True

    async def close(self) -> None:
        """Close the connection pool (only if this store created it)."""
        if self._pool is not None and self._owns_pool:
            await self._pool.close()
            self._pool = None

    # ------------------------------------------------------------------

    @staticmethod
    def _iso(dt: datetime | None) -> str | None:
        return dt.isoformat() if dt is not None else None

    def _row_to_item(self, row: asyncpg.Record) -> Item:
        value = row["value"]
        if isinstance(value, str):
            value = json.loads(value)
        return Item(
            namespace=tuple(row["namespace"]),
            key=row["key"],
            value=value,
            created_at=self._iso(row["created_at"]) or "",
            updated_at=self._iso(row["updated_at"]) or "",
            expires_at=self._iso(row["expires_at"]),
        )

    async def get(self, namespace: Namespace, key: str) -> Item | None:
        pool = await self._get_pool()
        row = await pool.fetchrow(
            f"SELECT namespace, key, value, created_at, updated_at, expires_at "
            f"FROM {self.table} "
            "WHERE namespace=$1 AND key=$2 AND (expires_at IS NULL OR expires_at > NOW())",
            list(namespace),
            key,
        )
        return self._row_to_item(row) if row is not None else None

    async def put(
        self,
        namespace: Namespace,
        key: str,
        value: Any,
        *,
        ttl: float | None = None,
        if_absent: bool = False,
    ) -> Item:
        validate_namespace(namespace)
        pool = await self._get_pool()
        blob = json.dumps(value, ensure_ascii=False)
        expires = f"NOW() + make_interval(secs => {float(ttl)})" if ttl is not None else "NULL"

        if if_absent:
            # Atomic claim: insert if the key is free, or overwrite only if the
            # existing row is expired (an expired row counts as absent, so its
            # created_at is reset). A LIVE row makes the DO UPDATE ... WHERE
            # no-op, so no row is RETURNED — that is the lost claim.
            row = await pool.fetchrow(
                f"""
                INSERT INTO {self.table} (namespace, key, value, expires_at)
                VALUES ($1, $2, $3::jsonb, {expires})
                ON CONFLICT (namespace, key) DO UPDATE SET
                    value = EXCLUDED.value,
                    created_at = NOW(),
                    updated_at = NOW(),
                    expires_at = EXCLUDED.expires_at
                WHERE {self.table}.expires_at IS NOT NULL
                  AND {self.table}.expires_at <= NOW()
                RETURNING namespace, key, value, created_at, updated_at, expires_at
                """,
                list(namespace),
                key,
                blob,
            )
            if row is None:
                raise StoreConflict(namespace, key)
            return self._row_to_item(row)

        row = await pool.fetchrow(
            f"""
            INSERT INTO {self.table} (namespace, key, value, expires_at)
            VALUES ($1, $2, $3::jsonb, {expires})
            ON CONFLICT (namespace, key) DO UPDATE SET
                value = EXCLUDED.value,
                updated_at = NOW(),
                expires_at = EXCLUDED.expires_at
            RETURNING namespace, key, value, created_at, updated_at, expires_at
            """,
            list(namespace),
            key,
            blob,
        )
        return self._row_to_item(row)

    async def delete(self, namespace: Namespace, key: str) -> bool:
        pool = await self._get_pool()
        result = await pool.execute(
            f"DELETE FROM {self.table} WHERE namespace=$1 AND key=$2",
            list(namespace),
            key,
        )
        return result.split()[-1] != "0"

    async def search(
        self,
        namespace_prefix: Namespace,
        *,
        limit: int = 100,
        offset: int = 0,
    ) -> list[Item]:
        pool = await self._get_pool()
        prefix = list(namespace_prefix)
        n = len(prefix)
        rows = await pool.fetch(
            f"SELECT namespace, key, value, created_at, updated_at, expires_at "
            f"FROM {self.table} "
            "WHERE namespace[1:$1] = $2::text[] "
            "AND (expires_at IS NULL OR expires_at > NOW()) "
            "ORDER BY namespace, key LIMIT $3 OFFSET $4",
            n,
            prefix,
            limit,
            offset,
        )
        return [self._row_to_item(r) for r in rows]

    async def list_namespaces(
        self, *, prefix: Namespace = (), limit: int = 100, offset: int = 0
    ) -> list[Namespace]:
        pool = await self._get_pool()
        pre = list(prefix)
        n = len(pre)
        rows = await pool.fetch(
            f"SELECT DISTINCT namespace FROM {self.table} "
            "WHERE namespace[1:$1] = $2::text[] "
            "AND (expires_at IS NULL OR expires_at > NOW()) "
            "ORDER BY namespace LIMIT $3 OFFSET $4",
            n,
            pre,
            limit,
            offset,
        )
        return [tuple(r["namespace"]) for r in rows]

close async 

close() -> None

Close the connection pool (only if this store created it).

Source code in agentflow/store/postgres.py 
112
113
114
115
116
async def close(self) -> None:
    """Close the connection pool (only if this store created it)."""
    if self._pool is not None and self._owns_pool:
        await self._pool.close()
        self._pool = None