Skip to content
Ratify Protocol Ratify Protocol

Python SDK

The Python SDK is byte-for-byte interoperable with Go, TypeScript, Rust, and C/C++. Python 3.10+. Runtime dependencies are cryptography (Ed25519) and pqcrypto (ML-DSA-65) — both audited, both maintained by mainstream Python crypto teams.

Stability: every primitive on this page is stable in 1.0.0a13. There are no preview / experimental APIs in the snippets below.

Install

Section titled “Install”
Terminal window
pip install ratify-protocol==1.0.0a13

The package on PyPI is ratify-protocol. The module you import is ratify_protocol (PEP 8 naming).

Three minutes, end to end

Section titled “Three minutes, end to end”
import time
from ratify_protocol import (
    PROTOCOL_VERSION,
    SCOPE_MEETING_ATTEND, SCOPE_MEETING_SPEAK,
    DelegationCert, ProofBundle, VerifyOptions,
    generate_agent, generate_challenge, generate_human_root,
    issue_delegation, sign_challenge, verify_bundle,
)


# 1. Alice generates her hybrid (Ed25519 + ML-DSA-65) root identity.
alice, alice_priv = generate_human_root()


# 2. Her AI agent generates its own hybrid keypair.
agent, agent_priv = generate_agent("Alice's Scheduler", "custom")


# 3. Alice signs a delegation cert for the agent.
now = int(time.time())
cert = DelegationCert(
    cert_id="cert-001",
    version=PROTOCOL_VERSION,
    issuer_id=alice.id, issuer_pub_key=alice.public_key,
    subject_id=agent.id, subject_pub_key=agent.public_key,
    scope=[SCOPE_MEETING_ATTEND, SCOPE_MEETING_SPEAK],
    issued_at=now,
    expires_at=now + 7 * 24 * 3600,
    signature=None,  # populated in place by issue_delegation
)
issue_delegation(cert, alice_priv)


# 4. Verifier issues a challenge. Agent signs it.
challenge = generate_challenge()
challenge_sig = sign_challenge(challenge, now, agent_priv)


# 5. Agent assembles a proof bundle.
bundle = ProofBundle(
    agent_id=agent.id,
    agent_pub_key=agent.public_key,
    delegations=[cert],
    challenge=challenge,
    challenge_at=now,
    challenge_sig=challenge_sig,
)


# 6. Verifier runs the verifier in a single call.
result = verify_bundle(
    bundle,
    VerifyOptions(required_scope=SCOPE_MEETING_ATTEND),
)


if result.valid:
    print(f"✓ Authorized")
    print(f"  human_id:        {result.human_id}")
    print(f"  agent_id:        {result.agent_id}")
    print(f"  granted_scope:   {result.granted_scope}")
    print(f"  identity_status: {result.identity_status}")
else:
    print(f"✗ Rejected: {result.identity_status} — {result.error_reason}")

Real output (IDs are derived from the freshly generated keys, so they’ll differ on every run):

✓ Authorized
  human_id:        b3ef7456eb0a741008ee97cb30ee4aec
  agent_id:        98f087953d2da9d52e78f02300df380d
  granted_scope:   ['meeting:attend', 'meeting:speak']
  identity_status: authorized_agent

JSON transport — sending a bundle over the wire

Section titled “JSON transport — sending a bundle over the wire”

A proof bundle is JSON on the wire. Convert bytes fields to base64 before transport, then decode them back to SDK dataclasses before calling verify_bundle. The result verifies in Go, TypeScript, Rust, and C/C++ without changing the protocol bytes.

from fastapi import FastAPI, Request, HTTPException
import base64, json
from ratify_protocol import (
    Constraint, DelegationCert, HybridPublicKey, HybridSignature,
    ProofBundle, VerifyOptions, verify_bundle,
)


def b64e(value: bytes) -> str:
    return base64.b64encode(value).decode("ascii")


def b64d(value: str) -> bytes:
    return base64.b64decode(value.encode("ascii"))


def pub_to_json(key: HybridPublicKey) -> dict:
    return {"ed25519": b64e(key.ed25519), "ml_dsa_65": b64e(key.ml_dsa_65)}


def sig_to_json(sig: HybridSignature) -> dict:
    return {"ed25519": b64e(sig.ed25519), "ml_dsa_65": b64e(sig.ml_dsa_65)}


def proof_bundle_to_json(bundle: ProofBundle) -> str:
    return json.dumps({
        "agent_id": bundle.agent_id,
        "agent_pub_key": pub_to_json(bundle.agent_pub_key),
        "delegations": [{
            "cert_id": cert.cert_id,
            "version": cert.version,
            "issuer_id": cert.issuer_id,
            "issuer_pub_key": pub_to_json(cert.issuer_pub_key),
            "subject_id": cert.subject_id,
            "subject_pub_key": pub_to_json(cert.subject_pub_key),
            "scope": cert.scope,
            "constraints": [c.to_canonical_dict() for c in cert.constraints],
            "issued_at": cert.issued_at,
            "expires_at": cert.expires_at,
            "signature": sig_to_json(cert.signature),
        } for cert in bundle.delegations],
        "challenge": b64e(bundle.challenge),
        "challenge_at": bundle.challenge_at,
        "challenge_sig": sig_to_json(bundle.challenge_sig),
        "session_context": b64e(bundle.session_context) if bundle.session_context else "",
        "stream_id": b64e(bundle.stream_id) if bundle.stream_id else "",
        "stream_seq": bundle.stream_seq,
    })


def proof_bundle_from_json(raw_json: str) -> ProofBundle:
    raw = json.loads(raw_json)
    def pub(raw_key: dict) -> HybridPublicKey:
        return HybridPublicKey(ed25519=b64d(raw_key["ed25519"]), ml_dsa_65=b64d(raw_key["ml_dsa_65"]))
    def sig(raw_sig: dict) -> HybridSignature:
        return HybridSignature(ed25519=b64d(raw_sig["ed25519"]), ml_dsa_65=b64d(raw_sig["ml_dsa_65"]))
    return ProofBundle(
        agent_id=raw["agent_id"],
        agent_pub_key=pub(raw["agent_pub_key"]),
        delegations=[
            DelegationCert(
                cert_id=cert["cert_id"],
                version=cert["version"],
                issuer_id=cert["issuer_id"],
                issuer_pub_key=pub(cert["issuer_pub_key"]),
                subject_id=cert["subject_id"],
                subject_pub_key=pub(cert["subject_pub_key"]),
                scope=cert["scope"],
                constraints=[Constraint(**c) for c in cert.get("constraints", [])],
                issued_at=cert["issued_at"],
                expires_at=cert["expires_at"],
                signature=sig(cert["signature"]),
            )
            for cert in raw["delegations"]
        ],
        challenge=b64d(raw["challenge"]),
        challenge_at=raw["challenge_at"],
        challenge_sig=sig(raw["challenge_sig"]),
        session_context=b64d(raw["session_context"]) if raw.get("session_context") else b"",
        stream_id=b64d(raw["stream_id"]) if raw.get("stream_id") else b"",
        stream_seq=raw.get("stream_seq", 0),
    )


# Agent side: serialize for transport.
bundle_json = proof_bundle_to_json(bundle)
proof_header = base64.b64encode(bundle_json.encode("utf-8")).decode("ascii")
# Send proof_header as X-Ratify-Proof, an MQTT payload field, gRPC metadata, etc.


app = FastAPI()


@app.post("/tool/execute")
async def execute_tool(request: Request):
    raw_proof = request.headers.get("x-ratify-proof")
    if not raw_proof:
        raise HTTPException(401, "missing proof")


    try:
        bundle = proof_bundle_from_json(base64.b64decode(raw_proof).decode("utf-8"))
    except Exception:
        raise HTTPException(400, "malformed proof")


    result = verify_bundle(bundle, VerifyOptions(required_scope="execute:tool"))
    if not result.valid:
        raise HTTPException(403, {
            "error": result.identity_status,
            "reason": result.error_reason,
        })


    return {"ok": True, "agent_id": result.agent_id}

Standard convention: agents send the bundle JSON in the X-Ratify-Proof header (base64-encoded). See the API Gateway guide.

Verifier-side: branching on identity_status

Section titled “Verifier-side: branching on identity_status”

verify_bundle always returns a VerifyResult. Inspect result.valid first; if it’s false, the specific failure mode is in result.identity_status (a literal string, matching the byte-for-byte protocol-level enum across all five SDKs):

status = result.identity_status


if status == "authorized_agent":
    # ✓ All checks passed. result.granted_scope is the intersection of
    #   every cert's scope (effective scope across the chain).
    pass
elif status == "expired":
    # ✗ At least one cert is past its expires_at OR not-yet-valid.
    pass
elif status == "revoked":
    # ✗ A cert ID in the chain matched a revoked entry from the
    #   revocation provider (or the legacy is_revoked closure).
    pass
elif status == "scope_denied":
    # ✗ required_scope was not in the chain's effective (intersected)
    #   scope, OR an advanced PolicyProvider rejected the bundle.
    pass
elif status == "constraint_denied":
    # ✗ A first-class constraint (geo/time/amount/rate) evaluated to
    #   "outside the allowed range".
    pass
elif status == "constraint_unverifiable":
    # ✗ A constraint required input the caller didn't provide
    #   (e.g. cert has geo_circle but VerifierContext.current_lat was None).
    pass
elif status == "constraint_unknown":
    # ✗ A cert declared a constraint type this SDK build doesn't recognize.
    pass
elif status == "delegation_not_authorized":
    # ✗ An intermediate cert sub-delegated without identity:delegate.
    pass
elif status == "invalid":
    # ✗ Catch-all for structural / cryptographic failures.
    #   result.error_reason carries a stable machine-readable prefix
    #   (e.g. "stale_challenge: challenge is 31 seconds old (max 300)").
    pass

The verifier is fail-closed by default — any error path returns valid=False.

Constraints (geo / time / amount / rate / speed)

Section titled “Constraints (geo / time / amount / rate / speed)”

When a delegation declares constraints, supply the runtime context:

from ratify_protocol import VerifierContext


ctx = VerifierContext(
    current_lat=37.7749,
    current_lon=-122.4194,
)


result = verify_bundle(
    bundle,
    VerifyOptions(
        required_scope="drone:deliver",
        context=ctx,
    ),
)

Unset fields (None) on VerifierContext mean “not supplied” — any geo_circle / geo_polygon / geo_bbox constraint will return constraint_unverifiable if you don’t set both current_lat and current_lon.

The full constraint vocabulary (geo_circle, geo_polygon, geo_bbox, time_window, max_speed_mps, max_amount, max_rate) and VerifierContext field requirements are documented in Constraints.

Sub-delegation

Section titled “Sub-delegation”

Agent-to-agent delegation uses the same primitive. Alice grants Agent A meeting:attend plus the identity:delegate privilege; A then issues a sub-delegation to Agent B. Without identity:delegate on A’s grant from Alice, A’s sub-delegation will be rejected with delegation_not_authorized.

from ratify_protocol import SCOPE_IDENTITY_DELEGATE


# Alice → A: meeting:attend + identity:delegate
alice_to_a = DelegationCert(
    cert_id="cert-alice-to-a",
    version=PROTOCOL_VERSION,
    issuer_id=alice.id, issuer_pub_key=alice.public_key,
    subject_id=agent_a.id, subject_pub_key=agent_a.public_key,
    scope=[SCOPE_MEETING_ATTEND, SCOPE_IDENTITY_DELEGATE],
    issued_at=now, expires_at=now + 86400,
    signature=None,
)
issue_delegation(alice_to_a, alice_priv)


# A → B: a subset of A's grant
a_to_b = DelegationCert(
    cert_id="cert-a-to-b",
    version=PROTOCOL_VERSION,
    issuer_id=agent_a.id, issuer_pub_key=agent_a.public_key,
    subject_id=agent_b.id, subject_pub_key=agent_b.public_key,
    scope=[SCOPE_MEETING_ATTEND],
    issued_at=now, expires_at=now + 3600,
    signature=None,
)
issue_delegation(a_to_b, agent_a_priv)


# Chain order: [leaf, ..., root]. A→B is the leaf, Alice→A is the root.
bundle = ProofBundle(
    agent_id=agent_b.id, agent_pub_key=agent_b.public_key,
    delegations=[a_to_b, alice_to_a],
    challenge=challenge,
    challenge_at=challenge_at,
    challenge_sig=agent_b_sig,
)


result = verify_bundle(
    bundle,
    VerifyOptions(required_scope=SCOPE_MEETING_ATTEND),
)

result.granted_scope is the lex-sorted intersection of every cert in the chain. identity:delegate doesn’t propagate to B because A didn’t grant it onward — that’s the intentional behavior.

Provider hooks (SPEC §17)

Section titled “Provider hooks (SPEC §17)”

The provider hooks land on VerifyOptions as Optional[...] fields:

from ratify_protocol import (
    VerifyOptions,
    RevocationProvider, PolicyProvider, AuditProvider,
    ConstraintEvaluator, AnchorResolver,
)


result = verify_bundle(
    bundle,
    VerifyOptions(
        required_scope="meeting:attend",
        revocation=my_revocation_provider,        # §17.1
        policy=my_policy_provider,                # §17.2
        audit=my_audit_provider,                  # §17.3
        constraint_evaluators=my_ext_registry,    # §17.7
        anchor_resolver=my_anchor_resolver,       # §17.8
        policy_verdict=cached_verdict,            # §17.6
        policy_secret=policy_secret,
        context=ctx,
    ),
)

RevocationProvider, PolicyProvider, AuditProvider, ConstraintEvaluator, and AnchorResolver are typing.Protocol classes — duck-typing applies. Any class implementing the expected method signature works; no inheritance required.

Each hook has fail-closed semantics. See Provider architecture for the full reference.

Running the conformance suite

Section titled “Running the conformance suite”
Terminal window
cd ratify-protocol/sdks/python
pip install -e ".[dev]"
pytest -q

You should see all 63 canonical wire-format fixtures plus the 10 cross-SDK byte-equivalence vectors pass. If any fail, the bytes have drifted from the reference and the SDK is not interoperable — file a bug.

Where to next

Section titled “Where to next”