Python SDK
Synchronous and asynchronous Python SDK for the Synapse API. Published as pyrx-synapse on PyPI. Single dependency (httpx).
Requires Python 3.10+.
Installation
bash
pip install pyrx-synapse
bash
poetry add pyrx-synapse
bash
uv add pyrx-synapse
Quick Start
Synchronous
python
from pyrx_synapse import Synapse
with Synapse(api_key="psk_live_your_api_key", workspace_id="your_workspace_id") as client:
# Track an event
client.track(
external_id="user_123",
event_name="purchase_completed",
attributes={
"order_id": "ord_456",
"amount": 99.99,
"currency": "USD",
},
)
# Identify a contact
client.identify(
external_id="user_123",
email="[email protected]",
first_name="Jane",
last_name="Doe",
properties={"plan": "pro", "signup_source": "website"},
tags=["paying", "beta-tester"],
)
# Send a transactional email
client.send(
template_slug="order-confirmation",
to={
"user_id": "user_123",
"email": "[email protected]",
"first_name": "Jane",
},
attributes={
"order_id": "ord_456",
"items": [{"name": "Widget", "price": 99.99}],
},
)
Asynchronous
python
from pyrx_synapse import AsyncSynapse
async with AsyncSynapse(api_key="psk_live_your_api_key", workspace_id="your_workspace_id") as client:
await client.track(
external_id="user_123",
event_name="purchase_completed",
attributes={"order_id": "ord_456", "amount": 99.99},
)
await client.identify(
external_id="user_123",
email="[email protected]",
first_name="Jane",
)
await client.send(
template_slug="order-confirmation",
to={"user_id": "user_123", "email": "[email protected]"},
attributes={"order_id": "ord_456"},
)
Tip
Get your API key and workspace ID from the dashboard at Settings > API Keys.
Configuration
python
client = Synapse(
api_key="psk_live_xxx", # Required. API key from workspace settings.
workspace_id="ws_xxx", # Required. Your workspace ID.
base_url="https://...", # Default: https://synapse-api.pyrx.tech
timeout=30.0, # Default: 30.0 seconds
max_retries=3, # Default: 3. Set to 0 to disable retries.
)
| Parameter | Type | Default | Description |
|---|---|---|---|
api_key | str | required | Your Synapse API key (psk_live_* or psk_test_*) |
workspace_id | str | required | Your workspace identifier |
base_url | str | https://synapse-api.pyrx.tech | API base URL |
timeout | float | 30.0 | Request timeout in seconds |
max_retries | int | 3 | Retry count for 429/5xx errors. Set to 0 to disable. |
Environment detection: The SDK detects test or live from your API key prefix (psk_test_* vs psk_live_*), available via client.environment.
Retry behavior: The SDK automatically retries on 429, 500, 502, 503, and 504 with exponential backoff and jitter (capped at 30s). On 429, uses the Retry-After header when present. Client errors (400, 401, 403, 404, 422) are never retried.
Track Events
Single Event
python
result = client.track(
external_id="user_123",
event_name="purchase_completed",
attributes={
"order_id": "ord_456",
"amount": 99.99,
"currency": "USD",
},
contact={
"email": "[email protected]",
"first_name": "Jane",
},
idempotency_key="purchase_ord_456", # optional, prevents duplicate processing
occurred_at="2026-04-29T10:30:00Z", # optional, defaults to now
)
print(result.event_id) # "evt_8f14e45f-..."
print(result.status) # "accepted"
| Parameter | Type | Required | Description |
|---|---|---|---|
external_id | str | Yes | Your unique user identifier |
event_name | str | Yes | Event name (e.g., purchase_completed) |
attributes | dict | No | Arbitrary key-value event data |
contact | dict | No | Contact fields to upsert alongside the event |
idempotency_key | str | No | Prevents duplicate processing (7-day TTL) |
occurred_at | str | No | ISO 8601 timestamp. Defaults to server time. |
Batch Events
Track up to 50 events in a single request.
python
result = client.track_batch(
events=[
{"external_id": "user_1", "event_name": "page_view", "attributes": {"page": "/pricing"}},
{"external_id": "user_2", "event_name": "page_view", "attributes": {"page": "/docs"}},
{"external_id": "user_1", "event_name": "button_clicked", "attributes": {"button": "upgrade"}},
],
)
print(result.accepted) # 3
print(result.rejected) # 0
Identify Contacts
Single Contact
Create or update (upsert) a contact by external_id.
python
contact = client.identify(
external_id="user_123",
email="[email protected]",
first_name="Jane",
last_name="Doe",
phone="+1234567890",
timezone="America/New_York",
locale="en-US",
properties={"plan": "pro", "signup_source": "website"},
tags=["paying", "beta-tester"],
)
print(contact.id) # UUID
print(contact.external_id) # "user_123"
print(contact.email) # "[email protected]"
Batch Identify
Upsert up to 1,000 contacts in a single request.
python
result = client.identify_batch(
contacts=[
{"external_id": "user_1", "email": "[email protected]", "first_name": "Alice"},
{"external_id": "user_2", "email": "[email protected]", "first_name": "Bob"},
],
on_conflict="merge", # "merge" | "skip" | "replace"
)
print(result.total) # 2
print(result.created) # 1
print(result.updated) # 1
Send Transactional Email
Send a one-off email using an NLT template, without a flow.
python
import time
result = client.send(
template_slug="otp-verification",
to={
"user_id": "user_123",
"email": "[email protected]",
"first_name": "Jane",
},
attributes={
"otp_code": "847293",
"expiry_minutes": 10,
},
idempotency_key=f"otp_user_123_{int(time.time())}",
)
print(result.status) # "sent" or "suppressed"
print(result.email_log_id) # "el_8f14e45f-..."
Note
Requires a data-scoped API key (or higher). The template must exist in your workspace.
Contact Management
The client.contacts sub-client provides full CRUD operations. Requires a management or full scoped API key.
List Contacts
python
result = client.contacts.list(
search="jane",
page=1,
per_page=25,
sort_by="created_at",
sort_order="desc",
)
print(result.meta.total) # 142
print(result.meta.total_pages) # 6
for c in result.data:
print(c.email, c.first_name)
Get a Contact
python
contact = client.contacts.get("contact_uuid")
Update a Contact
python
client.contacts.update("user_123", {
"email": "[email protected]",
"add_tags": ["vip"],
"remove_tags": ["trial"],
})
Delete a Contact
python
client.contacts.delete("user_123")
Template Management
The client.templates sub-client manages email templates. Requires a management or full scoped API key.
List Templates
python
templates = client.templates.list()
Get a Template
python
template = client.templates.get("welcome-email")
Create a Template
python
template = client.templates.create({
"name": "Welcome Email",
"slug": "welcome-email",
"subject": "Welcome, [first name of contact]!",
"body_html": "<h1>Welcome!</h1><p>Thanks for joining.</p>",
"sender_name": "PYRX Team",
"from_email": "[email protected]",
})
Update a Template
python
template = client.templates.update("welcome-email", {
"subject": "Welcome aboard, [first name of contact]!",
})
Preview with Sample Data
python
preview = client.templates.preview("welcome-email", {
"contact": {"first_name": "Jane", "email": "[email protected]"},
"trigger_event": {"order_id": "ord_123"},
})
print(preview.subject) # Rendered subject
print(preview.html) # Rendered HTML
print(preview.suppressed) # False
print(preview.suppressed_reason) # None
Delete a Template
python
client.templates.delete("old-template")
Error Handling
The SDK provides typed error classes for every failure mode.
python
from pyrx_synapse import (
SynapseError,
SynapseAuthError,
SynapseRateLimitError,
SynapsePlanLimitError,
SynapseValidationError,
)
try:
client.track(external_id="u1", event_name="test")
except SynapsePlanLimitError as e:
print(f"Plan limit: {e.limit_type} ({e.current}/{e.maximum})")
print(f"Current plan: {e.plan}")
except SynapseRateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after}s")
except SynapseValidationError as e:
for err in e.errors:
print(f"{err['field']}: {err['message']}")
except SynapseAuthError as e:
print(f"Authentication failed: {e.message}")
except SynapseError as e:
print(f"API error {e.status}: {e.message}")
Error Types
| Error Class | HTTP Status | Properties | When |
|---|---|---|---|
SynapseError | Any | status, message, code, request_id | Base class for all API errors |
SynapseAuthError | 401, 403 | message | Invalid or expired API key, scope mismatch |
SynapseValidationError | 422 | errors[] with field + message | Request body validation failed |
SynapseRateLimitError | 429 | retry_after (seconds) | Rate limit exceeded (auto-retried) |
SynapsePlanLimitError | 403 | limit_type, current, maximum, plan | Plan limit reached |
Environment Variables
For production deployments, load credentials from environment variables.
python
import os
from pyrx_synapse import Synapse
client = Synapse(
api_key=os.environ["SYNAPSE_API_KEY"],
workspace_id=os.environ["SYNAPSE_WORKSPACE_ID"],
)
bash
export SYNAPSE_API_KEY=psk_live_a1b2c3d4e5f67890abcdef1234567890
export SYNAPSE_WORKSPACE_ID=your_workspace_id
Full Method Reference
| Method | Description | Required Scope |
|---|---|---|
client.track(...) | Track a single event | data |
client.track_batch(...) | Track up to 50 events | data |
client.identify(...) | Upsert a single contact | data |
client.identify_batch(...) | Upsert up to 1,000 contacts | data |
client.send(...) | Send a transactional email | data |
client.contacts.list(...) | List contacts with pagination | management |
client.contacts.get(id) | Get a single contact | management |
client.contacts.update(id, data) | Update a contact | management |
client.contacts.delete(id) | Delete a contact | management |
client.templates.list() | List all templates | management |
client.templates.get(slug) | Get a template by slug | management |
client.templates.create(params) | Create a template | management |
client.templates.update(slug, params) | Update a template | management |
client.templates.preview(slug, data) | Preview rendered template | management |
client.templates.delete(slug) | Delete a template | management |
Framework Examples
FastAPI
python
import os
from fastapi import FastAPI, Request
from pyrx_synapse import AsyncSynapse
app = FastAPI()
synapse = AsyncSynapse(
api_key=os.environ["SYNAPSE_API_KEY"],
workspace_id=os.environ["SYNAPSE_WORKSPACE_ID"],
)
@app.post("/signup")
async def signup(request: Request):
data = await request.json()
# Identify the new user
await synapse.identify(
external_id=data["user_id"],
email=data["email"],
first_name=data.get("name", "").split(" ")[0],
properties={"plan": data.get("plan", "free")},
tags=["new-signup"],
)
# Track the signup event (triggers flows)
await synapse.track(
external_id=data["user_id"],
event_name="user_signed_up",
attributes={"plan": data.get("plan", "free"), "source": "web"},
)
return {"success": True}
Django
python
import os
from django.http import JsonResponse
from pyrx_synapse import Synapse
synapse = Synapse(
api_key=os.environ["SYNAPSE_API_KEY"],
workspace_id=os.environ["SYNAPSE_WORKSPACE_ID"],
)
def signup_view(request):
user = request.user
synapse.identify(
external_id=str(user.id),
email=user.email,
first_name=user.first_name,
)
synapse.track(
external_id=str(user.id),
event_name="user_signed_up",
attributes={"source": "web"},
)
return JsonResponse({"success": True})
Flask
python
import os
from flask import Flask, request, jsonify
from pyrx_synapse import Synapse
app = Flask(__name__)
synapse = Synapse(
api_key=os.environ["SYNAPSE_API_KEY"],
workspace_id=os.environ["SYNAPSE_WORKSPACE_ID"],
)
@app.route("/track", methods=["POST"])
def track():
data = request.get_json()
synapse.track(
external_id=data["user_id"],
event_name=data["event"],
attributes=data.get("properties", {}),
)
return jsonify({"status": "accepted"})
Context Managers
Both clients support context managers for automatic resource cleanup.
python
# Sync -- use "with"
with Synapse(api_key="psk_live_xxx", workspace_id="ws_xxx") as client:
client.track(external_id="user_123", event_name="test")
# httpx.Client is closed automatically
# Async -- use "async with"
async with AsyncSynapse(api_key="psk_live_xxx", workspace_id="ws_xxx") as client:
await client.track(external_id="user_123", event_name="test")
# httpx.AsyncClient is closed automatically
For long-lived processes (web servers), instantiate the client at module level and call client.close() (or await client.close()) on shutdown.
Related Resources
-
Node.js Server SDK -- server-side SDK for Node.js
-
Browser SDK -- client-side event tracking
-
Event Ingestion API -- raw REST API
-
API Reference -- interactive API explorer
-
Error Codes -- complete error code list