Python server SDK

Start with the supplied source

The Python package is capp_server. It contains a framework-neutral CappServer, route matching, gateway verification, receipt stores and a FastAPI adapter. This documentation does not assume a published package-registry release.

Download the Python server source. The archive contains the four selected SDK modules, not your application's backend. Extract capp_server/ next to your Python application, or use the same package from sdk/python/ in a cApp source checkout.

The supplied core imports cryptography for Ed25519 verification. Use an environment with that dependency installed. FastAPI is optional for the core; the example below also needs your FastAPI application stack. The source archive is not a dependency lockfile or a turnkey deployment.

Describe a backend contract

This example only describes a route. It does not start an HTTP server, authenticate a user or authorize an external effect.

from capp_server import CappServer

server = CappServer(
    server_id="example-catalogue",
    origin="https://api.example.com",
)
server.register_route(
    "/api/items",
    ["GET"],
    capability="backend.read",
    group="catalogue",
    risk="low",
    receipt=False,
)
contract = server.server_contract()
assert contract["protocol"] == "capp.server-contract.v1"

The constructor's gateway enforcement defaults to False. Treat this default as passive integration, not a production security setting. For protected routes, configure enforcement and the appropriate gateway trust and replay stores before performing the business action. A declared auth or approval label does not implement your application's user authentication.

Python and TypeScript are not identical adapters

The wire protocols are shared, but API names, return envelopes and some adapter behaviours differ. Python methods use snake_case; idempotent() returns a (replayed, value) tuple rather than TypeScript's {duplicate, value}. Python CappSyncAck needs to_json() and can include optional null fields; when validating against the selected schema, omit absent optional fields that are not nullable.

Core methods are synchronous. Do not pass an async business function to idempotent() expecting it to await a coroutine. Use your application's async transaction layer explicitly when that is required. The source download includes no package installer metadata, dependency lock, ASGI server or application database.

For the examples, create an isolated Python environment, install cryptography, and place the downloaded capp_server/ package beside your application. The FastAPI example additionally needs fastapi and an ASGI server such as uvicorn; those are your application dependencies, not files embedded in the download.

Core API

Use ctx.require_enabled() and ctx.require_gateway_if_needed() at the appropriate request boundary, before any protected action. Gateway checks depend on the enforcement configuration; a passive context is not verified authority. ctx.emit_receipt(result=..., response_body=..., ...) records a receipt and returns a CappReceipt.

The public Python exports also include CappRoutePolicy, GatewayAssertion, GatewayVerifier, CappInstallTrustRecord, trust/replay-store interfaces, receipt and idempotency interfaces, and route-normalization/matching helpers. They use Python snake_case names; the serialized contracts use their documented protocol field names.

Add FastAPI without replacing your application

FastAPICapp wraps CappServer. Register the framework route with FastAPI and put the cApp decorator underneath it, so FastAPI receives the wrapped handler. Include a Request argument: the wrapper needs the actual request to build a context.

from fastapi import FastAPI, Request
from capp_server.fastapi import FastAPICapp

app = FastAPI()
capp = FastAPICapp(
    server_id="example-catalogue",
    origin="https://api.example.com",
)

@app.get("/.well-known/capp/server-contract")
async def server_contract():
    return capp.server_contract()

@app.get("/api/items")
@capp.route(
    "/api/items", ["GET"],
    capability="backend.read", group="catalogue", risk="low",
)
async def read_items(request: Request):
    return {"items": []}

This is a read-only wiring example, not an authenticated service. The adapter offers context(request, path=None), server_contract(), binding_seed(...), receipt_json(receipt_id) and recent_receipts_json(limit=100). Receipt retrieval belongs behind your own authorization; do not make all receipts publicly browsable just because the helper exists.

The wrapper can emit a receipt after a successfully returned handler when receipt=True. You still own exception handling, business validation, user permissions and error-response mapping, including handling CappPolicyError. Do not treat a framework return value or an automatically emitted receipt as independent proof that an external service completed its action.

A fail-closed core request boundary

This complete helper requires enforcement, checks the actual path/method through registered routes and insists on exact body bytes for mutations. It does not authenticate a user or authorize a resource: run your own permission checks before the business operation and use the returned verified context.

from collections.abc import Mapping
from capp_server import CappServer, CappPolicyError

def protected_context(
    server: CappServer, *, path: str, method: str,
    headers: Mapping[str, str], body: bytes | None = None,
):
    if not server.enforce_gateway_assertions:
        raise CappPolicyError("Gateway enforcement is required", status_code=500)
    if method.upper() not in {"GET", "HEAD"} and body is None:
        raise CappPolicyError("Exact request body bytes are required", status_code=400)
    context = server.context(path, method, headers=headers, body=body)
    assertion = context.require_gateway_if_needed()
    if not assertion.verified:
        raise CappPolicyError("Verified installation required", status_code=401)
    return context

Configure gateway_trust_store, gateway_replay_store and the expected audience/app/site on the CappServer before using it. Never enrol an arbitrary public key as part of this request helper. A concrete resource permission check is still necessary for a pattern such as /api/notes/:id.

With FastAPICapp.route, include a real Request parameter and register the FastAPI decorator above the cApp decorator as shown. The current adapter skips context creation when it cannot find a Request; a directly invoked decorated function is therefore not a safe substitute for the HTTP boundary. The wrapper has already verified the request when it sets request.state.capp; do not verify the same replay-protected assertion a second time.

Add an explicit exception handler so a policy denial returns the intended HTTP status, not a generic server error. This example assumes app is the FastAPI instance defined above:

from fastapi import Request
from fastapi.responses import JSONResponse
from capp_server import CappPolicyError

def install_policy_error_handler(app):
    @app.exception_handler(CappPolicyError)
    async def policy_error(request: Request, error: CappPolicyError):
        return JSONResponse(
            {"ok": False, "error": "Request rejected", "protocol": "capp.policy-error.v1"},
            status_code=error.status_code,
        )

Call install_policy_error_handler(app) during application setup. Keep detailed diagnostics server-side and redacted. The adapter reads request bodies and may serialize a result for receipt hashing; impose your server's body/response limits and avoid using automatic receipt buffering for unbounded streams. Python's automatic receipt path labels a normally returned handler result ok; it does not independently inspect every framework response's business or HTTP outcome.

Return an acknowledgement, not a simulated mutation

This helper formats the acknowledgement after your application has committed a mutation. It does not perform, authorize or deduplicate that mutation. Supply the actual incoming operation identity and authoritative revision.

from capp_server import CappServer

def applied_note_ack(
    server: CappServer, operation_id: str, key: str, revision: int, value: dict,
) -> dict:
    if not operation_id or not key or type(revision) is not int or revision < 0:
        raise ValueError("A correlated operation, key and server revision are required")
    ack = server.sync_ack(
        operation_id=operation_id, status="applied", collection="notes",
        key=key, revision=revision, value=value,
    ).to_json()
    return {name: item for name, item in ack.items() if item is not None}

For conflicts return HTTP 409 with the intended conflict payload. For a duplicate return the stored acknowledgement for the same request, not the latest unrelated record. The sync schema validates the envelope but cannot establish that storage was committed.

First-service acceptance

Before accepting protected traffic, test the real framework request path: missing assertion rejected before the handler, valid enrolled install accepted, altered body and wrong app rejected, replay rejected, and user/resource checks enforced. Then test process restart and multiple workers with your chosen stores.

The core description example and public GET example do not meet that protected-service gate; they are deliberately passive wiring examples. The source archive and successful import are not deployment qualification.

Choose storage deliberately

MemoryReceiptStore, MemoryIdempotencyStore, MemoryGatewayTrustStore and MemoryGatewayReplayStore are process-local. Their state does not survive restart. FileReceiptStore(path) appends local receipt records; it is not a distributed database or transaction coordinator.

For valuable effects, use durable, atomic application-level claims and explicit reconciliation. The SDK helper's get-then-execute-then-put sequence cannot prevent two concurrent requests from executing the same external action. Redact receipt metadata and keep backend-only keys outside application packages.

Continue with server integration, route and data contracts and security boundaries.