Migrate from Supermemory to Synap

This guide moves a Supermemory export into Synap. It comes with a script that does the mechanical work, and explains the two decisions the script cannot make for you: which scope each container tag belongs to, and what to do about memory quality.

This is the hands-on procedure for Supermemory. For the method every migration shares — scope mapping, configuring your instance, verifying, and cutting over — see How migration works.

How Supermemory stores memory

Supermemory separates documents (the content you ingest) from memories (short facts extracted from those documents). Both are organised by container tags — flat string identifiers, each with its own isolated namespace. Tags can encode structure by convention, such as org:acme:user:john, but they are not hierarchical: to Supermemory they are opaque strings. A tag holding one person’s chat history and a tag holding your company handbook look identical. That last point is the whole migration. Synap organises memory into three scopes — client, customer, and user — and nothing in a Supermemory export records which tag belongs where. You decide that, and this guide has you verify it twice before it becomes permanent.

Before you start

Export your data from Supermemory

Download the export from your Supermemory dashboard. You get a single JSON file with a documents section (your original content) and a memories section (the facts Supermemory extracted from it).

Check the export is complete

Confirm truncated is false in both sections. If either says true, the export was cut short at its item limit — re-export before going further. The script warns you, but cannot recover content that is not in the file.

Configure your Instance for your agent

Extraction quality depends on your Instance’s memory architecture, generated from the use-case file you supply at instance creation. If you created your Instance without one, add it before this import — otherwise your historical data is extracted with generic defaults, and you would re-import later to benefit from a tuned configuration. See Memory Architecture.

Have a Synap Instance ready

You need an Instance and an API key; see Quickstart. Note whether your Instance treats customers and users as separate (B2B) or as the same thing (B2C) — you need this in step 2.

Get the script

Save the script below as supermemory_to_synap.py in your working directory. Use the copy button in the top right of the block.

supermemory_to_synap.py

#!/usr/bin/env python3
"""
supermemory_to_synap.py — migrate a Supermemory export into Synap.

Supermemory organises data with flat `container tags`. Synap organises it with a
three-level scope hierarchy (client > customer > user). Nothing in a Supermemory
export records which tag is one person, which is a team, and which is company-wide
reference material — so that mapping cannot be inferred. You declare it once, in a
scope map, and this tool applies it.

Usage — four steps:

  1. List every container tag in your export and write a scope map to fill in:
       python3 supermemory_to_synap.py map export.json -o scope_map.json

  2. Open scope_map.json and set the scope for each tag (instructions are inside).

  3. Convert to Synap-ready batch files. Writes files only; contacts no server:
       python3 supermemory_to_synap.py convert export.json -m scope_map.json -o ./synap_import

  4. Load them into Synap (supports --dry-run, and resumes if interrupted):
       export SYNAP_API_KEY=...
       python3 supermemory_to_synap.py ingest ./synap_import

`map` and `convert` need only the standard library. `ingest` additionally needs
the Synap SDK:  pip install maximem-synap

Requires Python 3.9 or newer.
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import re
import sys
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

# ---------------------------------------------------------------------------
# Ingest constraints. Very short content carries too little signal to memorise
# and is discarded on arrival without raising an error, so we check locally and
# report it rather than letting documents disappear silently.
# ---------------------------------------------------------------------------
MIN_CONTENT_LEN = {
    "ai-chat-conversation": 10,
    "email": 50,
    "meeting-transcript": 50,
    "document": 100,
}
DEFAULT_MIN_LEN = 50

# Supermemory document type -> Synap document_type. Types with no direct Synap
# equivalent become "document"; the export stores their extracted text anyway.
TYPE_MAP = {
    "text": "document",
    "pdf": "pdf",
    "image": "image",
    "audio": "audio",
    "video": "audio",
    "granola": "meeting-transcript",
    "tweet": "document",
    "webpage": "document",
    "notion_doc": "document",
    "google_doc": "document",
    "google_slide": "document",
    "google_sheet": "document",
    "github_markdown": "document",
    "onedrive": "document",
}

# Supermemory wraps chat sessions as a header plus a JSON turn array. Both are
# optional; plain-text documents pass through untouched.
SESSION_DATE_RE = re.compile(
    r"(?:date .{0,40}took place|session date)\s*:\s*"
    r"([0-9]{1,2}:[0-9]{2}\s*[ap]\.?m\.?\s+on\s+[0-9]{1,2}\s+\w+,?\s*[0-9]{4}"
    r"|[0-9]{4}-[0-9]{2}-[0-9]{2}[T ][0-9]{2}:[0-9]{2}(?::[0-9]{2})?"
    r"|[0-9]{1,2}\s+\w+,?\s*[0-9]{4})",
    re.IGNORECASE,
)
TURN_ARRAY_RE = re.compile(r"stringified JSON\s*:\s*(\[.*)", re.S)
DATE_FORMATS = (
    "%I:%M %p on %d %B, %Y", "%I:%M %p on %d %b, %Y",
    "%I:%M %p on %d %B %Y",  "%I:%M %p on %d %b %Y",
    "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M",
    "%d %B, %Y", "%d %b, %Y", "%d %B %Y", "%d %b %Y",
)

# Supermemory's documented tag convention is colon-delimited key:value pairs,
# e.g. "org:acme:user:john". Used only to pre-fill suggestions in the scope map.
KNOWN_CUSTOMER_KEYS = ("org", "organisation", "organization", "team", "workspace",
                       "account", "company", "tenant", "customer", "project")
KNOWN_USER_KEYS = ("user", "member", "person", "employee", "uid")


# ---------------------------------------------------------------------------
# Export loading
# ---------------------------------------------------------------------------
def load_export(path: Path) -> tuple[list[dict], list[dict]]:
    """Read a dashboard export or a hand-assembled API dump.

    Dashboard exports look like {documents:{items:[]}, memories:{items:[]}}.
    API dumps are often {documents:[], memories:[]} or a bare document list.
    """
    raw = json.loads(path.read_text(encoding="utf-8"))

    def section(name: str) -> list[dict]:
        node = raw.get(name) if isinstance(raw, dict) else None
        if node is None:
            return []
        if isinstance(node, list):
            return node
        if node.get("truncated"):
            print(
                f"  !! WARNING: '{name}' was truncated at limit {node.get('limit')}."
                f" This export is INCOMPLETE — re-export before migrating.",
                file=sys.stderr,
            )
        return node.get("items", [])

    if isinstance(raw, list):
        return raw, []
    docs, mems = section("documents"), section("memories")
    if not docs and not mems:
        raise SystemExit(f"{path}: no 'documents' or 'memories' found — not a Supermemory export?")
    return docs, mems


def tag_of(doc: dict) -> str | None:
    """A document's container tag, from either the current or deprecated field."""
    if doc.get("containerTag"):
        return doc["containerTag"]
    tags = doc.get("containerTags") or []
    if len(tags) == 1:
        return tags[0]
    if len(tags) > 1:
        # Multiple tags cannot map to one Synap scope; caller reports and skips.
        return "\x00MULTI\x00" + "|".join(sorted(tags))
    return None


# ---------------------------------------------------------------------------
# Subcommand: map
# ---------------------------------------------------------------------------
def suggest_scope(tag: str) -> dict[str, Any]:
    """Best-effort starting point for a tag. Always review these by hand."""
    if ":" in tag:
        parts = tag.split(":")
        pairs = dict(zip(parts[0::2], parts[1::2]))
        cust = next((pairs[k] for k in KNOWN_CUSTOMER_KEYS if k in pairs), None)
        user = next((pairs[k] for k in KNOWN_USER_KEYS if k in pairs), None)
        if cust or user:
            return {"user_id": user, "customer_id": cust}
    return {"user_id": tag, "customer_id": None}


def cmd_map(args) -> None:
    docs, mems = load_export(Path(args.export))
    doc_tags: Counter = Counter()
    for d in docs:
        t = tag_of(d)
        doc_tags[t if t else "\x00UNTAGGED\x00"] += 1
    mem_tags = Counter(m.get("containerTag") for m in mems if m.get("containerTag"))

    tags: dict[str, Any] = {}
    for tag in sorted(set(doc_tags) | set(mem_tags)):
        entry = (suggest_scope(tag) if not tag.startswith("\x00")
                 else {"user_id": None, "customer_id": None})
        entry["_documents"] = doc_tags.get(tag, 0)
        entry["_memories"] = mem_tags.get(tag, 0)
        if tag.startswith("\x00MULTI\x00"):
            entry["_note"] = "document carries MULTIPLE tags — pick one scope or split it"
        tags[tag] = entry

    out = {
        "_README": [
            "Set user_id and customer_id for every tag. Synap derives the scope",
            "level from which of the two you provide — you never name a scope:",
            "  user_id + customer_id -> USER scope     (this person's own memories)",
            "  customer_id only      -> CUSTOMER scope (shared across that customer's users)",
            "  neither (both null)   -> CLIENT scope   (shared across your whole account)",
            "  user_id only          -> INVALID in b2b; set customer_id too",
            "",
            "isolation: 'b2b' if your Instance separates customers from users;",
            "'b2c' if one customer == one user. Must match the Instance's setting.",
            "",
            "The suggestions below are guesses from the tag string. Review every one:",
            "a tag holding company-wide reference material belongs at CLIENT scope,",
            "not USER scope, and nothing in the export can tell them apart.",
        ],
        "isolation": args.isolation,
        "tags": tags,
    }
    Path(args.out).write_text(json.dumps(out, indent=2), encoding="utf-8")
    print(f"Wrote {args.out}")
    print(f"  {len(tags)} container tags | {len(docs)} documents | {len(mems)} memories")
    print(f"  Next: edit {args.out}, then run `convert`.")


# ---------------------------------------------------------------------------
# Conversion helpers
# ---------------------------------------------------------------------------
def parse_session_date(content: str) -> tuple[datetime | None, bool]:
    """(date, header_present). A present-but-unparseable header is an error.

    Supermemory's createdAt is when the document was UPLOADED, which for
    imported history is not when the conversation happened. Where the content
    carries a real session date we must use it, or every memory is timestamped
    to the upload date and time-relative questions answer wrongly.
    """
    m = SESSION_DATE_RE.search(content[:600])
    if not m:
        return None, False
    raw = re.sub(r"\s+", " ", m.group(1).replace(".", "")).strip()
    for fmt in DATE_FORMATS:
        try:
            return datetime.strptime(raw, fmt), True
        except ValueError:
            continue
    return None, True


def render_content(content: str) -> tuple[str, bool, int]:
    """(text, is_conversation, n_turns).

    Turn arrays are flattened to `role: text` lines. Keeping the speaker labels
    lets Synap split a long session on turn boundaries rather than mid-sentence,
    which preserves who said what.
    """
    m = TURN_ARRAY_RE.search(content)
    if m:
        try:
            turns = json.loads(m.group(1))
            lines = [
                f"{t.get('role', 'user')}: {(t.get('content') or '').strip()}"
                for t in turns
                if isinstance(t, dict) and (t.get("content") or "").strip()
            ]
            if lines:
                return "\n\n".join(lines), True, len(lines)
        except (json.JSONDecodeError, AttributeError):
            pass  # fall through and treat as plain text
    text = content.strip()
    looks_chatty = bool(re.search(r"^(user|assistant|human|ai)\s*:", text, re.I | re.M))
    return text, looks_chatty, 0


def resolve_scope(tag: str, scope_map: dict, isolation: str) -> tuple[str | None, str | None]:
    entry = scope_map.get(tag)
    if entry is None:
        raise ValueError(f"tag {tag!r} is not in the scope map — re-run `map` or add it")
    user_id = entry.get("user_id") or None
    customer_id = entry.get("customer_id") or None
    if isolation == "b2b" and user_id and not customer_id:
        raise ValueError(
            f"tag {tag!r}: user_id without customer_id is rejected by a b2b Instance"
        )
    return user_id, customer_id


def scope_level(user_id, customer_id, isolation: str) -> str:
    if isolation == "b2c":
        return "user" if (user_id or customer_id) else "client"
    if user_id and customer_id:
        return "user"
    if customer_id:
        return "customer"
    return "client"


# ---------------------------------------------------------------------------
# Subcommand: convert
# ---------------------------------------------------------------------------
def cmd_convert(args) -> None:
    docs, mems = load_export(Path(args.export))
    cfg = json.loads(Path(args.map).read_text(encoding="utf-8"))
    scope_map = cfg.get("tags", {})
    isolation = args.isolation or cfg.get("isolation", "b2b")

    baseline = defaultdict(list)
    for m in mems:
        for did in m.get("documentIds") or []:
            baseline[did].append({"id": m.get("id"), "memory": m.get("memory")})

    requests: list[dict] = []
    skipped: list[dict] = []
    stats: Counter = Counter()
    levels: Counter = Counter()
    dct_src: Counter = Counter()

    for doc in docs:
        did = doc.get("id") or doc.get("customId")
        tag = tag_of(doc)

        if not tag or tag.startswith("\x00"):
            skipped.append({"id": did, "reason": "no single container tag"})
            stats["skip_no_tag"] += 1
            continue
        try:
            user_id, customer_id = resolve_scope(tag, scope_map, isolation)
        except ValueError as e:
            skipped.append({"id": did, "tag": tag, "reason": str(e)})
            stats["skip_scope"] += 1
            continue

        raw = doc.get("content") or doc.get("raw") or ""
        if not raw.strip():
            # Connector-sourced documents often keep no local copy of the text.
            skipped.append({"id": did, "tag": tag, "reason": "empty content in export"})
            stats["skip_empty"] += 1
            continue

        body, is_convo, n_turns = render_content(raw)
        sm_type = (doc.get("type") or "text").lower()
        doc_type = "ai-chat-conversation" if is_convo else TYPE_MAP.get(sm_type, "document")

        min_len = MIN_CONTENT_LEN.get(doc_type, DEFAULT_MIN_LEN)
        if len(body) < min_len:
            skipped.append({
                "id": did, "tag": tag,
                "reason": f"{len(body)} chars is below the {min_len}-char minimum "
                          f"for '{doc_type}' and would be discarded silently",
            })
            stats["skip_too_short"] += 1
            continue

        session_dt, had_header = parse_session_date(raw)
        if session_dt:
            dct, src = session_dt, "session_header"
        elif had_header:
            skipped.append({"id": did, "tag": tag,
                            "reason": "session-date header present but unparseable"})
            stats["skip_bad_date"] += 1
            continue
        elif doc.get("createdAt"):
            try:
                dct = datetime.fromisoformat(doc["createdAt"].replace("Z", "+00:00"))
            except ValueError:
                dct, src = datetime.now(timezone.utc), "now"
            else:
                src = "created_at"
        else:
            dct, src = datetime.now(timezone.utc), "now"
        dct_src[src] += 1

        lvl = scope_level(user_id, customer_id, isolation)
        levels[lvl] += 1
        stats["turns"] += n_turns

        requests.append({
            "document": body,
            "document_type": doc_type,
            "document_id": did,                       # re-runs stay idempotent
            "document_created_at": dct.isoformat(),
            "user_id": user_id,
            "customer_id": customer_id,
            "mode": args.mode,
            "metadata": {
                "source": "supermemory_export",
                "supermemory_doc_id": did,
                "container_tag": tag,
                "synap_scope": lvl,
                "dct_source": src,
                "title": doc.get("title"),
                "sm_type": sm_type,
                "sm_created_at": doc.get("createdAt"),
                "sm_memory_count": len(baseline.get(did, [])),
            },
        })
        stats["converted"] += 1

    out = Path(args.out)
    out.mkdir(parents=True, exist_ok=True)
    batches = [requests[i:i + args.batch_size] for i in range(0, len(requests), args.batch_size)]
    for i, b in enumerate(batches):
        (out / f"batch_{i:03d}.json").write_text(
            json.dumps({"documents": b, "fail_fast": False}, indent=1), encoding="utf-8")
    if baseline:
        (out / "baseline.json").write_text(json.dumps(dict(baseline), indent=1), encoding="utf-8")

    report = {
        "source": args.export,
        "isolation": isolation,
        "documents_in_export": len(docs),
        "memories_in_export": len(mems),
        "converted": stats["converted"],
        "turns_rendered": stats["turns"],
        "batches": len(batches),
        "scope_breakdown": dict(levels),
        "dct_sources": dict(dct_src),
        "skipped": {k: v for k, v in stats.items() if k.startswith("skip")},
        "skipped_detail": skipped,
    }
    (out / "report.json").write_text(json.dumps(report, indent=2), encoding="utf-8")

    print(f"Converted {stats['converted']}/{len(docs)} documents -> {len(batches)} batches in {out}/")
    print(f"  scope breakdown : {dict(levels) or 'none'}")
    print(f"  date sources    : {dict(dct_src) or 'none'}")
    print(f"  skipped         : {report['skipped'] or 'none'}")
    if skipped:
        print(f"  -> see {out}/report.json for the reason on each skipped document")
    if dct_src.get("created_at") or dct_src.get("now"):
        print("  note: some documents fell back to their upload date because the export"
              "\n        carried no original timestamp; time-relative recall on those is approximate.")


# ---------------------------------------------------------------------------
# Subcommand: ingest
# ---------------------------------------------------------------------------
async def _ingest(args) -> None:
    try:
        from maximem_synap import MaximemSynapSDK, CreateMemoryRequest, RateLimitError
    except ImportError:
        raise SystemExit("ingest requires the Synap SDK:  pip install maximem-synap")

    files = sorted(Path(args.dir).glob("batch_*.json"))
    if not files:
        raise SystemExit(f"no batch_*.json in {args.dir} — run `convert` first")

    # Resume marker, so an interrupted run continues instead of re-sending.
    done_file = Path(args.dir) / ".ingested"
    done = set(done_file.read_text().split()) if done_file.exists() else set()
    if done:
        print(f"Resuming — {len(done)} batch(es) already sent.")

    if args.dry_run:
        for f in files:
            if f.name in done:
                continue
            n = len(json.loads(f.read_text())["documents"])
            print(f"  [dry-run] would send {f.name} ({n} documents)")
        return

    if not (args.api_key or os.environ.get("SYNAP_API_KEY")):
        raise SystemExit("set SYNAP_API_KEY or pass --api-key")

    sdk = MaximemSynapSDK(api_key=args.api_key) if args.api_key else MaximemSynapSDK()
    await sdk.initialize()

    totals: Counter = Counter()
    ingestion_ids: list[str] = []
    try:
        for f in files:
            if f.name in done:
                continue
            payload = json.loads(f.read_text(encoding="utf-8"))
            batch = [CreateMemoryRequest(**d) for d in payload["documents"]]

            for attempt in range(1, args.retries + 1):
                try:
                    result = await sdk.memories.batch_create(documents=batch, fail_fast=False)
                    break
                except RateLimitError:
                    # The whole batch is checked against your quota up front, so
                    # this means the batch did not fit rather than that it failed.
                    if attempt == args.retries:
                        raise
                    wait = args.backoff * attempt
                    print(f"  {f.name}: quota exceeded, waiting {wait}s "
                          f"(attempt {attempt}/{args.retries})")
                    await asyncio.sleep(wait)

            totals["succeeded"] += result.succeeded
            totals["failed"] += result.failed
            ingestion_ids += [str(r.ingestion_id) for r in result.results if r.ingestion_id]
            done.add(f.name)
            done_file.write_text("\n".join(sorted(done)), encoding="utf-8")
            print(f"  {f.name}: {result.succeeded}/{len(batch)} accepted  batch_id={result.batch_id}")

        (Path(args.dir) / "ingestion_ids.json").write_text(
            json.dumps(ingestion_ids, indent=1), encoding="utf-8")
        print(f"\nAccepted {totals['succeeded']} documents ({totals['failed']} rejected).")
        print(f"Wrote {len(ingestion_ids)} ingestion ids to {args.dir}/ingestion_ids.json")
        print("Memories are built in the background — run `verify` to watch them finish.")
    finally:
        await sdk.shutdown()


def cmd_ingest(args) -> None:
    asyncio.run(_ingest(args))


# ---------------------------------------------------------------------------
# Subcommand: verify
# ---------------------------------------------------------------------------
async def _verify(args) -> None:
    try:
        from maximem_synap import MaximemSynapSDK
    except ImportError:
        raise SystemExit("verify requires the Synap SDK:  pip install maximem-synap")

    ids_file = Path(args.dir) / "ingestion_ids.json"
    if not ids_file.exists():
        raise SystemExit(f"{ids_file} not found — run `ingest` first")
    ids = json.loads(ids_file.read_text(encoding="utf-8"))

    sdk = MaximemSynapSDK(api_key=args.api_key) if args.api_key else MaximemSynapSDK()
    await sdk.initialize()

    outcomes: Counter = Counter()
    memories = 0
    incomplete: list[dict] = []
    try:
        from uuid import UUID
        for i, ing in enumerate(ids, 1):
            try:
                st = await sdk.memories.wait_for_completion(
                    UUID(ing), timeout_seconds=args.timeout)
            except TimeoutError:
                outcomes["timeout"] += 1
                incomplete.append({"ingestion_id": ing, "status": "timeout"})
                continue
            status = st.status.value if hasattr(st.status, "value") else str(st.status)
            outcomes[status] += 1
            memories += st.memories_created
            if status != "completed":
                incomplete.append({"ingestion_id": ing, "status": status,
                                   "error": st.error_message})
            if i % 25 == 0:
                print(f"  checked {i}/{len(ids)} ...")
    finally:
        await sdk.shutdown()

    print(f"\n{len(ids)} ingestions -> {dict(outcomes)}")
    print(f"{memories} memories created.")
    if incomplete:
        p = Path(args.dir) / "incomplete.json"
        p.write_text(json.dumps(incomplete, indent=1), encoding="utf-8")
        print(f"{len(incomplete)} did not complete cleanly — details in {p}")
        print("'partial_success' means the document was processed but some memories "
              "were not stored; re-ingesting that document is safe.")


def cmd_verify(args) -> None:
    asyncio.run(_verify(args))


# ---------------------------------------------------------------------------
def main() -> None:
    p = argparse.ArgumentParser(
        description="Migrate a Supermemory export into Synap.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="Run `map` first, edit the scope map, then `convert`, `ingest`, `verify`.",
    )
    sub = p.add_subparsers(dest="cmd", required=True)

    m = sub.add_parser("map", help="list every container tag and write a scope map to fill in")
    m.add_argument("export")
    m.add_argument("-o", "--out", default="scope_map.json")
    m.add_argument("--isolation", choices=("b2b", "b2c"), default="b2b")
    m.set_defaults(func=cmd_map)

    c = sub.add_parser("convert", help="apply the scope map and write Synap batch files")
    c.add_argument("export")
    c.add_argument("-m", "--map", default="scope_map.json")
    c.add_argument("-o", "--out", default="./synap_import")
    c.add_argument("--isolation", choices=("b2b", "b2c"), default=None,
                   help="override the value stored in the scope map")
    c.add_argument("--batch-size", type=int, default=25)
    c.add_argument("--mode", choices=("fast", "long-range"), default="long-range")
    c.set_defaults(func=cmd_convert)

    i = sub.add_parser("ingest", help="send the batches to Synap (resumable)")
    i.add_argument("dir")
    i.add_argument("--api-key", default=None, help="defaults to $SYNAP_API_KEY")
    i.add_argument("--dry-run", action="store_true")
    i.add_argument("--retries", type=int, default=3)
    i.add_argument("--backoff", type=int, default=30)
    i.set_defaults(func=cmd_ingest)

    v = sub.add_parser("verify", help="wait for ingestion to finish and report results")
    v.add_argument("dir")
    v.add_argument("--api-key", default=None, help="defaults to $SYNAP_API_KEY")
    v.add_argument("--timeout", type=int, default=300)
    v.set_defaults(func=cmd_verify)

    args = p.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()

map and convert run offline using only the standard library. ingest and verify talk to Synap through the SDK:

pip install maximem-synap

The migration, step by step

List your container tags

python3 supermemory_to_synap.py map supermemory-export.json -o scope_map.json

This reads the export and writes a scope_map.json listing every container tag it found, with the number of documents and memories in each:

{
  "isolation": "b2b",
  "tags": {
    "org:acme:user:john": {
      "user_id": "john",
      "customer_id": "acme",
      "_documents": 128,
      "_memories": 941
    },
    "acme_handbook": {
      "user_id": "acme_handbook",
      "customer_id": null,
      "_documents": 12,
      "_memories": 87
    }
  }
}

Where a tag follows Supermemory’s key:value convention, the script pre-fills a suggestion. Everything else is a guess you need to correct.

Assign a scope to every tag

Open scope_map.json and set user_id and customer_id for each tag.You never name a scope directly. Synap works out the scope from which identifiers you provide:

`user_id`	`customer_id`	Resulting scope	Use for
set	set	user	One person’s own history
not set	set	customer	Shared across everyone at that customer
not set	not set	client	Shared across your entire account
set	not set	Rejected on a B2B Instance	—

Also set isolation to match your Instance: b2b if customers and users are separate, b2c if one customer is one user.In the example above, acme_handbook is company reference material, so its correct mapping is both fields null — client scope — not the user_id the script guessed.

Container tags are flat and mutually isolated, so nothing in the export distinguishes one person’s history from company-wide material. A tag left at client scope by mistake becomes readable across your whole account. Step 4 has you check this before anything is sent.

Convert the export

python3 supermemory_to_synap.py convert supermemory-export.json \
    -m scope_map.json -o ./synap_import

This writes files only and contacts no server:

File	What it holds
`batch_000.json` …	Your documents, ready to ingest, 25 per file
`baseline.json`	Supermemory’s own extracted memories, keyed by document
`report.json`	Conversion stats and a reason for every skipped document

Open report.json and confirm scope_breakdown matches what you intended, and that skipped is empty or contains only documents you expect to lose.

Review the scope assignment

report.json gives you totals per scope. Before ingesting, check the assignment tag by tag against the actual content — the totals will look correct even when a tag is in the wrong place.

import json, glob, collections

tags = collections.defaultdict(list)
for f in glob.glob("./synap_import/batch_*.json"):
    for d in json.load(open(f))["documents"]:
        tags[d["metadata"]["container_tag"]].append(d)

for tag in sorted(tags):
    docs = tags[tag]
    m = docs[0]
    print(f"\n{tag}")
    print(f"  scope: {m['metadata']['synap_scope'].upper()}   "
          f"user_id={m['user_id']}  customer_id={m['customer_id']}  "
          f"documents={len(docs)}")
    for d in docs[:2]:
        print(f"    - {(d['metadata']['title'] or '(untitled)')[:70]}")
        print(f"      {d['document'][:90].replace(chr(10), ' ')}...")

Read the samples, not just the scope labels. Two things to look for:

A tag at client scope whose samples are somebody’s personal conversation. That content is about to become readable by every user on your account.
A tag at user scope whose samples read like policy, product, or reference documentation. That content will be copied into one person’s memory instead of shared, and no one else will be able to retrieve it.

Fix scope_map.json and re-run convert until every tag reads correctly. Nothing has been sent yet, so this loop is free.

Check how timestamps were resolved

report.json includes a dct_sources breakdown:

session_header — the original conversation date was recovered from the content. This is what you want.
created_at — no original date was available, so the document’s upload date was used instead.

Supermemory’s createdAt records when a document was uploaded to Supermemory, not when the conversation happened. If you imported history into Supermemory, those two dates can be years apart. Memories dated from the upload date still work for recall, but questions like “what did I decide last spring?” answer against the wrong timeline.

Pilot one tag

Migrate a single tag first and confirm it behaves before committing the rest.Copy your scope map, keep one representative tag, and convert that alone into its own directory:

import json

full = json.load(open("scope_map.json"))
tag = "org:acme:user:john"        # pick one real tag
json.dump(
    {"isolation": full["isolation"], "tags": {tag: full["tags"][tag]}},
    open("pilot_map.json", "w"), indent=2,
)

python3 supermemory_to_synap.py convert supermemory-export.json \
    -m pilot_map.json -o ./pilot
python3 supermemory_to_synap.py ingest ./pilot
python3 supermemory_to_synap.py verify ./pilot

Every other tag will be reported as skipped during the pilot conversion. That is expected — they are not in pilot_map.json.Now run steps 9 and 10 against this one tag. Only continue once its isolation and retrieval both check out.

Ingest everything

Dry-run first to see what would be sent:

export SYNAP_API_KEY=your_key_here
python3 supermemory_to_synap.py ingest ./synap_import --dry-run

Then run it for real:

python3 supermemory_to_synap.py ingest ./synap_import

The script records progress as it goes, so if it is interrupted you can re-run the same command and it resumes rather than sending anything twice. Each document is submitted with its original Supermemory ID, so re-running a batch does not create duplicates.

Your whole batch is checked against your quota before any of it is accepted. If you hit the limit, the script waits and retries automatically — you do not need to split the files yourself.

Verify ingestion completed

Memories are built in the background, so ingestion finishing is not the same as memories being ready.

python3 supermemory_to_synap.py verify ./synap_import

This waits for every submitted document and reports how many memories were created. Anything that did not complete cleanly is written to incomplete.json. A partial_success result means the document was processed but some of its memories were not stored; re-ingesting that document is safe.

Confirm scope isolation

This is the check that catches a wrong scope map. Do it on the pilot, and again after the full migration.Fetch context as one user and confirm nothing belonging to another user comes back:

from maximem_synap import MaximemSynapSDK

sdk = MaximemSynapSDK()
await sdk.initialize()

ctx = await sdk.user.context.fetch(
    user_id="john",
    customer_id="acme",
    search_query=["<a topic only another user discussed>"],
)

Choose a search term you know belongs to a different user’s history. Nothing from that user should appear.Then check the reverse for anything you placed at client scope: fetch as two different users and confirm the shared material is retrievable by both. If it reaches only one, that tag is at user scope and needs correcting.

Spot-check retrieval coverage

Confirm the facts you expect actually come back:

ctx = await sdk.user.context.fetch(
    user_id="john",
    customer_id="acme",
    search_query=["dietary preferences"],
)

baseline.json is useful here: it holds Supermemory’s own extracted memories for each document, so you can sample a few and check they are retrievable from Synap too.Do not compare counts. One document does not become one memory, and the two systems extract differently by design — judge the migration by what you can retrieve.

If the scope assignment was wrong

If you discover a misplaced tag after ingesting, correct it one tag at a time. Leave every other tag alone.

Do not delete all of ingestion_ids.json and re-ingest the whole export. Ingestion recognises content it has already seen at the same scope for several days, so a blanket re-ingest returns the earlier result instead of rebuilding. The tags you deleted but did not re-scope would stay deleted. Only remove the memories belonging to the tag you are actually fixing.

Step 1 — Delete only the affected tag’s memories. ingest writes ingestion_ids.json in the output directory, and each ingestion’s status names the document it came from, which the batch files tie back to a container tag:

import json, glob
from uuid import UUID
from maximem_synap import MaximemSynapSDK

BAD_TAG = "acme_handbook"        # the tag you are correcting

# document_id -> container_tag, from the converted batches
doc_tag = {
    d["document_id"]: d["metadata"]["container_tag"]
    for f in glob.glob("./synap_import/batch_*.json")
    for d in json.load(open(f))["documents"]
}

sdk = MaximemSynapSDK()
await sdk.initialize()

removed = 0
for ingestion_id in json.load(open("./synap_import/ingestion_ids.json")):
    status = await sdk.memories.status(UUID(ingestion_id))
    if doc_tag.get(status.document_id) != BAD_TAG:
        continue                  # leave correctly-scoped tags untouched
    for memory_id in status.memory_ids:
        await sdk.memories.delete(UUID(memory_id))
        removed += 1

print(f"removed {removed} memories for {BAD_TAG}")

Step 2 — Re-convert that tag alone. Fix its entry in scope_map.json, then build a single-tag map and convert it into its own directory, exactly as in the pilot step:

import json

full = json.load(open("scope_map.json"))
json.dump(
    {"isolation": full["isolation"], "tags": {BAD_TAG: full["tags"][BAD_TAG]}},
    open("refix_map.json", "w"), indent=2,
)

python3 supermemory_to_synap.py convert supermemory-export.json \
    -m refix_map.json -o ./refix
python3 supermemory_to_synap.py ingest ./refix
python3 supermemory_to_synap.py verify ./refix

Because the tag’s scope has changed, this content is treated as new and is processed rather than matched against the earlier run. Re-run the isolation check from step 9 before moving on.

Keep every output directory until the migration is fully verified. ingestion_ids.json is the only record of which memories the migration created — without it, telling them apart from memories your live application has written since is difficult.

Why the script ingests documents, not memories

A Supermemory export contains both your original documents and the memories Supermemory extracted from them. The script deliberately ingests the documents. Supermemory’s memory entries are short summaries of your content. Importing them means Synap extracts from summaries rather than from what your users actually said — you inherit whatever the original extraction got wrong, and lose the detail it dropped. Because every memory in the export points back to the document it came from, you can re-extract from the original source instead, which is almost always better. There is a second, more important reason to re-extract. Every Synap instance runs its own Memory Architecture Configuration (MACA) — a per-instance memory policy generated from the use-case file you provide when you create the instance. It governs what gets extracted and how, tuned to your agent’s domain and audience. Re-extracting your imported documents through that configuration means your migrated history is processed by the same rules as your live traffic. From the very same source conversations, a support agent’s instance surfaces issues and resolutions, while a companion agent’s instance surfaces preferences and emotional context. Importing Supermemory’s pre-extracted memories would bypass this entirely and leave your historical data shaped by generic rules that do not match your agent — so re-ingestion is not overhead, it is how your old data starts behaving as if your agent created it all along. There is one case where re-extraction is not possible: documents brought in through a Supermemory connector sometimes keep no local copy of their text. Those appear in report.json as empty content in export. Re-sync the connector on the Supermemory side and export again, or accept the loss. The extracted memories are still useful — that is what baseline.json is for. Use them to check your coverage after migrating, not as the thing you migrate.

Supermemory’s memory entries are typically a single short sentence. Content that brief carries too little signal to memorise on its own and is discarded on arrival, so importing those strings directly would quietly lose most of them. The script checks length locally and reports anything at risk rather than letting it disappear.

What you gain

Typed extractions — facts, preferences, episodes, emotions, and temporal events as separate lists, rather than one undifferentiated pool of strings.
Three scopes, not one flat namespace — client, customer, and user, with roll-up between them, so shared knowledge is stored once instead of copied into every tag.
Entity resolution across conversations — the same person or product recognised across sessions.
Context compaction — long histories stay usable without you managing the window.

What you’ll need to adapt

Some Supermemory structure is not present in a dashboard export, and no migration can recover it:

Version history and superseded facts. The export contains only current memories, without their revision chains.
Relationships between memories. Supermemory’s updates / extends / derives links are not included.
Inferred-fact flags. There is no way to tell which memories Supermemory derived rather than observed.
Embeddings. Vectors are never exported by either system; Synap generates its own during ingestion.

None of this is a real loss, because re-extracting from your original documents rebuilds the equivalent structure natively — Synap tracks its own memory lineage and relationships as it ingests.

Troubleshooting

convert skipped every document

Your scope map still has unfilled entries. On a B2B Instance a tag with a user_id but no customer_id is rejected, which is deliberate — it prevents documents landing in an unintended scope. Set customer_id for those tags, or switch isolation to b2c if that matches your Instance.

A tag is missing from the scope map

map only lists tags that appear in the export. If a tag exists in Supermemory but has no documents or memories in the file, it will not appear — and nothing needs migrating for it.

Documents skipped as 'empty content in export'

These came from a Supermemory connector that kept no local copy of the text. Re-sync the connector in Supermemory and export again.

Documents skipped as 'session-date header present but unparseable'

The document announces a session date the script could not read. This is treated as an error rather than silently falling back to the upload date, because a wrong date is worse than a skipped document. Report the format and we will add it.

Fewer memories than documents ingested

Expected. One document does not become one memory — long conversations are split, and content with nothing worth remembering produces none. Judge the migration by what you can retrieve, not by counting rows.

Ingestion stopped partway through

Re-run the same ingest command. Completed batches are recorded and skipped, so it resumes rather than re-sending.

After you cut over

Work through the shared migration method, which covers the parts common to every source: configuring your instance so extraction quality is good from day one, deciding your conversation_id strategy, swapping your retrieval call sites, and adding graceful degradation. Once you are cut over, retire the old service. Do not dual-write — diverging memory state is a harder problem than a clean cutover.

Start here

From your current platform

Migrate from Supermemory to Synap

How Supermemory stores memory

Before you start

Get the script

The migration, step by step

If the scope assignment was wrong

Why the script ingests documents, not memories

What you gain

What you’ll need to adapt

Troubleshooting

After you cut over

​How Supermemory stores memory

​Before you start

​Get the script

​The migration, step by step

​If the scope assignment was wrong

​Why the script ingests documents, not memories

​What you gain

​What you’ll need to adapt

​Troubleshooting

​After you cut over

How Supermemory stores memory

Before you start

Get the script

The migration, step by step

If the scope assignment was wrong

Why the script ingests documents, not memories

What you gain

What you’ll need to adapt

Troubleshooting

After you cut over