Storage

Documentation Index

Fetch the complete documentation index at: https://docs.getbindu.com/llms.txt

Use this file to discover all available pages before exploring further.

​

How Bindu Storage Works

​

What gets stored

​

Backends at a glance

​

Configuration

​

Memory (default)

# Default — no env vars needed. Just run your agent.
STORAGE_TYPE=memory

​

PostgreSQL

STORAGE_TYPE=postgres
DATABASE_URL=postgresql+asyncpg://bindu_user:secret@localhost:5432/bindu_db

​

The Storage Interface

# Tasks
async def submit_task(context_id, message, caller_did=None) -> Task
async def load_task(task_id, history_length=None) -> Task | None
async def update_task(task_id, state, new_artifacts=None,
                      new_messages=None, metadata=None) -> Task
async def update_task_state_if(task_id, from_state, to_state) -> bool  # CAS
async def list_tasks(length=None, offset=0, owner_did=None) -> list[Task]
async def list_tasks_by_context(context_id, length=None, offset=0,
                                owner_did=None) -> list[Task]
async def count_tasks(status=None) -> int

# Contexts
async def load_context(context_id) -> ContextT | None
async def append_to_contexts(context_id, messages) -> None
async def update_context(context_id, context) -> None
async def list_contexts(length=None, offset=0, owner_did=None) -> list[ContextT]

# Webhooks (persisted per task for long-running callbacks)
async def save_webhook_config(task_id, config) -> None
async def load_webhook_config(task_id) -> PushNotificationConfig | None
async def delete_webhook_config(task_id) -> None
async def load_all_webhook_configs() -> dict[UUID, PushNotificationConfig]

# Ownership / lifecycle
async def get_task_owner(task_id) -> str | None
async def get_context_owner(context_id) -> str | None
async def clear_context(context_id) -> None
async def clear_all() -> None
async def close() -> None

• update_task_state_if is a compare-and-swap. In Postgres it’s a single UPDATE ... WHERE state = :from; in memory it’s a comparison with no await between read and write. This is how cancel-vs-complete races stay correct.
• OwnershipError is raised by submit_task when a context already exists with a different owner_did. Handlers catch it and convert to ContextNotFoundError so they don’t leak existence to the caller.
• Terminal states are immutable. submit_task on a task already in completed, failed, canceled, or rejected raises ValueError — you must create a new task with reference_task_ids.

​

Using the factory

from bindu.server.storage.factory import create_storage, close_storage

storage = await create_storage()        # reads STORAGE_TYPE from env
# or, with DID-based schema isolation:
storage = await create_storage(did="did:bindu:alice:agent1:abc123")

task = await storage.load_task(task_id)
# ... use storage ...
await close_storage(storage)

​

The Storage Lifecycle

​

The Postgres Schema

id                UUID         PRIMARY KEY
context_id        UUID         FK contexts(id) ON DELETE CASCADE
owner_did         VARCHAR(255) NULL          -- authenticated caller
kind              VARCHAR(50)  NOT NULL DEFAULT 'task'
state             VARCHAR(50)  NOT NULL      -- submitted|working|...
state_timestamp   TIMESTAMPTZ  NOT NULL
history           JSONB        NOT NULL DEFAULT '[]'
artifacts         JSONB        NULL DEFAULT '[]'
metadata          JSONB        NULL DEFAULT '{}'
created_at        TIMESTAMPTZ  NOT NULL DEFAULT now()
updated_at        TIMESTAMPTZ  NOT NULL DEFAULT now() ON UPDATE now()

id              UUID         PRIMARY KEY
owner_did       VARCHAR(255) NULL          -- write-once on first submit
context_data    JSONB        NOT NULL DEFAULT '{}'
message_history JSONB        NULL DEFAULT '[]'
created_at      TIMESTAMPTZ  NOT NULL DEFAULT now()
updated_at      TIMESTAMPTZ  NOT NULL DEFAULT now() ON UPDATE now()

id            SERIAL       PRIMARY KEY
task_id       UUID         FK tasks(id) ON DELETE CASCADE
feedback_data JSONB        NOT NULL
created_at    TIMESTAMPTZ  NOT NULL DEFAULT now()

task_id    UUID         PRIMARY KEY, FK tasks(id) ON DELETE CASCADE
config     JSONB        NOT NULL
created_at TIMESTAMPTZ  NOT NULL DEFAULT now()
updated_at TIMESTAMPTZ  NOT NULL DEFAULT now() ON UPDATE now()

{
  "id": "9b7f5e10-7b8d-4d4d-8c3d-6b6a4b8c1a99",
  "context_id": "3f0e2c1a-1f4b-4f5b-8a2e-cdb8a9c3e7d1",
  "owner_did": "did:bindu:alice:research-agent:abc123",
  "kind": "task",
  "state": "completed",
  "state_timestamp": "2026-05-18T12:00:00+00:00",
  "history": [
    {"role": "user", "parts": [{"text": "Summarize this document"}]},
    {"role": "agent", "parts": [{"text": "Here is the summary..."}]}
  ],
  "artifacts": [
    {"name": "summary", "parts": [{"text": "The document covers..."}]}
  ],
  "metadata": {},
  "created_at": "2026-05-18T11:59:58+00:00",
  "updated_at": "2026-05-18T12:00:00+00:00"
}

​

Helpers

• validation.validate_uuid_type / normalization.normalize_uuid — coerce string UUIDs to UUID and reject bad input early.
• normalization.normalize_message_uuids — fix up task_id, context_id, message_id, reference_task_ids on inbound messages.
• serialization.serialize_for_jsonb — recursively convert UUID to str so JSONB stays JSON-safe.
• security.mask_database_url — strip the password before logging (user:***@host).
• security.sanitize_identifier — used when DID-based schema isolation rewrites search_path; rejects anything that isn’t [A-Za-z0-9_].
• db_operations.get_current_utc_timestamp, prepare_jsonb_value, create_update_values — common building blocks for UPDATE statements.

​

PostgreSQL Setup

docker run -d \
  --name bindu-postgres \
  -e POSTGRES_USER=bindu_user \
  -e POSTGRES_PASSWORD=secret \
  -e POSTGRES_DB=bindu_db \
  -p 5432:5432 \
  postgres:16

uv add 'sqlalchemy[asyncio]' asyncpg alembic

export STORAGE_TYPE=postgres
export DATABASE_URL='postgresql+asyncpg://bindu_user:secret@localhost:5432/bindu_db'

export STORAGE__RUN_MIGRATIONS_ON_STARTUP=true

uv run alembic upgrade head

uv run alembic history          # see applied/pending migrations
uv run alembic downgrade -1     # rollback one revision

​

Retries, Pooling, Durability

• Retries. Memory wraps each call in a @retry_storage_operation decorator (3 attempts, 0.1s → 1.0s backoff). Postgres routes every call through execute_with_retry from bindu.utils.retry, governed by STORAGE__POSTGRES_MAX_RETRIES and STORAGE__POSTGRES_RETRY_DELAY. Same Tenacity primitives used elsewhere — see Retry.
• Pooling. PostgresStorage creates a single async_engine per process with pool_size=postgres_pool_max, max_overflow=0, and pool_pre_ping=True so dead connections are detected before they bite a request.
• Transactions. Multi-step writes (submit_task, update_task) run inside session.begin(). The status update and the history/artifacts append commit atomically.
• Durability. Memory loses everything on restart, including queued webhook callbacks. Postgres persists every state transition; webhook configs reload on startup via load_all_webhook_configs.

​

Real-World Use Cases

​

Security Best Practices

​

Related

• Architecture
• Retry
• Scheduler
• Observability

Storage is what turns a stateless script into an agent that remembers — every task, every turn, every artifact, every callback.