> ## Documentation Index
> Fetch the complete documentation index at: https://docs.maximem.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Migrate from Supermemory to Synap
> Move your Supermemory export into Synap with a ready-to-run script: map container tags onto Synap scopes, convert the export, verify the scope assignment, and ingest.
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](/migrations/how-it-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
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).
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.
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](/concepts/memory-architecture).
You need an Instance and an API key; see [Quickstart](/getting-started/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.
```python supermemory_to_synap.py expandable theme={null}
#!/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:
```bash theme={null}
pip install maximem-synap
```
## The migration, step by step
```bash theme={null}
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:
```json theme={null}
{
"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.
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.
```bash theme={null}
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.
`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.
```python theme={null}
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.
`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.
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:
```python theme={null}
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,
)
```
```bash theme={null}
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.
Dry-run first to see what would be sent:
```bash theme={null}
export SYNAP_API_KEY=your_key_here
python3 supermemory_to_synap.py ingest ./synap_import --dry-run
```
Then run it for real:
```bash theme={null}
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.
Memories are built in the background, so ingestion finishing is not the same as memories being ready.
```bash theme={null}
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.
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:
```python theme={null}
from maximem_synap import MaximemSynapSDK
sdk = MaximemSynapSDK()
await sdk.initialize()
ctx = await sdk.user.context.fetch(
user_id="john",
customer_id="acme",
search_query=[""],
)
```
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.
Confirm the facts you expect actually come back:
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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,
)
```
```bash theme={null}
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](/concepts/memory-architecture))** — 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
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.
`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.
These came from a Supermemory connector that kept no local copy of the text. Re-sync the connector in Supermemory and export again.
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.
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.
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](/migrations/how-it-works), 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.