Custom Connectors

• Internal data sources — your source has no public API or is behind a VPN that MemorySync cannot reach.
• Custom field mapping — you need to transform proprietary fields into specific metadata shapes before storing them.
• Higher sync frequency — you need tighter control over when and how often data is pushed.
• Unsupported provider — the service you use (e.g., Confluence, Asana, your own CMS) does not have a managed integration yet.
• Pre-processing — you want to clean, filter, or enrich content before it becomes a memory.

A custom connector is a consumer of the MemorySync API. It follows this pattern:

1. 1Your connector authenticates with MemorySync using an API key (set via the X-API-Key header).
2. 2It fetches data from your external source using whatever client library or API that source provides.
3. 3For each item, it calls POST /memory/add (or client.add() via the SDK) to store the content as a memory.
4. 4It persists a cursor so the next run only fetches new or updated items.

To avoid re-importing the same data on every run, maintain a cursor that tracks your position in the external source. This is the same pattern the managed integrations use internally:

• Persist the cursor in your own store — a database row, a file, or an environment variable. Advance it only after the batch commits successfully.
• Use timestamps or IDs — most APIs support since, updated_after, or cursor tokens. Use whatever the source provides.
• Handle failures gracefully — if a batch fails mid-way, do not advance the cursor. The next run will retry from the same position.

To make replays and retries safe, pass a stable Idempotency-Key header with each request. Key it on the native object ID from your source:

import requests

response = requests.post(
    "https://api.memorysync.io/memory/add",
    headers={
        "X-API-Key": api_key,
        "Idempotency-Key": f"my-source-{item.id}",
    },
    json={
        "user_id": user_id,
        "text": item.content,
        "source": "my_tool",
        "metadata": {"native_id": item.id},
    },
)

If you re-send the same idempotency key, MemorySync returns the previously created memory instead of creating a duplicate.

Here is a minimal connector using the Python SDK. It fetches changes from your source, stores each item as a memory, and advances the cursor:

from memorysync import MemorySyncClient

client = MemorySyncClient(api_key="ms_live_...")
cursor = load_cursor()  # your persistence layer

for item in fetch_changes(cursor):
    client.add(
        user_id=1,
        text=item.text,
        source="my_tool",
        metadata={"native_id": item.id, "source": "my_tool"},
        deduplicate=True,
    )

save_cursor(cursor)  # advance only after success

Tag every memory with a source field and include a metadata.source key so you can filter by origin later:

# Store with source tag
client.add(
    user_id=1,
    text="Meeting notes from Q4 planning",
    source="confluence",
    metadata={"source": "confluence", "space": "ENGINEERING"},
)

# Later, query only memories from this source
results = client.query(
    user_id=1,
    query="Q4 planning",
    filters={"sources": ["confluence"]},
)

• Subscribe to memory.created webhooks — confirm that your connector's writes are being received by MemorySync in real time.
• Track error rates — the SDK raises RateLimitError (429) and ValidationError (400/422). Log these and alert on spikes.
• Monitor cursor lag — if the cursor timestamp falls too far behind the current time, your connector is not keeping up.
• Use the health endpoint — GET /health (or client.health()) returns the API status. Call it from your connector's startup to fail fast if MemorySync is unreachable.
• Handle rate limits gracefully — the SDK automatically retries on 429 if the Retry-After header is ≤ 120 seconds. For longer waits, it raises RateLimitError with the retry_after value.