Metadata-Version: 2.4
Name: synapsor
Version: 0.1.3
Summary: Python SDK for Synapsor, the agent-native database for auditable AI applications
Author: Synapsor
License-Expression: MIT
Project-URL: Homepage, https://synapsor.ai
Project-URL: Documentation, https://synapsor.ai/docs
Project-URL: Source, https://github.com/synapsor/synapsor
Keywords: synapsor,database,agents,sql,rag,audit
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown

# 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

```python
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:

```bash
python -m pip install synapsor
```

```python
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.

```python
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:

```python
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:

```sql
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.

```python
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:

```python
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.
