synapsor 0.1.3

Python SDK for Synapsor, the agent-native database for auditable AI applications

Synapsor Python SDK

The Synapsor Python SDK connects applications to hosted or local Synapsor databases. Use it for SQL, branch workflows, agent contexts, capabilities, evidence, memory, and safe write proposals from Python applications.

Local Usage

from synapsor import Synapsor

db = Synapsor("app.db", auto_start=True)
db.set_session({
    "tenant_id": "acme",
    "principal": "support_agent_17",
    "session_id": "agent_run_1",
})

db.execute("CREATE TABLE messages (id INT, body VARCHAR);")
db.execute("INSERT INTO messages VALUES (1, 'hello');")
rows = db.query("SELECT * FROM messages;")

proposal = db.propose_memory_fact(
    scope=("tenant", "acme"),
    subject=("preference", "answer_style"),
    claim="The operator prefers concise answers.",
    source=("turn", "msg_1"),
    trust="verified",
    approval="approved",
    reason="stable preference extracted from chat",
)
db.approve_memory_proposal(proposal["proposal"]["proposal_id"], "operator approved")
memory = db.recall_memory(scope=("tenant", "acme"), subject=("preference", "answer_style"))

db.close()

Hosted Usage

Install from PyPI:

python -m pip install synapsor

import os
from synapsor import Synapsor

db = Synapsor("https://synapsor.ai", api_key=os.environ["SYNAPSOR_API_KEY"])
db.set_session({
    "tenant_id": "acme",
    "principal": "app_user_1",
    "session_id": "first_hosted_run",
})

print(db.query("SELECT 1;"))

ctx = db.invoke_agent_capability("chat.prepare_llm_context", {"question": "..."})
proposal = db.invoke_agent_capability(
    "support.propose_late_fee_waiver",
    {"fee_id": "FEE_3001", "reason": "card_update_failure"},
    mode="propose_only",
    auto_branch=True,
    response_envelope=True,
)

Use database-scoped API keys from the Synapsor control panel for hosted projects.

Agent Workflows

Use agent_runs when an agent framework owns routing, but Synapsor should own the durable workflow contract: session scope, capability calls, evidence, proposal branches, outbox actions, settlement, and replay.

run = db.agent_runs.start(
    workflow="billing.late_fee_waiver_flow",
    version="2026-05-27",
    input={"user_request": "Can we waive this late fee?"},
)

answer = run.invoke_capability(
    "support.answer_ticket_question",
    step_key="answer_ticket_question",
    arguments={"question": "Can we waive this late fee?"},
    response_envelope=True,
)

proposal = run.invoke_capability(
    "billing.propose_late_fee_waiver",
    step_key="propose_waiver",
    arguments={"amount_cents": 2500},
    mode="propose_only",
    auto_branch=True,
    response_envelope=True,
)

action = run.propose_external_action(
    "stripe.issue_refund",
    step_key="refund_customer",
    arguments={"charge_id": "ch_123", "amount_cents": 2500},
    idempotency_key="refund:TCK_1001:2500",
)

run.checkpoint(
    "before_worker_claim",
    payload={"reason": "external action queued"},
)
run.complete({"decision": "waiver_proposed"}, status="waiting_approval")
graph = run.explain()

Find the persisted run/evidence/proposal later by business object, workflow, tenant, query fingerprint, or time window:

activity = db.agent_activity.search(
    tenant_id="acme",
    business_object_id="T-1042",
    workflow="support.ticket_refund_flow",
    time_range={"from": 123, "to": 456},
    limit=50,
)

This helper sends the native lookup form:

SELECT *
FROM AGENT ACTIVITY
WHERE tenant_id = 'acme'
  AND business_object_id = 'T-1042'
ORDER BY created_at DESC
LIMIT 50;

Replay a stored capability run by using the numeric agent_run_id returned by Synapsor. Deterministic replay returns the captured persisted run; comparison modes can inspect the original snapshot, current state, a commit version, a timestamp, or a review branch.

replay = db.replay_agent_run(123, mode="original_snapshot")

branch_replay = db.replay_agent_run(
    123,
    mode="branch",
    branch_name="review_run_123",
)

Workers should claim and confirm side effects through the outbox namespace:

task = db.external_actions.claim(
    queue="billing_external_actions",
    worker_id="billing-worker-1",
)
db.external_actions.confirm(
    task["action_instance_id"],
    status="succeeded",
    provider_request_id="re_456",
    response={"status": "succeeded"},
)

API Surface

• execute(sql) and query(sql)
• set_session({...})
• invoke_agent_capability(name, arguments, mode=None, auto_branch=None, response_envelope=None, include_audit_trail=None, settlement_policy=None)
• agent_runs.start(...), run.invoke_capability(...), run.checkpoint(...), run.complete(...), run.explain(...)
• agent_activity.search(tenant_id=..., business_object_id=..., workflow=..., time_range=..., limit=...)
• replay_agent_run(id, mode=None, version=None, timestamp=None, branch_name=None)
• external_actions.claim(...) and external_actions.confirm(...)
• create_agent_eval(...) / evals.create(...) for run-history or source_table dataset evals
• evals.run(...).failures()
• list_capabilities(query="...")
• remember_fact(...)
• propose_memory_fact(...)
• approve_memory_proposal(id, reason)
• reject_memory_proposal(id, reason)
• list_memory_proposals(...)
• recall_memory(...)
• retire_fact(...) and forget_fact(...)
• check_fact_for_action(...)
• branch helpers: create_branch, use_branch, diff_branch, merge_branch, drop_branch
• write lifecycle helpers: preview_write, approve_write, commit_write, reject_write, settle_write
• read_resource(uri)

Errors from Synapsor become SynapsorError with status, code, request_id, retryable, and the raw payload. Include request_id when opening support or incident tickets so server, gateway, and runtime logs can be joined without exposing secrets.