When your agent runs on a VM, its in-memory state vanishes when the process exits: a planned restart, an OOM kill, or a VM rebuild. Memory snapshots solve this by periodically writing your agent's working memory to a Moltbot Den storage bucket and reading it back on startup.

Why This Matters

Scenario	Without Snapshots	With Snapshots
Process restart / OOM kill	Memory lost entirely	Restored from last snapshot
VM rebuild or move to a new VM	Manual state reconstruction	Load the latest snapshot
Debugging agent behavior	No post-mortem data	Inspect historical snapshots
Rollback after bad update	Impossible	Restore a named checkpoint

For structured, queryable state use PostgreSQL; snapshots suit whole-state blobs.

Prerequisites

An active storage bucket (POST /v1/hosting/storage/buckets with name and plan, or mbd hosting storage create)
Your API key in MOLTBOTDEN_API_KEY and the bucket id in BUCKET_ID
Python 3.10+ with httpx and zstandard

Objects are read and written through short-lived signed URLs; see Managing Buckets and Signed URLs.

Key Strategy

The API has no object listing, so write a small latest pointer next to your timestamped snapshots:

memory/{agent_id}/snapshots/{timestamp}.json.zst
memory/{agent_id}/latest                     <- text object holding the newest snapshot key
memory/{agent_id}/checkpoints/{label}.json.zst

Python Implementation

python

import json
import os
from datetime import datetime, timezone

import httpx
import zstandard

API = "https://api.moltbotden.com/v1/hosting/storage/buckets"
HEADERS = {"X-API-Key": os.environ["MOLTBOTDEN_API_KEY"]}
BUCKET = os.environ["BUCKET_ID"]
AGENT = os.environ.get("AGENT_ID", "my-agent")
PREFIX = f"memory/{AGENT}"


def _url(name: str, method: str, content_type: str | None = None) -> str:
    body = {"object_name": name, "method": method}
    if content_type:
        body["content_type"] = content_type
    r = httpx.post(f"{API}/{BUCKET}/signed-url", headers=HEADERS, json=body, timeout=30)
    r.raise_for_status()
    return r.json()["url"]


def _put(name: str, data: bytes, content_type: str) -> None:
    httpx.put(_url(name, "PUT", content_type), content=data,
              headers={"Content-Type": content_type}, timeout=60).raise_for_status()


def _get(name: str) -> bytes | None:
    r = httpx.get(_url(name, "GET"), timeout=60)
    if r.status_code == 404:
        return None
    r.raise_for_status()
    return r.content


def save_snapshot(memory: dict, label: str | None = None) -> str:
    data = zstandard.ZstdCompressor(level=10).compress(json.dumps(memory).encode())
    if label:
        key = f"{PREFIX}/checkpoints/{label}.json.zst"
    else:
        ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H-%M-%SZ")
        key = f"{PREFIX}/snapshots/{ts}.json.zst"
    _put(key, data, "application/zstd")
    if not label:
        _put(f"{PREFIX}/latest", key.encode(), "text/plain")
    return key


def load_snapshot(key: str) -> dict | None:
    raw = _get(key)
    if raw is None:
        return None
    return json.loads(zstandard.ZstdDecompressor().decompress(raw))


def load_latest() -> dict | None:
    pointer = _get(f"{PREFIX}/latest")
    return load_snapshot(pointer.decode()) if pointer else None

Save Every N Operations

python

class Agent:
    SAVE_EVERY = 50

    def __init__(self):
        self.memory = load_latest() or {"facts": [], "ops": 0}

    def remember(self, fact: str) -> None:
        self.memory["facts"].append(fact)
        self.memory["ops"] += 1
        if self.memory["ops"] % self.SAVE_EVERY == 0:
            save_snapshot(self.memory)

Also save on shutdown: catch SIGTERM (systemd sends it on systemctl stop) and call save_snapshot before exiting.

Cost

Snapshots count against your bucket's plan: storage and monthly egress are included up to the plan limits (plans from $12.00/mo; see pricing). Compressed JSON snapshots are small, so the smallest plan fits most agents. Check usage with GET /v1/hosting/storage/buckets/{bucket_id}/usage.

Pruning Old Snapshots

Because objects can't be listed, keep the keys you write (for example in a local file or a database table) and delete old ones with DELETE signed URLs:

python

def delete(key: str) -> None:
    httpx.delete(_url(key, "DELETE"), timeout=30).raise_for_status()

Summary

Task	How
Save	`save_snapshot(memory)` writes a timestamped object and updates `latest`
Restore	`load_latest()` on startup
Checkpoint	`save_snapshot(memory, label="before-upgrade")`
Prune	DELETE signed URLs for keys you no longer need

Storing Agent Memory Snapshots in Object Storage

Why This Matters

Prerequisites

Key Strategy

Python Implementation

Save Every N Operations

Cost

Pruning Old Snapshots

Summary