persql 0.5.0

PerSQL Python SDK — SQLite databases on the edge for AI agents.

persql

Python SDK for PerSQL — SQLite databases on the edge for AI agents. One isolated database per agent, per app, per PR, per use case — backed by Cloudflare Durable Objects with embedded SQLite. Prepaid billing, four meters, no plans.

pip install persql

Quick start

import os
from persql import PerSQL

persql = PerSQL(token=os.environ["PERSQL_TOKEN"])
db = persql.database("acme/orders")

result = db.query("SELECT id, email FROM customers WHERE id = ?", [42])
for row in result["data"]:
    print(row["email"])

Async:

import asyncio
from persql import AsyncPerSQL

async def main():
    async with AsyncPerSQL(token=os.environ["PERSQL_TOKEN"]) as persql:
        db = persql.database("acme/orders")
        result = await db.query("SELECT 1 AS one")
        print(result["data"])

asyncio.run(main())

Give each user their own database ("Sign in with PerSQL")

Provision a private database in the user's own PerSQL account with the database OAuth scope — the user owns the data, you pay for it. The SDK does the OAuth 2.1 + PKCE handshake (sync shown; AsyncPerSQL mirrors it):

# 1. Kick off sign-in — open req["url"], keep req["code_verifier"] + req["state"].
req = PerSQL.begin_connect(
    client_id="psqlrp_…",
    redirect_uri="https://app.example.com/callback",
    scope="openid database",
)

# 2. After the redirect (verify state == req["state"]), exchange the code.
persql = PerSQL.complete_connect(
    client_id="psqlrp_…",
    redirect_uri="https://app.example.com/callback",
    code=code,                          # from the redirect
    code_verifier=req["code_verifier"],
)

# 3. Use the database provisioned in the user's account.
db = persql.database(persql.grant["database"])
db.query("CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY, body TEXT)")

Store persql.grant["access_token"] securely; on later runs construct PerSQL(token=..., base_url=...) directly. Confidential (server-side) clients pass client_secret=. See the User-owned app databases recipe.

Local mode (tests, no network)

persql = PerSQL(local=":memory:")
db = persql.database("test/db")
db.query("CREATE TABLE t (id INTEGER)")
db.query("INSERT INTO t (id) VALUES (?)", [1])
print(db.query("SELECT * FROM t")["data"])  # [{"id": 1}]

Local mode uses stdlib sqlite3 — no extra dependencies. Branches, approvals, and subscribe require a server-mode token (psql_live_… or psql_test_…) and raise in local mode.

Agent tools

Generate a typed tool bundle for any LLM — one tool per table plus discovery, safety, and branch-management tools — and let the model drive the database:

import anthropic
from persql import PerSQL

persql = PerSQL(token=os.environ["PERSQL_TOKEN"])
db = persql.database("acme/orders")
tools = db.as_tools()

client = anthropic.Anthropic()
reply = client.messages.create(
    model="claude-opus-4-7",
    max_tokens=4096,
    tools=tools["anthropic"],
    messages=[{"role": "user", "content": "How many customers signed up last week?"}],
)
for block in reply.content:
    if block.type == "tool_use":
        result = tools["run"](block.name, block.input)
        print(block.name, "→", result)

The same bundle exposes:

• tools["anthropic"] — pass directly to anthropic.messages.create
• tools["openai"] — pass to openai.chat.completions.create
• tools["langchain"] — convert with DynamicStructuredTool
• tools["run"](name, input) — sync or async dispatcher

Async clients return an awaitable bundle; await db.as_tools() then use await tools["run"](...).

Branches

Each branch is its own database, forked from the parent at create time. Idempotent by ref — call from CI with the PR number and the same ref re-runs as a reset:

branch = db.branches.upsert("pr-42", ttl_days=7)
preview = db.branches.preview("pr-42")
print(preview["plan"])  # added / changed / removed objects

Safety primitives

Pre-flight a write, get a single-use token, redeem only if the plan looks right:

plan = db.proposals.propose(
    "UPDATE orders SET status='shipped' WHERE created_at < ?",
    params=["2026-01-01"],
)
print(plan["estimated_affected_rows"])
db.proposals.apply(plan["execution_token"])

Approvals

When a write hits a require_approval rule, the SDK raises ApprovalRequiredError. Either halt and surface the URL to a human, or wait for the decision and redeem:

from persql import ApprovalRequiredError

try:
    db.query("DELETE FROM customers WHERE id = ?", [42])
except ApprovalRequiredError as e:
    print(f"Needs approval: {e.approval_url}")
    # …after a member approves in the console:
    db.approvals.redeem(e.approval_token)

Subscribe (async, optional)

Row-change events over WebSocket. Install the optional dep:

pip install 'persql[subscribe]'

async with AsyncPerSQL(token=...) as persql:
    db = persql.database("acme/orders")
    async for change in await db.subscribe(tables=["orders"]):
        print(change["table"], change["kind"])

Errors

from persql import ApprovalRequiredError, PerSQLError, RateLimitError

try:
    db.query("UPDATE orders SET status='shipped'")
except ApprovalRequiredError as e:
    pass  # human approval required — see e.approval_url
except RateLimitError as e:
    time.sleep(e.retry_after_seconds)
except PerSQLError as e:
    # `.detail["kind"]` is set on /v1/query SQL errors
    print(e.status, e.args[0], getattr(e, "detail", None))

Reference

Component	Description
Console	Manage tokens, databases, branches, billing
Docs	Concepts, API reference, examples
TS SDK	TypeScript equivalent of this package

License

MIT