ZAP Python SDK — pure-stdlib zero-copy wire codec, capability layer, RPC, and a decorator app.

Python Binding

zap-proto is the Python implementation of ZAP. The dist name on PyPI is zap-proto; the import is zap. Its core — the zero-copy wire codec (zap.wire), the capability runtime (zap.cap), the router envelope (zap.frame + ZapClient), the RPC client/server (zap.rpc), W3C DID identity, and agent consensus — is pure standard library: nothing third-party is on the import path of import zap.

The wire codec is byte-for-byte compatible with the canonical Go runtime (zap-proto/go) — the same 16-byte ZAP\x00 header and fixed-offset layout, so a buffer this library writes is read by Go and vice versa. Post-quantum crypto and the decorator app layer are optional extras.

Installation

pip install zap-proto
# or
uv add zap-proto

Optional extras:

pip install "zap-proto[crypto]"  # real ML-KEM-768 / ML-DSA-65 / X25519
pip install "zap-proto[app]"     # the FastMCP-style decorator app (pydantic)

Round-trip

Build a message, then read it back zero-copy. start_object lays out a struct's fixed section; parse validates the header and root() returns an Object that reads fields straight out of the buffer — no decode step.

from zap import Builder, parse

# Write: field @0 uint32, @8 text, @16 bytes.
b = Builder()
obj = b.start_object(24)             # 24-byte fixed section
obj.set_uint32(0, 0xDEADBEEF)
obj.set_text(8, "zap")
obj.set_bytes(16, b"\x01\x02\x03\x04")
obj.finish_as_root()
buf = b.finish()                     # bytes — the ZAP wire message

# Read: zero-copy. The Message aliases buf; nothing is deserialized.
root = parse(buf).root()
assert root.uint32(0) == 0xDEADBEEF
assert root.text(8) == "zap"
assert root.bytes(16) == b"\x01\x02\x03\x04"

Field offsets are explicit here to show the runtime; in real code you generate a typed View + Builder per struct from a .zap schema (see Code Generation). Lists (flat fixed-stride and variable-element) and nested objects are supported; the reader rejects out-of-bounds and backward pointers that would alias the wire header, and clamps list lengths — so an adversarial buffer that Go rejects is rejected here too.

RPC

zap.rpc is a real request/response transport over TCP. The Server dispatches the Zap interface method ordinals declared in the zap.zap schema (init @0 … log @8, exposed as the Method enum); the Client calls them.

import threading
from zap.rpc import Server, Client, Method

# Server: register a handler per Method ordinal, then serve.
srv = Server(host="127.0.0.1", port=0)
srv.handle(Method.CALL_TOOL, lambda params: {"content": f"hello {params['name']}"})
port = srv.bind()
threading.Thread(target=srv.serve_forever, daemon=True).start()

# Client: connect and call by ordinal.
c = Client.connect("127.0.0.1", port)
result = c.call(Method.CALL_TOOL, {"name": "world"})
assert result["content"] == "hello world"
c.close()
srv.close()

The wire body is JSON inside a magic-framed envelope (the zap.protocol framing, ZAP\x01 + type + length + JSON), so the RPC layer is debuggable without a codec.

Capabilities

zap.cap is the capability runtime — a faithful port of zap-proto/go/cap over the same capabilities.zap schema. A Cap is a signed, attenuable token of authority over a target, granted to a holder by an issuer. The wire, canonical-bytes, and CapID paths are pure stdlib; signing needs the [crypto] extra.

from zap import cap

# Mint a root cap. Ed25519Signer is the built-in mandatory scheme.
signer = cap.Ed25519Signer.generate()

target = bytes(range(32))            # 32-byte content hash of the resource
holder = bytes(range(31, -1, -1))    # 32-byte content hash of the holder
c = cap.issue(
    cap.Issuance(
        kind=int(cap.CapKind.IAM_SESSION),
        target=target,
        holder=holder,
        permissions=cap.PERM_ATTENUATE,  # bits the holder may exercise / delegate
        expires_at=2_000_000_000,        # unix seconds; 0 = never
    ),
    signer,
)

# Verify a single cap. The Verifier holds the policy: issuer-key resolution,
# revocation lookup, and (optionally) a PQ signature hook.
issuer = signer.public()
v = cap.Verifier(
    issuer_key=lambda h: signer.public_key_bytes() if h == issuer else None,
)
v.verify(c, now=1_700_000_000)       # raises ExpiredError / RevokedError / … on failure

Derive a narrower child with attenuate — the signer must hold the parent's holder key, permissions intersect with the parent's, expiry can only shrink, and the parent must carry PERM_ATTENUATE (or be a CapKind.DELEGATE cap):

child = cap.attenuate(c, child_holder, cap.PERM_AUDIT, None, 0, signer)

cap.id is SHA-256(canonical_bytes ‖ Sig) and canonical_bytes is the SPEC §3 signed scope — both byte-identical to the Go runtime, pinned by a cross-language known-answer test. verify_chain(leaf, chain, op, target, holder, now) validates a full chain; revoke / verify_revocation handle revocation. Schemes are Ed25519 (mandatory bootstrap), ML-DSA-65 (FIPS 204), and secp256k1 ECDSA — a missing crypto backend raises SchemeUnavailable (fail-closed, never a silent downgrade).

Decorator app

The [app] extra adds ZAP, a FastMCP-style decorator app that builds JSON Schemas from your function signatures and serves them over the real zap.rpc transport.

from zap import ZAP, PromptMessage

app = ZAP("my-agent", version="1.0.0")

@app.tool
def search(query: str, limit: int = 10) -> list[dict]:
    """Search for content in the knowledge base"""
    return [{"title": f"Result for {query}", "score": 0.95}]

@app.resource("file://{path}")
def read_file(path: str) -> str:
    """Read a file from disk"""
    return open(path).read()

@app.prompt
def greeting(name: str) -> list[PromptMessage]:
    """Generate a personalized greeting"""
    return [PromptMessage(role="assistant", content=f"Hello, {name}!")]

if __name__ == "__main__":
    app.run(port=9999)               # serves real ZAP RPC

The matching high-level client (Client, also from the [app] extra) speaks the same service:

from zap import Client

with Client("localhost:9999") as client:
    client.connect()
    tools = client.list_tools()
    result = client.call_tool("search", {"query": "hello world"})
    print(result.content)            # bytes
    content = client.read_resource("file:///tmp/test.txt")
    print(content.text)

Type hints

The package is fully typed and checked under mypy --strict. The wire, cap, and RPC surfaces import with no third-party dependency:

from zap import Builder, Message, Object, parse
from zap import cap
from zap.rpc import Server, Client, Method
from zap.identity import DID, DIDMethod

Develop

uv sync
ruff check
mypy --strict src/zap/
pytest

Examples

tests/test_wire.py — round-trips against a Go-produced golden vector
tests/testdata/cap_go_kat.json — a Go-signed cap that decodes, recomputes the same CapID, and verifies in Python

Python

On this page