Skip to main content

Sync API

Cursor-based delta sync for efficiently pulling changes since your last sync. Designed for third-party apps that need to keep a local index or cache in sync with a user's Paradigm graph.

Base URL​

https://api.ofself.ai/api/v1

Authentication​

All requests require:

  • API Key: X-API-Key: <your-key> + X-User-ID: <user-id> (third-party apps)
  • JWT: Authorization: Bearer <token> (first-party only)

Endpoints​

GET /sync​

GET Delta sync — returns changes (creates, updates, deletes) since the given cursor, filtered by the app's exposure profile.

curl -X GET "https://api.ofself.ai/api/v1/sync?cursor=abc123&limit=100&include_tags=true" \
  -H "X-API-Key: your-api-key" \
  -H "X-User-ID: user-123"

Query Parameters:

ParameterTypeDefaultDescription
cursorstring-Opaque cursor from a previous sync response. Omit for initial full sync.
scopestringallFilter scope: all, tag:<tag_uuid>, or schema:<schema_uuid>
limitinteger100Max changes per page (max 1000)
ids_onlybooleanfalseIf true, return only node IDs instead of full node data
include_tagsbooleanfalseIf true, include tags in node data

Response: 200 OK

{
  "changes": [
    {
      "action": "create",
      "timestamp": "2026-01-15T10:30:00Z",
      "version": 1,
      "node": {
        "id": "node-uuid",
        "title": "First day teaching",
        "node_type": "EXPERIENCE",
        "meaning_level": "IDENTITY",
        "metadata": { "properties": { "significance": "Defining career moment" } },
        "created_at": "2026-01-15T10:30:00Z"
      }
    },
    {
      "action": "update",
      "timestamp": "2026-01-16T14:00:00Z",
      "version": 2,
      "changed_fields": ["title", "value"],
      "node": {
        "id": "node-uuid",
        "title": "Updated title",
        "node_type": "EXPERIENCE"
      }
    },
    {
      "action": "delete",
      "timestamp": "2026-01-17T09:00:00Z",
      "version": 3,
      "node_id": "node-uuid",
      "deleted_title": "Old title"
    }
  ],
  "cursor": "opaque_cursor_string",
  "has_more": false,
  "stats": {
    "created": 1,
    "updated": 1,
    "deleted": 1,
    "total": 3
  }
}
Cursor Behavior
  • Omit cursor for the first sync — you'll receive all historical changes.
  • Pass the cursor from each response in the next request to resume from where you left off.
  • The server automatically stores a SyncCursor per (app, user, scope) for third-party apps.
  • If has_more is true, immediately request again with the new cursor to get the next page.
Scoped Sync

Use scoped sync to track only a subset of a user's graph:

  • scope=tag:<uuid> — only nodes tagged with a specific tag
  • scope=schema:<uuid> — only nodes using a specific schema

Each scope maintains its own cursor, so you can sync different subsets at different rates.

GET /sync/status​

GET Sync health and statistics. Returns pending change count, last sync time, and per-scope breakdown.

curl -X GET "https://api.ofself.ai/api/v1/sync/status" \
  -H "X-API-Key: your-api-key" \
  -H "X-User-ID: user-123"

Response: 200 OK

{
  "total_versions": 150,
  "latest_change": "2026-01-17T09:00:00Z",
  "last_sync_at": "2026-01-16T14:00:00Z",
  "pending_changes": 5,
  "scopes": {
    "all": {
      "last_sync_at": "2026-01-16T14:00:00Z",
      "last_version_seen": 145,
      "pending_changes": 5
    },
    "tag:abc-123-def": {
      "last_sync_at": "2026-01-15T10:30:00Z",
      "last_version_seen": 100,
      "pending_changes": 12
    }
  }
}

For JWT (direct user) access, pending_changes is 0 and scopes is omitted since cursor tracking is for third-party apps.

Sync Workflow​

Initial Full Sync​

import requests

API_KEY = "ofs_tp_xxxx.yyyy"
USER_ID = "user-123"
headers = {"X-API-Key": API_KEY, "X-User-ID": USER_ID}

cursor = None
all_changes = []

while True:
    params = {"limit": 500, "include_tags": "true"}
    if cursor:
        params["cursor"] = cursor

    resp = requests.get(
        "https://api.ofself.ai/api/v1/sync",
        headers=headers, params=params
    ).json()

    all_changes.extend(resp["changes"])
    cursor = resp["cursor"]

    if not resp["has_more"]:
        break

# Store cursor for next sync
save_cursor(cursor)
print(f"Synced {len(all_changes)} changes")

Incremental Sync​

cursor = load_saved_cursor()

resp = requests.get(
    "https://api.ofself.ai/api/v1/sync",
    headers=headers,
    params={"cursor": cursor, "include_tags": "true"}
).json()

for change in resp["changes"]:
    if change["action"] == "create":
        index_node(change["node"])
    elif change["action"] == "update":
        update_indexed_node(change["node"])
    elif change["action"] == "delete":
        remove_from_index(change["node_id"])

save_cursor(resp["cursor"])

Lightweight ID-only Sync​

For bandwidth-sensitive scenarios, fetch only node IDs first, then selectively fetch full data:

resp = requests.get(
    "https://api.ofself.ai/api/v1/sync",
    headers=headers,
    params={"cursor": cursor, "ids_only": "true"}
).json()

# Only fetch full data for nodes you actually need
for change in resp["changes"]:
    if change["action"] != "delete" and is_relevant(change["node_id"]):
        node = requests.get(
            f"https://api.ofself.ai/api/v1/nodes/{change['node_id']}",
            headers=headers
        ).json()
        process_node(node)

Error Responses​

400 Bad Request — Invalid cursor or scope

{
  "error": {
    "code": "INVALID_CURSOR",
    "message": "The provided cursor is invalid or corrupted. Omit cursor for a full sync."
  }
}
{
  "error": {
    "code": "INVALID_SCOPE",
    "message": "Scope must be 'all', 'tag:<uuid>', or 'schema:<uuid>'"
  }
}

403 Forbidden — App lacks read permission

{
  "error": {
    "code": "FORBIDDEN",
    "message": "App does not have read permission for nodes"
  }
}

Next Steps​