Synthigy Python SDK

Thin, zero-dependency (stdlib-only) client for Synthigy's /data endpoint. ONE engine, async-native: everything runs on an asyncio core (AsyncClient) — hand-rolled keep-alive HTTP/1.1 pool, one multiplexed SSE connection, watches as suspended coroutines. The blocking Client (and every module-level verb) is a thin facade over that engine driving a single background event-loop thread — same API scripts and notebooks always had, ~1 extra thread total no matter how many watches are open. Python ≥ 3.10.

Quick start

1. Get a server and connect as your app — steps 1–3 of Code generation.

2. Install:

pip install synthigy            # or: uv add synthigy — zero dependencies

3. Query:

import synthigy

synthigy.connect()          # endpoint + identity from `synthigy exec`

print(sorted(synthigy.schema()["entities"]))     # what this app can see

# XSQL: the shape you write is the shape you get back
for ds in synthigy.query("""
dataset
  name
  ->versions
    name
"""):
    print(ds["name"], [v["name"] for v in ds.get("versions", [])])

synthigy exec -- python app.py

That query runs on any instance — dataset is the built-in model registry.

4. Bring your own model. Deploy a model export (the JSON the modeler writes) and every entity in it becomes a table and an API at once:

with open("My Model@1.0.json") as f:
    synthigy.deploy(f.read())                # needs the Dataset Developer role

synthigy.stack("movie", {"title": "Dune", "release_year": 2021})
rows = synthigy.query("movie (release_year >= ?y:int)\n  title\n", {"y": 2021})

5. Get types — see Code generation.

Where to next: Async-first for FastAPI services, Logging users in for a browser login, Live data for watches.

One process, one client

connect() installs a module-wide default — any previous default is destroyed first (watches close, SSE drops). All module-level verbs operate on it. Identity is multiplexed per-call via acting_as=, never a second connect. Constructing synthigy.Client(...) directly is the escape hatch for tests.

Async-first (BFFs, FastAPI/uvicorn services)

The async surface IS the engine — no thread hops, native cancellation:

synthigy.aconnect()          # once, at startup
c = synthigy.aclient()

rows = await c.query("""
movie (release_year > ?y:int, limit 10)
  title
""", {"y": 1990}, acting_as=user.xid)
await c.stack("movie", {"xid": xid, "title": "Dune"}, acting_as=user.xid)

@app.get("/movies/stream")
async def stream(user=Depends(current_user)):
    async def gen():
        async with c.watch_query_xsql("movie (limit 10)\n  title\n", entity="movie",
                                      acting_as=user.xid) as w:
            yield render(w.list())
            async for ev in w:            # derived query/* events
                yield render(w.list())
    return StreamingResponse(gen(), media_type="text/event-stream")

Request teardown is task cancellation in asyncio; watch cleanup is cancellation-safe by design (async with guarantees close, close never awaits). An open watch costs a buffer on the shared SSE connection — not a thread, not a second connection.

AsyncClient/Client are themselves context managers too (like httpx.AsyncClient/httpx.Client) — async with AsyncClient(...) as c: / with Client(...) as c: close everything on scope exit, exception or not. aconnect/connect (the module-default singleton) stay the norm for long-lived processes; the context-manager form is for scripts and one-off scoped clients (tests, short jobs).

The blocking Client below is for scripts/seeds/notebooks; don't call it from a coroutine (it blocks the loop, and calling it from its own loop thread raises).

Reads

rows  = synthigy.query("""
movie (release_year > ?y:int, limit 10)
  title
  ->genres
    name
""", {"y": 1990})
movie = synthigy.query("""
movie (xid = ?xid:string)
  title
""", {"xid": xid}, op="get")                 # one record, or None
count = synthigy.sql_template(
    "SELECT COUNT(*) AS n FROM {movie} WHERE {movie.release_year} > ?", [1990])

• ?name:type=default are named params, passed as a dict.
• ->genres is a left pull — a movie with no genres is still returned; -genres is inner and keeps only movies that have one.
• Empty relations are omitted, never [] — use row.get("genres", []).
• Counting and aggregation compute in the database: _count / _agg in XSQL, or sql_template.

Writes

synthigy.sync("movie", {"xid": xid, "title": "Dune"})                  # upsert, REPLACES link-sets
synthigy.stack("user_rating", {"value": 5, "movie": {"xid": xid}})     # additive
synthigy.delete("movie", {"xid": xid})                                 # soft delete

slice (unlink without deleting) and purge (hard delete by filter) exist too.

Writes are silent by default — sync/stack answer {"count": n}, not the record. Mint the id up front when you need it; that is cheaper than the echo and makes a retried write idempotent rather than duplicating a row:

from synthigy import new_xid

xid = new_xid()                                     # 22-char Base58
synthigy.sync("Movie", {"xid": xid, "title": "Dune"})          # -> {"count": 1}
synthigy.sync("Movie", {"xid": xid, "title": "Dune"},
              returning=True)                       # -> the written record

Batch heterogeneous ops in one round trip:

from synthigy import ops
results = synthigy.exec_([
    ops.query("movie (limit 3)\n  title\n"),
    ops.stack("user_rating", {"value": 4, "movie": {"xid": xid}}),
], acting_as=user_xid)

Live data

The blessed pattern is notify-then-refetch: events are pokes; the SDK re-runs the query through the IAM-filtered read path, so RLS is enforced on every refetch.

w = synthigy.watch_query_xsql("movie (limit 5, order by release_year desc)\n  title\n",
                              entity="movie", acting_as=user_xid)
w.ready()
for ev in w.events():        # blocking iterator; break to stop
    if ev["type"] in ("query/added", "query/changed", "query/removed"):
        render(w.list())

w2 = synthigy.watch_sql_template(
    "SELECT COUNT(*) AS n FROM {movie}", entities=["movie"])
w2.ready(); print(w2.first())

Lower-level: watch(interest) (records/entities/relations, shaped record/relation deltas with computed changed), watch_schema(), listen() (raw SSE envelopes), observe(descriptor, backfill=True) (one-call subscribe + reconnect + /history gap replay). One SSE connection per client — all watches fuse onto it. keep_alive=True on connect pins the SSE open across watch churn (BFFs).

Raw subscription calls (subscribe/set_subscriptions/...) exist but the server set is per-identity full-replace — raw calls clobber a live watch multiplexer's union. Prefer the watch family.

History

h = synthigy.history()
h.events(record_xid=xid)                  # recent events up to now
h.get_at(xid, "2026-01-01T00:00:00Z")
h.diff(xid, t1, t2)

Raises HISTORY_UNAVAILABLE when the server has no audit provider.

Errors

Everything raises synthigy.SynthigyError with stable .code, derived .category (auth | iam | validation | not_found | conflict | rate_limit | network | internal) and .retryable. Structured fields when the server sends them: .hint, .entity, .path, .line/.col, .diagnostics, .request_id (matches the X-Request-Id the SDK sends — correlate with server logs). Discriminate on .code, never the message.

Auth

• Client credentials (client_id+client_secret): tokens minted from /oauth/token, cached per audience, refreshed 30s before expiry, single-flight; one automatic clear-and-retry on 401.
• audience= (or $SYNTHIGY_AUDIENCE): binds one audience to every mint this client makes. The platform's audience model is opt-in by design — a token minted naming no audience resolves to an identity-only audience that /data rejects, so without this every data call 401s. Set it to the server's /data audience, published at /.well-known/synthigy as auth.oidc.audience. Left unset the SDK names no audience, so an unentitled client keeps a soft 401 rather than a hard invalid_target. client.token(audience) still overrides per call, for minting tokens aimed at a different audience.
• Static token="..." for scripts/tests (token="" for authless dev).
• With none of the above, resolution continues: under SYNTHIGY_SUPERVISED=1 the SDK asks its supervising parent (synthigy exec/agent, or a robotics commander) for a token over the process's own stdio, then falls back to the SYNTHIGY_TOKEN env var, then raises SynthigyError(code="NO_TOKEN") with a message that teaches the fix. The pipe beats the env var deliberately: exec injects the cached token and supervises, and only the pipe can refresh mid-run. A bot written as synthigy.Client(endpoint) — nothing else — therefore runs unchanged bare, under exec, and under a production commander.
• acting_as is server-verified impersonation for trusted confidential clients (the BFF model).

Logging users in (client.login)

Authorization code + PKCE for a confidential server (BFF): the SDK owns the protocol, your app owns sessions, cookies and routing. The SDK never serves a route or issues a redirect — start() returns a URL, complete() takes the callback's code/state; the HTTP handlers are yours. The one piece of state the flow needs between /login and the callback lives in a login_store you supply — anything with put(state, login) and one-shot take(state), sync or async (a dict, redis.asyncio, a DB table).

client = synthigy.AsyncClient(ENDPOINT, client_id=ID, client_secret=SECRET,
                              login_store=synthigy.MemoryLoginStore())

@app.get("/login")
async def login(returnTo: str = "/"):
    started = await client.login.start(redirect_uri=CALLBACK, return_to=returnTo)
    return RedirectResponse(started["url"], 302)

@app.get("/auth/callback")
async def callback(code: str = "", state: str = "", error: str = ""):
    if error:                                 # user cancelled at the IdP
        back = await client.login.cancel(state)
        return RedirectResponse(back["return_to"] if back else "/", 302)
    done = await client.login.complete(code=code, state=state,
                                       redirect_uri=CALLBACK)
    # done["user"] = {"xid", "name", "scopes"}; done["tokens"]; done["return_to"]
    ...  # your session, your cookie

• user["xid"] is read from the id_token — no /data lookup. Pass it as acting_as to act on the user's behalf.
• MemoryLoginStore is single process: behind a load balancer the callback can land on a worker that never saw /login (LOGIN_STATE_UNKNOWN). Back the store with what holds your sessions.
• No store configured → NO_LOGIN_STORE; no client_secret → LOGIN_REQUIRES_CONFIDENTIAL_CLIENT. A browser or native app is a public client and uses its own PKCE library.
• start(..., public_endpoint=) when the browser reaches the IdP on a different URL than this process does (containers, reverse proxies).
• The code exchange never retries: an authorization code is one-shot.

Code generation

Write your queries in .xsql files and get typed functions for them. The server compiles the queries, so the types always match what it returns.

1. Get a server. In your project folder:

synthigy env init
synthigy up

The first up prints a /setup link; open it and pick a database. (No browser? synthigy up --db sqlite skips the wizard.) Already have a server? Skip this step.

2. Deploy your data model in the modeler (or from code with synthigy.deploy(), as a client with the Dataset Developer role).

3. Connect as your app. Create its client once, then save it to the project:

synthigy iam add-client "My App" --id my-app --type confidential \
  --role "Dataset Explorer" --api Synthigy --grant client_credentials --local
synthigy connect http://localhost:7887 --client-id my-app

add-client prints the secret once; connect asks for it. Code is generated for what this app is allowed to see. (--local works on the server's own machine; for a remote server, create the client in the console.)

4. Install the SDK:

pip install synthigy

5. Write a query in xsql/movies.xsql:

@search list
movie (release_year > ?since:int=1990, limit ?limit:int=20)
  xid
  title
  release_year

6. Generate:

synthigy exec -- python -m synthigy.codegen gen xsql/ --out ops.py

This writes ops.py, and saves xsql/schema.json and xsql/ops.ir.json next to your queries.

7. Use it:

import synthigy
import ops

synthigy.connect()
movies = ops.Movie.list({"since": 2000})
print([m["title"] for m in movies])

synthigy exec -- python app.py

synthigy exec gives your program the server address and the app's identity. Without it, pass them yourself: synthigy.connect(endpoint, client_id=..., client_secret=...).

After you edit a query, run step 6 again. As long as the .xsql files are unchanged it works offline from xsql/ops.ir.json; after an edit it needs the server, and it never generates from outdated results. In CI:

python -m synthigy.codegen check xsql/

What to commit: your .xsql files and xsql/ops.ir.json. xsql/schema.json is your whole data model, so commit it only in a private repo.

@watch ops also get watch_<name>(), @batch ops become one-round-trip functions, every entity gets typed sync_<entity> / stack_<entity> / delete_<entity>, and each namespace has an ...Async twin for asyncio apps.

Tests

python3 -m unittest discover -s tests            # hermetic (stub server)
SYNTHIGY_TEST_CLIENT_ID=... SYNTHIGY_TEST_CLIENT_SECRET=... \
  python3 -m unittest tests.test_integration -v  # live (default localhost:7887)

Register a dedicated OAuth client for the live suite (trusted confidential, client_credentials) — never share identity with a live app. TestLiveLogin drives the real authorization-code flow headlessly and needs its own code-flow client plus a password user: SYNTHIGY_TEST_LOGIN_CLIENT_ID/_SECRET/_USER/_PASSWORD (setup snippet in tests/test_integration.py).

License

MIT — see LICENSE. The SDKs are permissive client libraries; the Synthigy engine is fair-code under the Sustainable Use License.