symfonic.core.tools.memory_tools

memory_tools

Memory tools — mediate between agent state and DocumentStore.

All tool functions follow the Phase 2 access pattern

store = state["deps"].get(StoreType)

No direct infrastructure imports allowed.

memory_read async

memory_read(state: dict[str, Any], doc_id: str) -> Document | None

Read a document by ID. Returns None if not found.

Source code in src/symfonic/core/tools/memory_tools.py

async def memory_read(state: dict[str, Any], doc_id: str) -> Document | None:
    """Read a document by ID. Returns None if not found."""
    store = _get_document_store(state)
    return await store.read(doc_id)

memory_search async

memory_search(state: dict[str, Any], query: str, limit: int = 10, offset: int = 0) -> list[Document]

Search documents.

Source code in src/symfonic/core/tools/memory_tools.py

async def memory_search(
    state: dict[str, Any],
    query: str,
    limit: int = 10,
    offset: int = 0,
) -> list[Document]:
    """Search documents."""
    store = _get_document_store(state)
    return await store.search(query, limit=limit, offset=offset)

memory_write async

memory_write(state: dict[str, Any], doc_id: str, content: str, metadata: dict | None = None) -> None

Write a document.

Source code in src/symfonic/core/tools/memory_tools.py

async def memory_write(
    state: dict[str, Any],
    doc_id: str,
    content: str,
    metadata: dict | None = None,
) -> None:
    """Write a document."""
    store = _get_document_store(state)
    doc = Document(id=doc_id, content=content, metadata=metadata or {})
    await store.write(doc)