Sessions & memory

Keep state across agent calls using sessions and scoped memory. This chapter covers revisions, expiry, vectors and change listeners. The SDK and a graph server are enough for the sessions and memory_lab examples; their vectors are supplied explicitly rather than generated by an embedding provider.

Sessions are created by the server

A session groups the runs of one conversation. Create it with await client.create_session(), backed by POST /api/v2/sessions, and attach runs with session_id= on submit or traverse. The CLI equivalents are --new-session and --session ID. This example comes from examples/python/sessions/call.py:

from __future__ import annotations

import asyncio
import os

from atfgraph import Client


async def main() -> None:
    async with Client(os.getenv("ATFGRAPH_SERVER")) as client:
        session = await client.create_session()
        for text in ("pack passports", "book airport transfer", "check visa rules"):
            result = await client.traverse("notebook.note", {"text": text}, session_id=session, timeout=60)
            print(result)
        stored = await client.memory_read("session", "notes", session_id=session)
        print("memory revision", stored.revision, "notes", stored.value)
        print(await client.traverse("notebook.forget", {}, session_id=session, timeout=60))


if __name__ == "__main__":
    asyncio.run(main())

Read the active session inside a handler through current_context().session_id. Child calls inherit it. Root turns in a session execute sequentially, while their children may run concurrently. test_root_session_turns_are_serialized_children_bypass checks that behavior.

The typed Client.create_session() method accepts no arguments. The HTTP endpoint also accepts a chosen ID and reuses its session on repeated creation, preserving memory (test_named_protocol_session_is_reused_without_losing_memory). To use that form, memory_lab/call.py calls client.transport.request("POST", "/api/v2/sessions", json_body={"id": name}).

If a root starts without a session, use await app.bind_session(session_id) before submitting children. For a block of code, with session_scope(session_id): selects the default session for memory and child submissions. Co-consult uses these helpers in _ensure_session and chat.

The four scopes

app.memory provides MemoryScope objects (python/atfgraph/memory.py). Authentication determines the namespace and the caller's principal determines the actor.

app.memory.default() selects session memory when a session is active, workflow memory otherwise inside a handler, and global memory outside execution. Convenience get, set, update and delete methods use the default unless you choose a scope.

This handler from examples/python/sessions/agent.py writes to all four scopes:

from __future__ import annotations

from atfgraph import Agent, current_context

app = Agent(node_id="notebook", version="1.0.0")


@app.traversal()
async def note(text: str) -> dict:
    """Append ``text`` to this session's notes; return everything remembered so far."""
    ctx = current_context()
    if not ctx.session_id:
        raise ValueError("submit this traversal with a session_id")
    session = app.memory.session()
    notes = await session.update("notes", lambda current: (current or []) + [text])
    turns = await app.memory.actor().update("turns", lambda c: (c or 0) + 1)
    total = await app.memory.global_scope().update("all_notes", lambda c: (c or 0) + 1)
    await app.memory.workflow().set("last_note", text)
    return {"session_id": ctx.session_id, "notes": notes, "actor_turns": turns, "global_notes": total}


@app.traversal()
async def forget() -> dict:
    deleted = await app.memory.session().delete("notes")
    return {"deleted": deleted}

The example tests check session isolation by writing three notes to one session and one to another. They also check revision 3 for the first session's notes and failure when a session-required call has no session_id.

Read, write, delete with revisions

Writes return the new revision. Reads return a MemoryValue containing value, revision and found. A missing value has revision 0.

To avoid overwriting a concurrent update, send the revision you read as expected_revision. Use 0 to create only when absent. A stale revision returns HTTP 409 and raises ConflictError with revision_conflict. scope.update performs the read, modification and conditional write for you, retrying conflicts up to attempts. The restart-and-revision integration test checks conflict rejection and persistence after a daemon restart.

Set ttl_seconds to expire a key. Once expired, it reads as absent and is excluded from listings. Expiry does not emit a change-feed event; test_change_feed_cursor_retention_and_ttl_expiry_semantics checks this.

Use scope.keys(prefix=None, cursor=None, limit=None, kind="value") for one ascending page of KeyPage.keys and a next_cursor. scope.list_keys(prefix=None, kind="value") follows all pages, raising ValueError if it would exceed max_keys, default 100000. Set kind="vector" to list vector keys.

Vector memory

Store embeddings in the same scopes as ordinary values, with separate revisions per vector key. Methods are on MemoryScope and forwarded by Memory; there is no app.memory.vector object. You supply the embedding because this module includes no embedding provider. The memory-lab example uses fixture vectors:

from __future__ import annotations

from dataclasses import asdict

from atfgraph import Agent

app = Agent(node_id="memory-lab", version="1.0.0")


@app.traversal()
async def vector_lookup(session_id: str) -> dict:
    scoped = app.memory.session(session_id)
    await scoped.set_vector("blue", [1.0, 0.0, 0.0], {"colour": "blue"})
    await scoped.set_vector("red", [0.0, 1.0, 0.0], {"colour": "red"})
    matches = await scoped.similarity_search([1.0, 0.0, 0.0], top_k=1)
    return {"matches": [asdict(match) for match in matches]}

The SDK validates these bounds before sending a request; the server also enforces them (python/atfgraph/retrieval.py):

The real-daemon vector test checks ranked search, metadata filtering, listing, rejection of a zero vector as invalid_vector, and stale-revision conflicts. Use vectors when you already have embeddings. For server-managed text chunking and BM25 search, use Document knowledge.

Change feeds

Writes and deletes of values and vectors append events to a persistent namespace change feed with gap-free sequence numbers. Read it in one of these ways:

• scope.changes(after=0, limit=None, prefix=None) returns a page with items, next_cursor and oldest_sequence.
• scope.watch("user:*", after=0, interval=0.25, stop=None) polls and yields MemoryChangeEvent objects matching shell-style patterns.
• @app.on_change(patterns, scope=None, after=0, interval=0.25) registers a listener for the worker's lifetime. @app.memory.on_change is the equivalent form.

The memory-lab example registers this listener:

from __future__ import annotations

from collections import deque

from atfgraph import Agent

app = Agent(node_id="memory-lab", version="1.0.0")
events = deque(maxlen=100)


@app.memory.on_change("memory-lab.*", scope="global")
async def changed(event):
    events.append({"key": event.key, "action": event.action})


@app.traversal()
async def emit_change(value: str) -> dict:
    await app.memory.global_scope().set("memory-lab.signal", value)
    return {"accepted": True, "note": "Poll recent_changes for asynchronous delivery."}


@app.skill()
def recent_changes() -> dict:
    return {"events": list(events)}

Each event identifies its sequence, scope, scope ID, key, kind (value or vector), action (set or delete), revision, principal and creation time. Value-write events include value only when its canonical JSON is at most 65536 bytes. Otherwise, value_omitted is true and you must reread the key. Vector events never include a value.

Events are retained for 30 days; compare saved cursors with oldest_sequence. Listeners can be synchronous or asynchronous. An exception stops the listener group and propagates out of serve, as checked by test_change_feed_watch_and_subscriptions. The decorator returns the function unchanged so unit tests can still call it directly.

Memory from outside a worker

Scripts can call Client.memory_read, memory_write and memory_delete with scope, key and the appropriate session_id, actor_id or root_id. For the same higher-level interface used by agents, wrap a client with Memory(lambda: client) or MemoryScope(lambda: client, "session", session_id="..."). The retrieval tests use these wrappers against a real daemon.

What a worker credential may reach

During a handler, memory requests include its task_id and lease_token through Client._execution_authority. The server uses that authority to resolve the inherited session and root. Outside a handler, no lease accompanies the request, so the caller's standalone credential governs access. A change listener therefore needs its own memory permission; a worker-only credential does not grant access to unrelated runs.

Not verified: the exact errors for worker-only reads of unrelated session or workflow memory, and for on_change outside a traversal, by the examples cited here. They use loopback development credentials with admin, invoke and worker roles. Check those boundaries with configured keys and policies using Identity and access.

Next

• Retrieve text using Document knowledge.
• Save per-run progress with Checkpoints and approvals.
• Understand scope permissions through Identity and access.