Community Python SDK for SolvaPay — agentic payment rails.
pip install solvapay-python · Python 3.10+ · Fully typed (py.typed) · PyPI
@payable_tool stamps a function with payment metadata once. Any framework reads it — no per-framework wiring.
# Define once
from solvapay.adapters.mcp import payable_tool
@payable_tool(product="prd_search")
def web_search(*, customer_ref: str, query: str) -> list[str]:
"""Search the web."""
return do_real_search(query)# Run on MCP (Claude Desktop / Claude Code)
from solvapay.adapters.mcp import register_payable_tool_fastmcp
from fastmcp import FastMCP
mcp = FastMCP("My App")
register_payable_tool_fastmcp(mcp, web_search)
mcp.run()# Run on LangChain
from solvapay.adapters.langchain import monetize_tool
paid = monetize_tool(web_search, product="prd_search")
# Blocked callers get {"paywall_required": True, "checkout_url": "..."}# Run raw async
async with AsyncSolvaPay() as sv:
limits = await sv.limits.acheck(customer_ref="cus_123", product_ref="prd_search")
if limits.within_limits:
result = await web_search(customer_ref="cus_123", query="hello")pip install 'solvapay-python[mcp]' · pip install 'solvapay-python[langchain]'
pip install solvapay-python
export SOLVAPAY_SECRET_KEY=sk_...from solvapay import SolvaPay
sv = SolvaPay()
# Ensure customer exists
customer_ref = sv.customers.ensure("user_alice")
# Create checkout session
session = sv.checkout.create_session(customer_ref=customer_ref, product_ref="prd_0QKI8NHF")
print(session.checkout_url)
# Check limits and track usage
limits = sv.limits.check(customer_ref=customer_ref, product_ref="prd_0QKI8NHF")
if limits.within_limits:
sv.usage.track(
customer_ref=customer_ref,
product_ref="prd_0QKI8NHF",
meter_name="requests",
units=1.0,
idempotency_key="req_abc123", # idempotent — safe to retry
)| Namespace | Sync | Async |
|---|---|---|
sv.customers |
ensure, get, update, balance |
aensure, aget, aupdate, abalance |
sv.checkout |
create_session |
acreate_session |
sv.limits |
check |
acheck |
sv.purchases |
cancel, reactivate |
acancel, areactivate |
sv.usage |
track |
atrack |
sv.products |
list, get, create, delete, clone |
a prefix on each |
sv.plans |
list, create, update, delete |
a prefix on each |
sv.merchant |
get, get_platform_config |
aget, aget_platform_config |
All mutating ops accept idempotency_key. Use solvapay.idempotency.from_payload to derive deterministic keys:
from solvapay.idempotency import from_payload
# Default: key includes UTC date — rolls at midnight, bounds replay past server TTL
key = from_payload("track_usage", customer_ref, product_ref, "requests", units)
sv.usage.track(..., idempotency_key=key) # retry-safe
# Hourly bucket (high-frequency ops)
key = from_payload("charge", customer_ref, time_bucket="hour")
# Pure payload hash — caller manages TTL externally
key = from_payload("idempotent_op", ref, time_bucket=None)Retried POSTs must reuse the same key. A bucket roll (midnight / hour boundary) produces a new key — the server treats it as a new request.
from solvapay import (
AuthenticationError, PermissionError, NotFoundError,
RateLimitError, InvalidRequestError, APIServerError,
APIConnectionError, APITimeoutError, SolvaPayError,
)
try:
sv.customers.ensure("cus_123")
except AuthenticationError:
... # 401 — bad key
except RateLimitError as e:
... # 429 — retry after e.retry_after seconds
except APIConnectionError:
... # network failure
except APITimeoutError:
... # request timed out (default 30s)
except SolvaPayError as e:
... # catch-allNo built-in retries by default. solvapay[retry] ships RetryTransport — exponential backoff with jitter, 3 attempts, respects OpSpec.retry_safety (won't retry non-idempotent ops without an idempotency key). Or layer tenacity manually.
import os
from solvapay.webhooks import WebhookPipeline
pipeline = WebhookPipeline(
[os.environ["SOLVAPAY_WEBHOOK_SECRET"]],
max_clock_skew_seconds=300,
replay_ttl_seconds=600,
)
envelope = pipeline.process(body=request.body, signature=request.headers["sv-signature"])Secret rotation — pass multiple secrets; primary tried first, secondary on mismatch:
pipeline = WebhookPipeline(["whsec_new...", "whsec_old..."])Sign a webhook (testing / outbound fanout) — available at the top level since v0.9.2:
from solvapay import sign_webhook # top-level (v0.9.2+)
# from solvapay.webhooks import sign_webhook # subpackage path also works
header = sign_webhook(body=b'{"type":"purchase.created"}', secret="whsec_...")
# → "t=1716470000,v1=abc123..."ASGI adapter — mount directly in FastAPI / Starlette / Litestar:
from solvapay.adapters.asgi import webhook_app
app.mount("/webhook", webhook_app(pipeline, on_event=handle))Typed events — discriminated union over 13 event types:
from solvapay import WebhookEvent, PurchaseCreated, PaymentSucceeded
from pydantic import TypeAdapter
event = TypeAdapter(WebhookEvent).validate_python(envelope.event)
match event:
case PurchaseCreated():
print(f"New purchase: {event.data['purchaseRef']}")
case PaymentSucceeded():
print(f"Payment: {event.data['amount']}")from solvapay.paywall import require, PaywallRequired
@require(product="prd_0QKI8NHF", client=sv)
def run_query(*, customer_ref: str, query: str) -> str:
return expensive_query(query)
try:
result = run_query(customer_ref="cus_123", query="hello")
except PaywallRequired as e:
print(f"Upgrade at: {e.checkout_url}")
if e.checkout_mint_error:
print(f"Could not auto-mint checkout URL: {e.checkout_mint_error}")Async version: @require_async. For MCP/LangChain: @payable_tool (see above).
AsyncSolvaPay is the supported async pattern. Always use async with — it guarantees aclose():
from solvapay import AsyncSolvaPay
async with AsyncSolvaPay() as sv:
customer_ref = await sv.customers.aensure("user_alice")
limits = await sv.limits.acheck(customer_ref=customer_ref, product_ref="prd_0QKI8NHF")
if limits.within_limits:
await sv.usage.atrack(
customer_ref=customer_ref,
product_ref="prd_0QKI8NHF",
meter_name="requests",
units=1.0,
)Flat methods still work but emit DeprecationWarning. Removed in v2.0.
| v0.7.x | v0.8+ |
|---|---|
sv.ensure_customer(ref) |
sv.customers.ensure(ref) |
sv.get_customer(ref) |
sv.customers.get(ref) |
sv.check_limits(...) |
sv.limits.check(...) |
sv.track_usage(...) |
sv.usage.track(...) |
sv.create_checkout_session(...) |
sv.checkout.create_session(...) |
sv.cancel_purchase(ref) |
sv.purchases.cancel(ref) |
sv.reactivate_purchase(ref) |
sv.purchases.reactivate(ref) |
sv.get_customer_balance(ref) |
sv.customers.balance(ref) |
sv.list_products() |
sv.products.list() |
sv.create_plan(prd, ...) |
sv.plans.create(prd, ...) |
sv.get_merchant() |
sv.merchant.get() |
pip install solvapay-python # core
pip install 'solvapay-python[mcp]' # + FastMCP adapter (FastMCP ≥0.4)
pip install 'solvapay-python[langchain]' # + LangChain adapter (langchain-core ≥0.3)
pip install 'solvapay-python[fastapi]' # + FastAPI webhook router
pip install 'solvapay-python[asgi]' # + raw ASGI webhook adapter (no extra deps)
pip install 'solvapay-python[retry]' # + RetryTransport (tenacity)
pip install 'solvapay-python[bench]' # + pytest-benchmark (dev/perf testing)| Variable | Required | Default |
|---|---|---|
SOLVAPAY_SECRET_KEY |
Yes | — |
SOLVAPAY_API_BASE_URL |
No | https://api.solvapay.com |
SOLVAPAY_WEBHOOK_SECRET |
For webhooks | — |
| Path | What |
|---|---|
examples/multi-framework-paywall/ |
One @payable_tool → FastMCP + LangChain + raw async |
examples/marketplace/ |
Streamlit AI-agent marketplace — real SolvaPay sandbox + Gemini LLM |
examples/fastmcp-paywall/ |
FastMCP server gated by @paywall.require |
examples/langchain-paywall/ |
LangChain agent with monetize_tool |
Pin the API version your code was written against — prevents silent breakage when the server evolves:
sv = SolvaPay(api_version="2026-05-22") # sends Solvapay-Version header
sv = SolvaPay(api_version=None) # omit header (use server default)Default is "2026-05-22" (v0.9 ship date). Bump only on major SDK versions.
| Version | Theme |
|---|---|
| v0.8 ✅ | V1 architecture spine — Transport kernel, OpSpec registry, paywall/webhook packages, @payable_tool, stability manifest, layer DAG CI gate |
| v0.9 ✅ | Production polish — API-version pinning, idempotency TTL, RetryTransport, RecordingTransport, ASGI adapter, secret rotation, sign_webhook, contract tests, lint automation, MkDocs site, supply-chain hygiene |
| v0.9.1 ✅ | Security & supply-chain quality — PyPI attestations (PEP 740 / Sigstore), CycloneDX SBOM on releases, bandit + osv-scanner CI, Hypothesis-driven secret-redaction property tests, constant-time verify_webhook smoke test, explicit PCI-scope statement |
| v0.9.2 ✅ | Performance & stability quality — cold-import baseline harness, PEP 562 lazy adapter imports, solvapay.sign_webhook top-level re-export, microbenchmark harness (solvapay[bench]), troubleshooting docs |
| v0.9.3 ✅ | Layer DAG hard-enforcement — single type = layers import-linter contract replaces three sparse forbidden contracts; closes the L2 models purity gap and the L5 paywall/webhooks framework-neutrality gap; exhaustive = true flags any unlayered top-level module; ignore_type_checking_imports = true keeps cross-layer type hints lint-clean |
| v1.0 | Gated on founder signal — OpenAPI-generated models, WSGI/Lambda adapters, V2 planning |
v0.9.3 — Layer DAG hard-enforcement quality patch. tools/importlinter.cfg now ships a single type = layers contract covering the full HLD §V1.1 DAG (_transport → _http → models → operations → client → paywall|webhooks → adapters → fastapi|langchain), replacing three sparse forbidden contracts. exhaustive = true flags any new top-level module that ships without an explicit layer assignment; ignore_type_checking_imports = true strips TYPE_CHECKING blocks from the dependency graph so cross-layer type hints do not trip the gate. New tests/test_layer_dag.py shells the contract from pytest as a belt-and-suspenders gate. docs/architecture/layers.md refreshed. CI-only change; no public-API impact. 306 tests. 87% line coverage. mypy --strict clean (45 files).
v0.9.2 — Performance & stability quality patch. Cold-import baseline harness (tests/test_cold_import.py, 1.5x regression gate). PEP 562 lazy adapter imports — bare import solvapay no longer loads framework adapter modules. solvapay.sign_webhook promoted to top-level (alongside verify_webhook). Microbenchmark harness under tests/benchmarks/ (opt-in solvapay[bench]). New docs: docs/troubleshooting.md (top-10 errors) and docs/architecture/cold-start.md. 305 tests. 87% line coverage. mypy --strict clean (45 files).
v0.9.1 — Security & supply-chain quality patch. PyPI uploads now carry PEP 740 attestations (Sigstore-backed via OIDC trusted publishing); each GitHub release ships a CycloneDX SBOM (sbom.cdx.json). mypy --strict clean (45 files). 302 tests. 87% line / 85% branch coverage.
Supply chain & security posture:
- PyPI trusted publishing with PEP 740 attestations (Sigstore)
- CycloneDX 1.6 SBOM attached to every release
- CI runs
bandit -r src/ -lll,osv-scanner --lockfile=uv.lock, andpip-audit(cross-DB vuln check) on PR and nightly cron - Constant-time HMAC comparison (
hmac.compare_digest) regression-smoked - Hypothesis-driven property tests assert no API key, hex token, or webhook secret leaks into log output at any level
- Tightened upper bounds on volatile deps (
httpx<0.29,pydantic<2.14,langchain-core<1.5)
Community SDK, not official. Proposal: solvapay/solvapay-sdk#187.