Protocols and delivery#

These modules expose application capabilities over standard protocols and connect the application to other services without inventing another identity or schema layer.

wreath.mcp#

A first-party Model Context Protocol server, over the routes you already have.

An MCP server is a small protocol wrapped around several hard problems: a schema for every callable a model may invoke, a session that survives across calls, cancellation that actually stops work, an audit trail of what was called with what, and an authorization boundary in front of all of it. Wreath owns every one of those already -- the binding layer computes the schemas, the Flight Recorder keeps the record, Cedar decides -- so serving MCP here is wiring rather than a new stack.

Declare a server, decorate the callables a model should be able to reach, and the endpoint is an ordinary route:

from dataclasses import dataclass
from typing import Annotated

from wreath import Wreath
from wreath.binding import Body
from wreath.mcp import MCP, ToolError

app = Wreath()
mcp = MCP(app, name="camera-trap", version="1.0.0", path="/mcp")


@dataclass
class SightingQuery:
    species: str
    since: str | None = None


@mcp.tool(description="Find recent sightings of a species.")
async def find_sightings(request, query: Annotated[SightingQuery, Body()]) -> dict:
    if query.species not in KNOWN_SPECIES:
        raise ToolError(f"no camera is trained on {query.species!r}")
    return {"sightings": [...]}

The annotations that would bind an HTTP request body are the tool's inputSchema, rendered by the same code that renders the OpenAPI document. There is no second declaration syntax, no second validator, and a test asserts that the two renderings agree so they cannot drift apart later.

Point it at real work and it needs a boundary. Pass auth=MCPAuth(...) and the endpoint becomes an OAuth 2.1 resource server: it publishes /.well-known/oauth-protected-resource, answers a request with no token with a 401 naming that document, and refuses a token minted for a different resource -- the confused-deputy failure the specification exists to prevent. Gate individual tools with @mcp.tool(action=...), which is the same Cedar decision @authorize makes on a route, and bound them with rate_limit=ToolRateLimit(...). Every call leaves one Flight Recorder marker carrying the tool, the caller, the duration and the outcome, with the arguments beside it under the recording policy that already hides a password.

More than tools. @mcp.resource(uri, ...) declares something a model reads rather than does, and a client may resources/subscribe to it: open the GET stream and mcp.notify_resource_updated(uri) tells every subscriber it changed. @mcp.prompt(...) declares text a person chooses -- a slash command, a menu entry -- whose arguments are a flat map of strings, refused at registration if they are annotated as anything else. A long-running tool reports through request.state.mcp.progress, which is an ordinary wreath.progress.ProgressReporter: the server relays what it writes as notifications/progress, so there is no MCP-specific progress mechanism to learn, and a ProgressRegistry given the message bus carries a durable job's progress across workers to whichever one holds the stream.

Routes you already have. expose_routes(mcp, app, tags=("public",)) turns selected routes into tools. It takes an explicit selector and has no all=True: converting an application's whole HTTP surface into model-callable actions in one line includes the destructive half, and nobody reviews a line that short. A route with no docstring is refused, because the description is the entire basis on which a model decides to call it, and an exposed route keeps whatever @authorize, @roles, @permissions or @second_factor it was behind -- exposing a route is never a way around what was put in front of it. The last of those is the interesting one: a route behind @second_factor(max_age=300) is a tool the model may call only while the person has re-proved a factor recently, which is how "the model may read, and the human must step up before the model may delete" is written down. A tool that was never a route says the same thing with @mcp.tool(second_factor=300), so writing it does not require inventing a route to hang it on.

Asking the client something. Three MCP methods travel the other way, and a tool reaches all three through request.state.mcp. sample(...) asks the client's model to generate -- declared per tool with @mcp.tool(sampling=...), because whether a tool may spend the caller's model is a policy question, and gated, throttled and recorded exactly as the call itself is. elicit(message, Form) asks the person to fill a dataclass in: its fields are the requested schema, through the same derivation a tool's inputSchema comes from, and the answer is validated by the same binding layer and recorded under the same redaction. It is declared and gated the same way, with @mcp.tool(elicitation=...), because a form renders inside a client UI the person already trusts -- a tool asking for an API key is a phishing surface wearing that chrome, and "they can decline" is the control social engineering is built to walk through. read_file(path) reads beneath MCP(file_root=...) through wreath._fsguard, and beneath whatever roots the client declared -- a declared root is a boundary here, not a decoration. Each is a request the client must answer; a client that never does gets a bounded wait and then a ClientRequestError, never a session parked forever.

Locally, over stdio. wreath mcp stdio app:app puts the same ASGI application behind a stdin/stdout pipe for an editor that speaks only that transport. It is a byte relay over the in-process transport, not a second server: the routing, the authentication, the limits and the record are the ones the HTTP endpoint has.

What is not here. logging/setLevel, stream resumption by Last-Event-ID, templated resources and listChanged notifications. Each answers method not found and says which stage it is waiting for, rather than failing in a way a client author has to reverse-engineer.

Reference: MCP. Guide: Serving MCP tools.

MCP#

A Model Context Protocol server mounted on a Wreath application.

Declare it once, decorate the callables you want a model to be able to invoke, and the endpoint is a route:

app = Wreath() mcp = MCP(app, name="camera-trap", version="1.0.0")

@mcp.tool(description="Find recent sightings of a species.") async def find_sightings(request, query: Annotated[SightingQuery, Body()]) -> dict: ...

There is no second declaration syntax and no second validator. A tool's inputSchema is derived from the same annotations that would bind an HTTP request body, by the same code that renders the OpenAPI document, so the schema a model reads and the validation a call meets cannot disagree.

Pass auth=MCPAuth(...) to run it as an OAuth 2.1 resource server: the endpoint then publishes protected-resource metadata, refuses a request with no bearer token by pointing at that metadata, and refuses a token minted for a different resource -- which is the failure the specification exists to prevent. Without it the endpoint is exactly as protected as the route is, which for a bare Wreath() means not at all.

Args:

• app — The application (or Router) to mount on. Pass None to mount later
• with mount().
• name — The server name reported in initialize.
• version — The server version reported in initialize.
• path — The single endpoint path. POST carries every client message;
• DELETE ends a session.
• instructions — Optional guidance returned from initialize, describing
• how a model should use this server as a whole.
• auth — The OAuth 2.1 resource-server configuration, or None to install no
• authentication of this server's own.
• limits — The bounds this server runs inside. Payload size is deliberately
• not among them — it is RequestLimits.max_body_bytes on the app,
• where every other request's size is already decided.
• authorizer — The policy authorizer for Cedar-gated tools. Defaults to the
• one app.configure_auth(...) installed, which is what you want
• unless this server is mounted on a bare Router.
• progress — The registry a running tool reports progress to. A
• ProgressRegistry(bus) makes it fleet-wide.
• progress_interval — How often the relay polls that registry, in seconds.
• file_root — The one directory request.state.mcp.read_file(...) may read
• beneath, opened once as a trusted descriptor and walked component by
• component through wreath._fsguard -- the same confinement static
• files and the template loader use, refusing every symlink. Without
• it a tool cannot read a file at all, which is the right default for
• a surface a model drives. A client's declared roots narrow it
• further; they never widen it.

MCP(app: Any = None, *, name: str, version: str, path: str = '/mcp', instructions: str | None = None, auth: MCPAuth | None = None, limits: MCPLimits = MCPLimits(max_tools=256, max_sessions=1024, max_concurrent_calls=8, session_idle_seconds=900.0, max_resources=256, max_prompts=128, max_subscriptions=256, max_pending_notifications=64, stream_keepalive_seconds=15.0, max_pending_requests=4, client_request_seconds=30.0, max_file_bytes=1048576, max_result_bytes=1048576), authorizer: Any = None, progress: ProgressRegistry | None = None, progress_interval: float = 0.25, file_root: str | os.PathLike[str] | None = None) -> None

auth (property)#

auth: MCPAuth | None

counters#

counters() -> Any

This MCP server's counters, for wreath.metrics.collect.

declared_actions#

declared_actions() -> dict[str, tuple[str, ...]]

Resource type -> the Cedar actions this server is gated on.

The same shape wreath.authorization.declared_actions returns for an application's routes, and derived the same way -- from what is enforced, so there is no second list to drift. Read it to see, in one place, every action a model can reach through this server: tools, resources, prompts, and whatever expose_routes carried in from a route's own @authorize.

executor#

executor(*names: str, max_tools: int = 32) -> Any

expired_sessions (property)#

expired_sessions: int

Sessions collected for going idle since this server started.

limits (property)#

limits: MCPLimits

metadata_path (property)#

metadata_path: str

Where this server's protected-resource metadata is served.

metadata_url (property)#

metadata_url: str

The absolute metadata URL a 401 challenge names, or "".

mount#

mount(app: Any) -> None

Register this server's routes on app, which may be a Router.

name (property)#

name: str

notify_resource_updated#

notify_resource_updated(uri: str) -> int

Tell every subscriber that uri changed. Returns how many were told.

Synchronous and non-blocking on purpose: the caller is the code that just wrote the row, and a fan-out that could block would put a client's reading speed on the write path. A session with no open stream keeps the notification queued up to MCPLimits.max_pending_notifications; past that it is dropped and counted in notifications_dropped.

path (property)#

path: str

progress (property)#

progress: ProgressRegistry

The registry a running tool reports progress to.

prompt#

prompt(handler: Callable[..., Any] | None = None, *, name: str | None = None, title: str | None = None, description: str | None = None, action: str | None = None) -> Any

Declare a callable as a prompt. Usable bare or with arguments.

A prompt is chosen by a person -- a slash command, a menu entry -- rather than by the model, which is why it has arguments instead of a schema and why every one of them is a string. Return the text of one message, or a sequence of {"role": ..., "content": ...} mappings.

@mcp.prompt(description="Draft a report on a species' sightings.") async def sighting_report(request, species: str) -> str: return f"Summarise this month's {species} sightings."

Args:

• name — The prompt name. Defaults to the function's name.
• title — A human-facing display name.
• description — What this prompt is for. Defaults to the docstring.
• action — The Cedar action rendering it is gated on, with the prompt's
• own name as the resource.

Raises:

• TypeError — The handler is not an async function.
• ValueError — There is no description, the name is taken, or the
• server is at its max_prompts ceiling.
• ToolSignatureError — A parameter is annotated as something other than
• a string, or binds from a source prompts/get cannot fill.

prompts (property)#

prompts: tuple[Prompt, ...]

Every declared prompt, in the order prompts/list renders them.

resource#

resource(uri: str, *, name: str | None = None, title: str | None = None, description: str | None = None, mime_type: str | None = None, action: str | None = None) -> Any

Declare a reader as a resource, addressed by uri.

A resource is what a model reads rather than what it does, so the reader takes the request and nothing else -- a resources/read carries a URI and no arguments. Return text, bytes, or a value to render as JSON; bytes travel as a base64 blob and everything else as text.

@mcp.resource( "camera://ridge/latest", description="The most recent frame from the ridge camera.", mime_type="image/jpeg", ) async def ridge_latest(request) -> bytes: ...

Args:

• uri — The identifier a client reads and subscribes to.
• name — A short programmatic name. Defaults to the function's.
• title — A human-facing display name.
• description — What this holds. Defaults to the docstring, and one of
• the two must be present.
• mime_type — The media type reads are served as. Inferred from the
• reader's return value when omitted.
• action — The Cedar action a read is gated on. The **URI is the Cedar
• resource** — a resource has a stable identity, unlike a tool, so
• there is no second identifier to pass and nothing for the two to
• disagree about.

Raises:

• TypeError — The reader is not an async function, or takes arguments.
• ValueError — The URI is empty or taken, there is no description, or
• the server is at its max_resources ceiling.

resources (property)#

resources: tuple[Resource, ...]

Every declared resource, in the order resources/list renders them.

sessions (property)#

sessions: int

Live sessions right now, expired ones already collected.

stats#

stats() -> dict[str, int]

Every counter this server keeps, by name.

The same shape messaging.MessageBus.stats() returns, and for the same reason: read one attribute at a time, an exporter has to know each name and gains nothing when one is added. counters() layers this mapping onto the canonical metrics protocol; mounting on a Wreath application registers it for Prometheus, OpenMetrics, StatsD and CloudWatch.

tool#

tool(handler: Callable[..., Any] | None = None, *, name: str | None = None, title: str | None = None, description: str | None = None, action: str | None = None, resource: object | Callable[[Any], object] = None, rate_limit: ToolRateLimit | None = None, second_factor: float | None = None, sampling: str | bool | None = None, elicitation: str | bool | None = None) -> Any

Declare a callable as a tool. Usable bare or with arguments.

The handler's first parameter is the Request, exactly as a route handler's is; every later parameter is bound from the call's arguments object using its annotation. Reach for Annotated[Model, Body()] when a tool takes a structured argument, and plain annotated parameters for scalars.

Args:

• name — The tool name. Defaults to the function's name.
• title — A human-facing display name.
• description — What the tool does. Defaults to the docstring, and one
• of the two must be present.
• action — The Cedar action this tool is gated on, conventionally
• `Type — :verb`. Every call is then a policy decision made by the
• same authorizer, against the same entity shapes, as a route
• carrying @authorize(action=...) -- and a caller the policy
• refuses counts in unauthorized_calls, never tool_errors.
• resource — The Cedar resource, or a callable over the Request that
• produces one. Only meaningful with action=.
• rate_limit — A ToolRateLimit bounding how often one caller may
• invoke this tool. Keyed on the verified subject when the
• endpoint carries MCPAuth, and on the session otherwise.
• second_factor — Seconds within which the caller must have proved a
• second factor for this tool to run -- step-up, declared the way
• @second_factor(max_age=...) declares it on a route, and
• enforced by the same check against the same
• second_factor_at stamp wreath.users writes. Implies
• authentication. This is the natural way to say "this tool
• deletes things, so ask for the code again"; before it existed,
• step-up reached a tool only by exposing a route that already
• carried it. An identity with no stamp -- a bearer token, an
• OIDC login -- never satisfies it, so the default is closed.
• sampling — Whether this tool may call
• request.state.mcp.sample(...), which asks the client's model
• to generate. Pass the Cedar action a caller must be allowed for
• it -- which is where this belongs, because "may this tool spend
• the caller's model on text of its own choosing" is a policy
• question and not a code one -- or True to declare it with no
• policy. A tool that declares neither may not sample at all, and
• a sampling request also spends a token from this tool's own
• rate_limit= bucket.
• elicitation — Whether this tool may call
• request.state.mcp.elicit(...), which puts a form in front of
• the person at the other end. Declared, gated, throttled and
• recorded exactly as sampling= is, and off by default for a
• sharper reason — an elicitation renders **inside a client UI the
• user already trusts**, so a tool that asks for
• `{"api_key" — str}` is a phishing surface wearing the client's
• chrome. "The user is asked and can decline" is not the control
• it looks like -- a consent dialog is the defence social
• engineering is built to defeat -- so which tools may prompt a
• human is a deployment's decision. Pass the Cedar action, or
• True to declare it with no policy.

Raises:

• TypeError — The handler is not an async function.
• ValueError — The tool has no description, the name is taken, the
• server is at its max_tools ceiling, resource= was given
• without action=, or second_factor= is not positive.
• ToolSignatureError — The signature binds from a source that an MCP
• call cannot supply -- a header, a cookie, a form field, an
• upload, a path placeholder, or a dependency.

tools (property)#

tools: tuple[Tool, ...]

Every declared tool, in the order tools/list renders them.

version (property)#

version: str

ClientRequestError#

A server-to-client request could not be sent, or was not answered.

Raised inside the tool that asked, so it is caught with an ordinary try/except and answered with a ToolError the model can read, or left to the boundary -- where it counts in tool_errors and the model is told only the type, like any other unplanned failure.

The message always names which of the four endings happened: the client never advertised the capability, the session already has as many questions outstanding as it may, the client answered with an error, or nobody answered at all.

JsonRpcError#

A transport- or protocol-level failure, rendered as a JSON-RPC error.

This is deliberately not how a tool reports that it could not do its job: a model needs to see that failure as a result it can reason about, which is what wreath.mcp.ToolError produces. Reserve this for envelopes the server could not honour at all.

JsonRpcError(code: int, message: str, data: Any = None) -> None

MCPAuth#

What an MCP endpoint publishes about itself, and what it will accept.

Attributes: resource: This endpoint's canonical identifier -- the absolute URL a client asks the authorization server for a token for, and the value it must find in the token's aud. It should be the public URL of the MCP endpoint itself: https://api.example.com/mcp. authorization_servers: Issuer identifiers this endpoint trusts. Purely advisory: they tell a client where to go, and change nothing about what is accepted here, which is decided by verifier and audience. verifier: The token verifier -- a JwtVerifier, an OidcProvider's bearer_verifier(), or any callable taking the compact token and returning an Identity or None. Without one, every request is refused: an endpoint that publishes metadata and then accepts anything is worse than one with no metadata at all. audience: The value a token's aud must contain. Defaults to resource, which is what RFC 8707 asks an authorization server to put there; set it only when the deployment's tokens name the resource by some other identifier. scopes_supported: Scopes a client may usefully request. Advertised only. resource_name: A human-readable name for the resource, for a consent screen that is deciding what to say about it. resource_documentation: A URL a developer can read.

Raises:

• ValueError — resource is empty, or no authorization server is named.

MCPAuth(resource: str, authorization_servers: Sequence[str] = (), verifier: Verifier | None = None, audience: str | None = None, scopes_supported: Sequence[str] = (), resource_name: str | None = None, resource_documentation: str | None = None) -> None

authenticate#

async authenticate(request: Request) -> Identity

Verify the request's bearer token and bind it to this resource.

Raises:

• Unauthenticated — No token, an unverifiable one, or one minted for a
• different resource.

challenge#

challenge(metadata_url: str, *, error: str | None, description: str | None) -> bytes

The WWW-Authenticate value for a 401 from this endpoint.

Always carries resource_metadata=<metadata_url> (RFC 9728 §5.3), which is how a client that has never seen this server discovers which authorization server to get a token from — a 401 without it says a token is needed and nothing about where to obtain one.

Args:

• metadata_url — The absolute URL of this endpoint's
• protected-resource metadata document.
• error — The RFC 6750 error code, or None. None is meaningful
• rather than lazy — a request that carried no credentials at all
• gets a bare challenge, because there is no token for an error
• code to describe.
• description — Human-readable detail, or None to omit it.

document#

document() -> dict[str, Any]

The RFC 9728 protected-resource metadata for this endpoint.

expected_audience (property)#

expected_audience: str

The value a token's aud must contain to be accepted here.

audience when one was set, and resource otherwise — which is what RFC 8707 asks an authorization server to put there. This is the check that makes a token minted for another resource useless at this one, so it is never empty: resource is required at construction.

MCPLimits#

Bounds for one MCP server.

Attributes: max_tools: Tools one server may declare. Registration past it raises, at import time rather than at call time, because a tools/list a model cannot hold in its context is a design problem and not a runtime one. max_sessions: Concurrently live sessions. initialize is the one method that can be reached without a session, so before authentication is configured every POST that lands can mint one; a dictionary with no ceiling is then a memory leak with a network interface. max_concurrent_calls: Calls one session may have in flight at once. The bound is per session rather than per server so that one busy client cannot starve another, and it is small because a model that fans out is usually looping rather than working. session_idle_seconds: How long a session may go unused before it is collected. A client that never sends DELETE is the normal case, not the exception -- a process crashes, a laptop sleeps, a model stops mid-conversation -- so without this every abandoned session is held until restart and max_sessions is reached by attrition. Set it to None to keep sessions until they are ended explicitly. max_resources: Resources one server may declare, for the same reason max_tools exists: resources/list is text a model reads. max_prompts: Prompts one server may declare. max_subscriptions: URIs one session may hold a resources/subscribe on. A subscription costs a set entry and a fan-out visit per change, and a client is free to subscribe to every resource it can list, so the ceiling is what keeps that from being unbounded per session. max_pending_notifications: Notifications one session may hold while its server-to-client stream is closed or slow. The queue is bounded because a client that opens no stream must not be able to make the server buffer for it forever; past the ceiling the newest notification is dropped and counted in MCP.stats(), because a dropped notification that nobody counts is the silent degradation this codebase keeps finding. stream_keepalive_seconds: How long the notification stream may stay silent before it emits an SSE comment. Nothing in SSEResponse is emitted on a timer, deliberately, so an idle stream is invisible to an intermediary that reaps idle connections unless the application says something -- this is the application saying it. max_pending_requests: Server-to-client requests one session may have outstanding at once -- a sampling/createMessage, an elicitation/create, a roots/list. Each one holds a future and an entry in a table until the client answers it, so an unbounded table is a memory leak a client can drive simply by asking for work and never replying. Small on purpose: a tool that has three questions outstanding for one user is not waiting on a person any more. client_request_seconds: How long to wait for the client's answer to one of those requests. A client that never answers must not be able to pin a session's call slot for the session's whole idle life, which is the deadlock this number exists to make impossible. max_file_bytes: The largest file ToolContext.read_file will return. A resource read is a JSON-RPC result held whole in memory and then base64-encoded, so the ceiling is on what one answer may cost rather than on what the filesystem happens to hold. max_result_bytes: Largest serialized tool or resource result.

Raises:

• ValueError — A bound is not positive.

MCPLimits(max_tools: int = 256, max_sessions: int = 1024, max_concurrent_calls: int = 8, session_idle_seconds: float | None = 900.0, max_resources: int = 256, max_prompts: int = 128, max_subscriptions: int = 256, max_pending_notifications: int = 64, stream_keepalive_seconds: float = 15.0, max_pending_requests: int = 4, client_request_seconds: float = 30.0, max_file_bytes: int = 1048576, max_result_bytes: int = 1048576) -> None

Prompt#

One declared prompt: what prompts/list renders and prompts/get fills.

Attributes: name: The name a client offers to the person choosing it. description: What this prompt is for. Never empty. handler: The async callable, invoked as handler(request, **arguments). arguments: The prompts/list argument descriptors, derived once. title: An optional human-facing display name. binding_spec: The compiled signature, or None for a prompt that takes only the request. requirement: What a caller must satisfy to render it. completions: Argument name -> the values completion/complete offers for it, read off the rendered schema of an argument annotated as a Literal or an Enum. Derived rather than declared: a second place to write the candidate values down is a second place for them to go stale.

Prompt(name: str, description: str, handler: Callable[..., Any], arguments: tuple[dict[str, Any], ...] = (), title: str | None = None, binding_spec: BindingSpec | None = None, requirement: AuthRequirement = AuthRequirement(authenticated=False, identify=False, role_checks=(), permission_checks=(), policies=(), second_factor=None, oauth_step_up=None, public=False), completions: Mapping[str, tuple[str, ...]] = mappingproxy({})) -> None

bind#

bind(arguments: Mapping[str, Any]) -> dict[str, Any]

Validate arguments and return the handler's kwargs.

Raises:

• ValidationError — One or more arguments are missing, unknown, or of
• the wrong shape.

manifest#

manifest() -> dict[str, Any]

The prompts/list entry for this prompt.

Resource#

One declared resource: what resources/list renders and read serves.

Attributes: uri: The identifier a client reads, subscribes to, and sees again on every notification about it. Unique within one MCP. name: A short programmatic name. description: What this resource holds, and when a model should read it. Never empty, for the same reason a tool's is not. handler: The async callable, invoked as handler(request). title: An optional human-facing display name. mime_type: The media type reads are served as, or None to infer it from what the reader returned. requirement: What a caller must satisfy to read it -- the same AuthRequirement a tool or a route carries, decided by the same authorizer.

Resource(uri: str, name: str, description: str, handler: Callable[..., Any], title: str | None = None, mime_type: str | None = None, requirement: AuthRequirement = AuthRequirement(authenticated=False, identify=False, role_checks=(), permission_checks=(), policies=(), second_factor=None, oauth_step_up=None, public=False)) -> None

manifest#

manifest() -> dict[str, Any]

The resources/list entry for this resource.

render#

render(value: Any, *, _encoded: bytes | None = None) -> dict[str, Any]

One contents entry for whatever the reader returned.

bytes travel as a base64 blob and everything else as text, because those are the two shapes the specification defines and a client chooses between them by which key is present, not by the media type.

Tool#

One declared tool: what tools/list renders and tools/call invokes.

A tool is not a route, but its handler obeys the route contract -- the first parameter is the Request, every later parameter is bound by name -- so the same signature works in both places and the schema derivation is shared.

Attributes: name: The name a client calls. Unique within one MCP. description: What the tool does, and when a model should reach for it. Never empty: a tool without a description is a tool that gets misused, and registration refuses one. handler: The async callable, invoked as handler(request, **arguments). input_schema: JSON Schema for the arguments object. title: An optional human-facing display name. binding_spec: The compiled signature, or None for a handler that takes only the request. requirement: What a caller must satisfy to invoke it -- exactly the AuthRequirement a route carries, so the authorizer cannot tell an MCP call from an HTTP one and expose_routes can hand a route's own requirement straight through rather than translating it into something weaker. NO_REQUIREMENT for a tool anyone who reached the endpoint may call. rate_limit: The declared per-caller ceiling, or None. limiter: The token bucket enforcing rate_limit. One per tool, because two tools sharing a bucket would spend each other's allowance. route: The path of the route this tool was derived from, or None for a tool declared with @mcp.tool. Carried so a refusal can name the route rather than only the tool it would have become. sampling_requirement: What a caller must satisfy for this tool to ask the client's model to generate, or None when the tool did not declare sampling= at all -- which is the default, and means it may not sample. Whether a given tool may spend the caller's model on text of its own choosing is exactly a policy question, so it is one more AuthRequirement decided by the same authorizer as everything else here; sampling=True declares the capability with no policy attached, which is the shape a development server wants. elicitation_requirement: The same, for putting a form in front of the person at the other end, and None by default for the same reason. An elicitation renders inside a client UI the user already trusts, so a tool that asks for {"api_key": str} is a phishing surface wearing the client's chrome; "the user can decline" is precisely the control social engineering defeats. Which tools may prompt a human is therefore a deployment's decision, decided by the same authorizer.

Tool(name: str, description: str, handler: Callable[..., Any], input_schema: dict[str, Any], title: str | None = None, binding_spec: BindingSpec | None = None, requirement: AuthRequirement = AuthRequirement(authenticated=False, identify=False, role_checks=(), permission_checks=(), policies=(), second_factor=None, oauth_step_up=None, public=False), rate_limit: ToolRateLimit | None = None, limiter: MemoryRateLimitStore | None = None, route: str | None = None, sampling_requirement: AuthRequirement | None = None, elicitation_requirement: AuthRequirement | None = None) -> None

bind#

bind(arguments: Mapping[str, Any]) -> dict[str, Any]

Validate arguments against the signature and return call kwargs.

Raises:

• ValidationError — One or more arguments are missing, unknown, or of
• the wrong shape.

manifest#

manifest() -> dict[str, Any]

The tools/list entry for this tool.

ToolContext#

What a running tool can learn about the call it is serving.

Reachable from the handler as request.state.mcp, which keeps the ownership visible: it lives exactly as long as the request that carries it, and no global or context variable has to be consulted to find it.

Cancellation arrives as asyncio.CancelledError inside the handler, because the call runs in its own task and notifications/cancelled cancels that task. A tool that holds a resource should therefore release it in a finally:, exactly as it would anywhere else; there is no cancellation flag to poll.

It is also the way out of this call and back to the client: sample, elicit, roots and read_file are the three MCP methods that travel server-to-client, and they are methods here rather than free functions because each needs the session the call arrived on and the request it is being served under. The same task that makes cancellation work is what makes them possible: the client's answer arrives on a different POST, which the endpoint can only serve because this one is parked in a task of its own.

Attributes: session_id: The Mcp-Session-Id this call arrived on. request_id: The JSON-RPC id, and what a cancellation would name. tool: The tool's registered name. progress_token: The client's _meta.progressToken, when it sent one. Present so a tool can tell whether anyone is listening; reporting progress does not depend on it. progress: A wreath.progress.ProgressReporter bound to this call. Always present, whether or not the client asked for progress: reporter.update(42, "processing invoices") writes to the server's ProgressRegistry exactly as it would from a durable job, and when the client did send a progressToken the server relays each report as a notifications/progress on the session's stream. There is no MCP-specific progress mechanism, deliberately -- the registry already models a task reporting a percentage and a message, already spans workers when it is given the message bus, and already has a status endpoint and an SSE stream of its own. identity: The verified caller, whenever the server had to resolve one -- through MCPAuth on the endpoint, or through the application's own app.configure_auth(...) backend for a tool that is gated or rate limited. None for a tool that is neither, on an endpoint with no MCPAuth: nothing about that call depended on who was asking, so the backend was never run for it. Also published as request.identity, which is where the authorizer and every other Wreath component reads it; it is repeated here so a tool does not have to know which of the two is the real one. arguments: The call's raw arguments object, before binding. Present so a Cedar resource= resolver can name which row is being asked for: a route resolves its resource from the path, and a tool has no path to resolve it from. The bound, validated values arrive as the handler's own parameters; this is for the policy decision that has to happen first.

ToolContext(session_id: str, request_id: Any, tool: str, progress_token: Any = None, identity: Any = None, arguments: Mapping[str, Any] = mappingproxy({}), progress: ProgressReporter | None = None, _server: Any = None, _session: Any = None, _request: Any = None) -> None

elicit#

async elicit(message: str, form: type[Any]) -> Any | None

Ask the person at the other end to fill form in.

form is a dataclass, and its fields are the requested schema: they go through derive_input_schema, the same path that renders a tool's inputSchema, and the answer comes back through the same binding validation. There is no second schema language and no hand-written check, which is why a field the client fills in wrongly is refused with the same error shape a bad tools/call argument gets.

The tool doing the asking must have declared that it may: @mcp.tool(elicitation="Form::ask") gates it on Cedar exactly as action= gates the call itself, and elicitation=True declares it with no policy. A tool that declared neither cannot elicit at all, because a form renders inside a client UI the person already trusts and "they can decline" is the control social engineering exists to defeat. The request also spends a token from the tool's own rate_limit= bucket and leaves its own Flight Recorder marker beside the call's.

Returns:

• An instance of form, or None when the person declined or cancelled.

Raises:

• ClientRequestError — The client never advertised the elicitation
• capability, the tool did not declare elicitation=, the caller
• was refused or throttled, nobody answered, or the answer did not
• match the schema that was asked for.
• TypeError — form is not a dataclass, or has a field MCP cannot
• carry -- the specification allows primitives only.

read_file#

async read_file(path: str) -> bytes

Read a file beneath the server's file_root, and beneath the client's roots.

Two confinements, both real. The bytes are opened through wreath._fsguard -- component by component beneath a trusted directory descriptor, refusing every symlink, the same walk static files and the template loader use -- so nothing outside MCP(file_root=...) is reachable however the path is spelled. And when the client declared roots, the file must also lie beneath one of them: a root a client declared is a boundary it expects the server to honour, not a hint.

Raises:

• FileNotFoundError — No such file beneath the root.
• PermissionError — The path escapes the root, traverses a symlink, is
• not a regular file, is larger than MCPLimits.max_file_bytes,
• or lies outside every root the client declared.
• RuntimeError — This server was declared with no file_root=.

roots#

async roots() -> tuple[str, ...]

The filesystem roots this client declared, as absolute paths.

Asked of the client once per session with roots/list and cached until it sends notifications/roots/list_changed. Empty when the client did not advertise the capability, which is the case read_file treats as "the server's own root is the only bound".

sample#

async sample(messages: Any, *, max_tokens: int = 1024, system_prompt: str | None = None, temperature: float | None = None, stop_sequences: Any = None, model_preferences: Any = None, include_context: str | None = None, metadata: Any = None) -> dict[str, Any]

Ask the client's model to generate, and wait for what it produced.

sampling/createMessage is the one MCP method where the server is the one asking for work, and the tool doing the asking must have declared that it may: @mcp.tool(sampling="Model::sample") gates it on Cedar exactly as action= gates the call itself, and sampling=True declares it with no policy. A tool that declared neither cannot sample at all. The request also spends a token from the tool's own rate_limit= bucket, and leaves its own Flight Recorder marker beside the call's.

messages is a string -- the common case, one user turn -- or a list of {"role": ..., "content": ...} mappings already in MCP's shape.

Returns:

• The client's result — role, content, model and stopReason.

Raises:

• ClientRequestError — The client never advertised the sampling
• capability, the tool did not declare sampling=, the caller was
• refused or throttled, or nobody answered.

ToolError#

A tool could not do its job, reported to the model rather than the wire.

This is the difference that makes an MCP server usable. A JSON-RPC error tells the client that the call failed, which a model can only retry blindly; a ToolError comes back as an ordinary result carrying isError: true and your message, which a model can read and act on. Raise it for "no sighting matched that species", not for "the database is down".

raise ToolError("no camera covers that trail; try 'ridge' or 'creek'")

ToolRateLimit#

How often one caller may invoke one tool.

Declared per tool rather than per endpoint, because the tools on one server are not alike: listing yesterday's sightings is cheap and idempotent, and the one that sends an email is neither. A single ceiling across the endpoint would have to be set for the expensive tool and would then make the cheap one useless.

The bucket is keyed on the caller -- so one client cannot spend another's allowance. Declaring a ceiling is itself enough to resolve who the caller is: a bounded tool authenticates the request before it charges the bucket, through MCPAuth when the endpoint carries one and through the application's own app.configure_auth(...) backend when it does not. Only when neither identifies anybody does the key fall back to the session -- and a session is free, so on an endpoint with no authentication at all a per-session ceiling is a ceiling per initialize, which is to say none.

Attributes: limit: Calls allowed per window. window: The window, in seconds. burst: Calls allowed back to back before the sustained rate binds. Defaults to limit. A model that fans out a plan into five parallel calls is not abusing anything, and a burst of one would refuse it.

Raises:

• ValueError — A bound is not positive. A burst below limit is allowed
• and simply means the window's full allowance is never reachable in
• one go, which is a legitimate shape rather than a mistake.

ToolRateLimit(limit: int, window: float = 60.0, burst: int | None = None) -> None

capacity (property)#

capacity: float

rate (property)#

rate: float

Tokens per second, the shape RateLimitStore.configure wants.

ToolSignatureError#

A handler's signature cannot be expressed as an MCP tool.

expose_routes#

expose_routes(mcp: Any, app: Any = None, *, tags: Iterable[str] = (), include: Iterable[str] = (), predicate: Callable[[Any], bool] | None = None, prefix: str = '', rate_limit: ToolRateLimit | None = None) -> tuple[Tool, ...]

Declare selected existing routes as MCP tools.

Call it after the routes exist: an application is read here, not retained, so a route registered afterwards is not picked up.

expose_routes(mcp, app, tags=("sightings",))

A route is selected when it matches any selector given, and at least one selector must be given -- there is deliberately no way to say "all of them". A selector that matches nothing raises rather than exposing nothing quietly, because a tag that no longer exists is a typo far more often than it is an intention.

The tool's name is the route's operation_id when it has one and the handler's name otherwise, its description is the handler's docstring, and its inputSchema is derived from the signature by the same code that derives a declared tool's. One tool per route, not per method: a tool invokes the handler directly, so the HTTP verb the route answers on is not part of the call.

Args:

• mcp — The MCP server to declare the tools on.
• app — The application or Router to read routes from. Defaults to the
• one mcp is mounted on.
• tags — Expose every route carrying any of these tags.
• include — Expose every route whose path is named here, exactly.
• predicate — Expose every route this returns true for. Receives the
• `RouteDefinition, so lambda route — "GET" in route.methods` is how
• you keep the mutating half of a tag out.
• prefix — Prepended to each derived tool name, for a server that exposes
• two applications and would otherwise collide on list_items.
• rate_limit — A ToolRateLimit applied to every tool this call declares.
• Route-derived tools are the ones most likely to want it — a route was
• written for a person clicking, and a model does not click.

Returns:

• The tools that were declared, in the order the routes were registered.

Raises:

• ValueError — No selector was given, no route matched, a selected route
• has no docstring, a selected route carries route middleware or route
• dependencies (see _check_carryable), or a derived name collides
• with a declared tool.
• ToolSignatureError — A selected route binds a parameter an MCP call
• cannot supply. The message names the route and the parameter.

wreath.graphql#

GraphQL over the Wreath ORM, sharing everything the rest of the app uses.

Point it at a registry and it derives the schema from your models -- the same ModelSpec the SQL compiler, OpenAPI, and typegen read, so the GraphQL surface cannot drift from the REST one:

from wreath.graphql import GraphQL

api = GraphQL(app.orm("main"), models=[User, Post])
app.include_router(api.router(public=True))

What owning this in-tree buys, none of which a bolt-on library can:

• No N+1 and no DataLoader. A relationship selection resolves through the session's batched select-in loader -- one statement per relationship per level, deduplicated by the identity map.
• One authorization language. Field access is checked against the authorizer you pass, with Type.field as the resource, so one Cedar policy set covers REST and GraphQL. The authorizer is asked the way a route asks it -- a PolicyRequirement whose resource is the field as a Cedar entity reference, User::"email" -- so the shipped CedarAuthorizer needs no adapter. Hand it the same object you handed configure_auth; there is deliberately no pickup from the application.
• Per-field latency in the Flight Recorder, as RESOLVER phases, without wiring an exporter.
• Typegen. wreath typegen emits GraphQL operations alongside the REST ones, sharing one TypeScript model set, so a client gets useGetUser() and useUserQuery() returning the same User.

Safety is not optional here. A public GraphQL endpoint is a denial-of-service surface, so depth, complexity, alias, and parse-step limits are enforced while parsing and cannot be turned off -- only widened. Introspection is off by default, because a schema dump is reconnaissance.

ExecutionError#

A document that parsed but cannot run against this schema.

ExecutionError(message: str, *, path: tuple[str, ...] = ()) -> None

GraphQL#

A GraphQL endpoint over one ORM registry.

GraphQL(registry: Any, *, models: list[Any] | None = None, dataclasses: Iterable[type] = (), expose: Iterable[str] = (), limits: Limits | None = None, authorizer: Any = None, action: str = 'read', introspection: bool = False, max_page_size: int = 100, cache_size: int = 512, on_denied: str = 'error') -> None

declared_actions#

declared_actions() -> dict[str, tuple[str, ...]]

Resource type -> the Cedar actions this endpoint is gated on.

The same shape wreath.authorization.declared_actions returns for an application's routes, and read off the same declarations enforcement uses -- every field's policy resource, resolved through policy_resource, grouped by its entity type.

GraphQL's half of the shared vocabulary is the mirror image of REST's. A route names one action over a resource type whose rows it cannot enumerate; a GraphQL endpoint names one action -- the action= it was constructed with -- over resources it can enumerate exactly, because a field is declared rather than stored. So the manifest's type-level answer is the coarse one here, and POST {prefix} with the field names as ids is the exact one: it asks authorize(action, User::"email"), which is character for character the decision the endpoint takes.

Empty when no authorizer was given, because then nothing is enforced and a vocabulary entry would advertise a control that does not exist.

field#

field(type_name: str, field_name: str, *, returns: str = 'String', is_list: bool = False, non_null: bool = False, requires: tuple[str, ...] | list[str] = (), batch: bool = True, policy: str | None = None, cost: int = 1)

Register a computed field on type_name.

The resolver is batched by default: it receives the whole level as a list and returns one value per object. Per-parent resolvers are how application code reintroduces the N+1 the data layer just solved, so the batched form is the easy one to write:

@api.field("User", "postCount", returns="Int", requires=["posts"])
async def post_count(users, info):
    return [len(user.posts) for user in users]

requires names sibling fields that must be resolved first. They are resolved in batch, and if the client did not select them they stay hidden -- asking for a computed field never widens the response.

Pass batch=False for the one-object-at-a-time form when the work is genuinely per-object and cannot be batched.

mutation#

mutation(name: str, *, returns: str, is_list: bool = False, policy: str | None = None, cost: int = 10)

Register a mutation.

Mutations are separate from queries on purpose: they are the write surface, and giving them their own namespace means an authorization policy can cover all writes (Mutation.*) without enumerating them.

parse#

parse(source: str) -> Any

Parse and cache source under the configured limits.

Documents past MAX_CACHED_QUERY_CHARS are parsed and not cached: the key is the client's own text, so caching by entry count alone let a few large documents hold far more memory than the count suggested, and a caller choosing the key is a caller choosing the size.

query#

query(name: str, *, returns: str, is_list: bool = False, policy: str | None = None, cost: int = 10)

Register a custom root query field.

Needs no backing table: use it for a search, an aggregate, or anything assembled from more than one model:

@api.query("search", returns="User", is_list=True)
async def search(info):
    return await info.session.fetch(
        User.select().where(User.email.like(info.arguments["term"]))
    )

router#

router(path: str = '/graphql', *, session_factory: Any = None, public: bool = False) -> Router

A router serving POST path (and GET for the SDL, if enabled).

session_factory(request) supplies the ORM session. Without one, a read session is opened per request from the registry and closed after. An endpoint without the constructor's authorizer= must opt into an intentionally unauthorised surface with public=True.

run#

async run(source: str, session: Any, *, operation_name: str | None = None, variables: dict[str, Any] | None = None, request: Any = None) -> dict[str, Any]

Parse and execute source, returning a GraphQL response body.

schema (property)#

schema: Schema

sdl#

sdl() -> str

The schema in SDL form.

validate#

validate() -> None

Check every resolver dependency resolves, and freeze the schema.

Called automatically by router. A missing or cyclic requires is a wiring mistake, and finding it on the first request that happens to select that field is far too late.

GraphQLSyntaxError#

A malformed document or one that exceeded a safety limit.

GraphQLSyntaxError(message: str, *, code: str = 'syntax', position: int = 0) -> None

Limits#

Read-only safety bounds for one parse.

Limits(*, max_depth: int = 12, max_complexity: int = 1000, max_aliases: int = 50, max_steps: int = 200000, max_document_bytes: int = 16384) -> None

ResolverError#

A resolver could not be registered, or its dependencies do not resolve.

ResolverInfo#

What a resolver is told about the call.

Deliberately small. A resolver that needs the database takes the session; a resolver that needs the caller takes the request. Nothing here exposes the executor's internals, so the execution strategy stays free to change.

ResolverInfo(request: Any, session: Any, arguments: dict[str, Any], path: tuple[str, ...], parent_type: str) -> None

Schema#

Schema(registry: 'Any', types: 'dict[str, ObjectType]', roots: 'dict[str, RootField]', mutations: 'dict[str, RootField]' = <factory>)

Schema(registry: Any, types: dict[str, ObjectType], roots: dict[str, RootField], mutations: dict[str, RootField] = <factory>) -> None

mutation#

mutation(name: str) -> RootField | None

root#

root(name: str) -> RootField | None

sdl#

sdl() -> str

The schema in GraphQL SDL, for tooling and for a /graphql GET.

type_of#

type_of(name: str) -> ObjectType | None

build_schema#

build_schema(registry: Any, models: list[Any] | None = None, *, expose: Iterable[str] = (), dataclasses: Iterable[type] = ()) -> Schema

Build a schema from registry, optionally narrowed to models.

Exposure is opt-in when models is given, and that is the intended use: a registry holds every table the application has, including ones with no business being queryable from the internet. Passing None exposes them all, which is convenient in development and rarely right in production.

Columns whose names look sensitive are left out of the schema, on the same rule and the same regex wreath.crud.sensitive_fields uses -- password, *_hash, token, secret, api_key, and the rest. Both surfaces are generated from one ModelSpec, so it would be strange for the REST one to hide a password hash and the GraphQL one to answer { user { passwordHash } }. Name a column in expose to put it back, which is the same deliberate, auditable act crud_router(expose=...) asks for.

Retrieval columns are left out too, on the same argument and through the same function: wreath.crud.retrieval_fields reads the declared type rather than guessing from a name, so a Vector embedding and a TsVector are withheld here exactly as they are from a generated REST response. A tsvector is derived from columns already in the same selection, and a page of twenty Vector(1536) rows is thirty thousand floats a client did not want -- on a surface where the client, not the server, chooses the page. expose puts one back, and that is the whole of the difference between the two surfaces: crud_router(expose=...) also widens what may be written, and there is nothing here for it to widen, because no mutation is generated. A GraphQL write is a resolver somebody wrote, and it is that resolver's business what it accepts.

wreath.grpc#

Serve gRPC methods over the native HTTP/2 server.

gRPC is HTTP/2 plus three conventions: a five-byte length prefix in front of every message, a grpc-status in the response trailers, and a small set of grpc--prefixed headers. Wreath's native HTTP/2 server already speaks the hard part -- it emits response trailers (http.response.trailers) and validates the te: trailers a gRPC client is required to send -- so this module is ordinary Python over the ASGI messages that server already understands. There is no C here and no change to the server.

This is a server-tier feature. wreath.server is the only conforming server in the project that implements HTTP/2, and the ASGI trailers extension is what carries grpc-status. Running a gRPC service behind a foreign ASGI server -- uvicorn, hypercorn -- fails at startup with GrpcUnsupported naming the reason, rather than answering requests that no client can interpret.

A method is a route. GrpcService.router() returns an ordinary Router whose routes are POST /{service}/{method}, so a method's controls are a route's controls, enforced by the same middleware tape and read by the same permissions_router and wreath mutant. There is no second authorization model, and there are exactly two spellings because a route has two:

• action= and resource= on the method decorator, which are @authorize -- the same keyword @mcp.tool(action=...) uses, so one vocabulary spans REST, gRPC and MCP.
• @roles, @permissions, @authorize and @second_factor stacked on the handler, exactly as on a route handler. These are decorators and not keywords; permissions=, dependencies= and middleware= are the route metadata that passes through, and anything the route decorator does not accept is a TypeError at import rather than a control that silently does nothing.

from wreath.grpc import GrpcService from wreath.protobuf import message, field

@message class PositionRequest: collar_id: int = field(1)

tracker = GrpcService("camera.Tracker")

@tracker.unary(request=PositionRequest, response=Position, action="Collar::read", resource=EntityUid("Collar", "7")) async def GetPosition(request, message: PositionRequest) -> Position: ...

app.include_router(tracker.router())

Compression is identity and gzip, negotiated per call and applied per message through wreath.compression -- no dependency, and no zstd, because grpc-encoding is a registry shared with every other implementation.

Not built: server reflection, gRPC-Web, the health-checking protocol, and client-side concerns (load balancing, retry configuration, xDS). Reflection is not an oversight -- it requires protobuf descriptors, which wreath.protobuf deliberately does not build, so the two decisions are coupled and would have to be reopened together.

GrpcError#

Refuse a call with an explicit gRPC status.

Raise this from a method body when the refusal is the answer. Anything else that escapes a handler is mapped by status_for, which is deliberately conservative: an unrecognised exception becomes UNKNOWN, never a status that would tell a client the call is safe to retry.

GrpcError(status: Status, message: str = '') -> None

GrpcService#

A named collection of gRPC methods, served as ordinary routes.

name is the fully-qualified protobuf service name (camera.Tracker) and it is also the path prefix, because a gRPC path is /{service}/{method}.

GrpcService(name: str, *, max_message_bytes: int = 4194304) -> None

bidi#

bidi(*, request: type, response: type, **metadata: Any) -> Callable[[Any], Any]

The handler consumes request messages and yields response messages.

client_stream#

client_stream(*, request: type, response: type, **metadata: Any) -> Callable[[Any], Any]

The handler consumes request messages and returns one response.

router#

router() -> Router

An ordinary Router carrying one POST route per method.

Metadata passed to a method decorator reaches RouteDefinition unchanged, so roles=, dependencies= and rate_limit= are enforced by the same tape as any REST route and are read by permissions_router and wreath mutant from the same place. There is no second model.

server_stream#

server_stream(*, request: type, response: type, **metadata: Any) -> Callable[[Any], Any]

One request message; the handler yields response messages.

unary#

unary(*, request: type, response: type, **metadata: Any) -> Callable[[Any], Any]

One request message, one response message.

action= (with an optional resource=) is @authorize spelled at the declaration, and every other keyword is route metadata -- roles= and the rest are still written as decorators on the method, exactly as on a route handler.

GrpcUnsupported#

The running server cannot carry gRPC.

Raised at startup rather than per request. gRPC needs HTTP/2 and response trailers; a server offering neither would answer with a body no client can read, and a 200 full of unreadable bytes is a worse failure than a refusal naming the reason.

Status#

The gRPC status codes, from the specification's status-codes table.

These are the values that travel in the grpc-status trailer. OK is zero and is sent on success -- a gRPC call that succeeded still carries an explicit status, which is why the trailer is mandatory rather than an error-only affordance.

frame_message#

frame_message(payload: bytes, *, compressed: bool = False) -> bytes

Prefix one encoded message with its gRPC five-byte header.

parse_timeout#

parse_timeout(value: str) -> float

Seconds from a grpc-timeout header value, e.g. 100m or 5S.

A client that sent a deadline has already decided when to give up, so an unparseable value is refused rather than ignored -- silently treating it as "no deadline" would let a call outlive the caller that is waiting on it.

wreath.protobuf#

Protocol Buffers, declared in Python and compiled once at startup.

A message is an ordinary dataclass whose fields carry their wire numbers:

from wreath.protobuf import decode, encode, field, message

@message
class Position:
    collar_id: int = field(1)
    lat: float = field(2)
    lon: float = field(3)

raw = encode(Position(collar_id=7, lat=-33.8, lon=151.2))
back = decode(Position, raw)

Field numbers are the wire contract, so they are stated rather than derived from declaration order. Reordering the class must not silently change the bytes, which is exactly what a positional scheme would do.

The declaration is compiled at class creation into an operation-owned C descriptor and never re-derived. A declaration error is therefore a startup error, and the request path reads fixed C rows rather than Python metadata.

What this is, and is not#

This is a codec, not an implementation of protobuf. It speaks the wire format for declarations written here. It deliberately has no descriptors, no reflection, no Any, no dynamic messages, and it does not parse .proto files at runtime — declarations are Python, the way schemas, routes and validators already are. Groups (deprecated since proto2) are refused by name rather than mis-parsed. The canonical protobuf-JSON mapping is a separate specification and is not implemented.

Unknown fields are preserved, not rejected#

This is the one place the codec deliberately parts from wreath's usual strictness, and the reason is that the two situations are not the same one.

An unexpected name in a JSON body is a typo or a mistake, and rejecting it catches a real class of bug — so wreath.binding rejects extra fields. An unexpected number on a protobuf wire is a version-skew signal: field numbers are a deliberate, allocated contract, and tolerating unknown ones is the mechanism that lets a newer producer talk to an older consumer at all. A codec that rejected them would break the property protobuf exists to provide.

So unknown fields are captured verbatim and re-emitted on encode, which means a message survives a round trip through an intermediary built against an older declaration. unknown_fields(msg) returns those bytes if you need to see them.

Where the work happens#

The codec is src/wreath/_native/protobuf.c, held to the protobuf wire specification by tests/test_protobuf_parity.py.

Only the codec is C. The vocabulary a plan is written in — the KIND_* codes, the FLAG_* bits and ProtobufDecodeError — is read by the declaration compiler here too, which is Python, so it lives in src/wreath/_protobuf_plan.py and both read the same integers.

ProtobufDecodeError#

A buffer could not be parsed as the declared message.

Always raised by name rather than allowed to surface as an IndexError or a UnicodeDecodeError: this codec sits in front of bytes a peer controls, and a caller needs one exception type to catch and turn into a 400.

ProtobufDeclarationError#

A message declaration cannot be compiled to a wire plan.

Raised at class creation, never at encode time: a declaration is a contract with every peer, and the moment to find out it is wrong is import.

decode#

decode(data: bytes) -> Any

Decode data as cls.

Raises:

• ProtobufDecodeError — The buffer is truncated, carries a length prefix
• past its end, a varint longer than ten bytes, a group wire type, an
• unknown wire type, field number zero, or invalid UTF-8 in a string.

encode#

encode(msg: Any) -> bytes

Encode msg to protobuf bytes.

Unknown fields captured when this message was decoded are re-emitted, so a round trip through a build with an older declaration loses nothing.

Raises:

• ValueError — A value is outside the range its declared kind can hold.

field#

field(number: int, *, kind: str | None = None, packed: bool | None = None, oneof: str | None = None) -> Any

Declare a message field with its wire number.

Args:

• number — The field's wire number, 1..536870911 and outside 19000..19999.
• This is the contract with every peer; changing it is a breaking change.
• kind — Narrow the wire type: int defaults to int64, so pass
• "sint32" for a small signed value or "fixed64" for a hash. Must
• be compatible with the annotation.
• packed — Force the packed representation on or off for a repeated scalar.
• Defaults to packed, which is proto3's default. A decoder accepts
• both regardless.
• oneof — Group name. At most one field in a group may be set; setting one
• on decode clears the others. Every member must be optional.

Returns:

• A marker consumed by @message; never a value you read yourself.

is_message#

is_message(candidate: Any) -> bool

Whether candidate is a @message class (or an instance of one).

Public because consumers need to ask. Without it, a caller that wants to branch on "is this a protobuf message?" reads the private plan marker directly and thereby encodes its own second notion of what a message is -- which then drifts from this one the first time the marker moves.

message#

message() -> type[T]

Compile cls into a protobuf message and make it a dataclass.

Raises:

• ProtobufDeclarationError — A field number is duplicated, reserved, out of
• range, or annotated with a type that has no wire mapping — all at
• class creation, never at encode time.

unknown_fields#

unknown_fields(msg: Any) -> bytes

The raw bytes of fields this build did not recognise when decoding msg.

wreath.jsonpath#

A bounded RFC 9535 JSONPath compiler and evaluator.

JSONPath#

A compiled JSONPath expression reusable across JSON values.

JSONPath(source: str, _segments: tuple[_Segment, ...], _program: tuple[Any, ...], max_visits: int = 1000000) -> None

find#

find(value: Any) -> list[JSONPathMatch]

JSONPathError#

A JSONPath is invalid or exceeds its configured evaluation bound.

JSONPathMatch#

One selected JSON value and its normalized absolute path.

JSONPathMatch(value: Any, path: str) -> None

compile_jsonpath#

compile_jsonpath(source: str, *, max_visits: int = 1000000) -> JSONPath

Compile and validate an RFC 9535 JSONPath expression.

jsonpath#

jsonpath(source: str | JSONPath, value: Any) -> list[Any]

Return the selected values for a source string or compiled JSONPath.

wreath.linkset#

Web links serialized as RFC 9264 application/linkset+json.

LinkContext#

Links sharing one context URI, grouped by relation type.

LinkContext(anchor: str | None, links: Mapping[str, tuple[LinkTarget, ...]]) -> None

document#

document() -> dict[str, Any]

from_document (classmethod)#

from_document(document: Any) -> LinkContext

Linkset#

A set of links grouped into RFC 9264 link-context objects.

Linkset(*contexts: LinkContext) -> None

document#

document() -> dict[str, Any]

from_document (classmethod)#

from_document(document: Any) -> Linkset

LinksetResponse#

A UTF-8 RFC 9264 JSON linkset response.

LinksetResponse(linkset: Linkset, *, profile: str | None = None, status: int = 200, headers: Iterable[tuple[bytes, bytes]] | None = None) -> None

delete_cookie (inherited from Response)#

delete_cookie(name: str, *, path: str = '/', domain: str | None = None) -> None

Expire a cookie by resending it empty with Max-Age=0 and a 1970 Expires.

A browser replaces a stored cookie only when the name, Path, and Domain all match, so this must be given the same path and domain the cookie was set with -- the defaults clear a cookie that was set with the defaults, and a cookie scoped to /admin is not cleared by a deletion at /. SameSite is omitted, and Secure is carried for a __Secure-/__Host- name because a prefixed cookie can only be cleared by a Set-Cookie that still meets the prefix's attribute rules.

headers (property, inherited from Response)#

headers: list[tuple[bytes, bytes]]

Mutable wire headers, materialized when application code observes them.

set_cache_control (inherited from Response)#

set_cache_control(policy: CacheControl) -> None

Replace every cache-control header on this response with policy.

Existing cache-control headers are dropped first rather than appended to, because a cache reads the first line it finds and a second one added after a permissive first would have no effect.

set_cdn_cache_control (inherited from Response)#

set_cdn_cache_control(policy: CacheControl) -> None

Replace CDN-Cache-Control with an RFC 9213 targeted policy.

set_content_digest (inherited from Response)#

set_content_digest(*algorithms: str, content: bytes | None = None) -> None

Replace Content-Digest with RFC 9530 digests of the message content.

set_cookie (inherited from Response)#

set_cookie(name: str, value: str, *, max_age: int | None = None, expires: str | None = None, path: str = '/', domain: str | None = None, secure: bool = False, httponly: bool = False, samesite: str | None = 'lax') -> None

Append one set-cookie header to this response.

The name and value are sent verbatim; a caller quotes or encodes a value containing a separator. By default the cookie is host-only (no Domain), scoped to Path=/, SameSite=Lax, not Secure, not HttpOnly, and a session cookie -- it expires when the browser does, because neither max_age nor expires is set.

Enforces the browser-mandated attribute rules so a cookie is never emitted in a form every browser silently drops: SameSite must be one of strict/lax/none and SameSite=None requires Secure (RFC 6265bis 5.4.7); the __Secure-/__Host- name prefixes require Secure (and, for __Host-, Path=/ with no Domain) per RFC 6265bis 4.1.3. A control character or an attribute separator in the name, value, path, domain, or expires is a header-injection vector and is refused here rather than trusted as far as the serializer.

Args:

• max_age — Lifetime in seconds; the attribute is omitted when None.
• expires — An HTTP-date string, emitted verbatim as Expires.
• path — Path attribute, / by default; an empty string omits it.
• domain — Domain attribute; None keeps the cookie host-only.
• secure — Adds Secure, restricting the cookie to HTTPS.
• httponly — Adds HttpOnly, hiding the cookie from scripts.
• samesite — strict, lax (the default), or none; None omits the attribute.

Raises:

• ValueError — Any of the attribute rules or injection checks above failed.
• TypeError — max_age is not an integer.

set_link_templates (inherited from Response)#

set_link_templates(*templates: LinkTemplate) -> None

Replace Link-Template with validated RFC 9652 templated links.

set_repr_digest (inherited from Response)#

set_repr_digest(*algorithms: str, representation: bytes | None = None) -> None

Replace Repr-Digest with digests of the complete selected representation.

set_want_content_digest (inherited from Response)#

set_want_content_digest(preferences: DigestPreferences) -> None

Ask the peer for Content-Digest on future requests.

set_want_repr_digest (inherited from Response)#

set_want_repr_digest(preferences: DigestPreferences) -> None

Ask the peer for Repr-Digest on future requests.

LinkTarget#

One target URI and its Web Linking target attributes.

LinkTarget(href: str, hreflang: tuple[str, ...] = (), media: str | None = None, title: str | None = None, type: str | None = None, attributes: Mapping[str, Any] = <factory>) -> None

document#

document() -> dict[str, Any]

from_document (classmethod)#

from_document(document: Any) -> LinkTarget

wreath.http_client#

Lifespan-managed dependency-free outbound HTTP/1.1 client.

One HTTPClient is one bounded connection pool for one origin. The origin is fixed at construction from base_url and never changes: every request target is origin-relative, and a redirect that would leave the origin is refused rather than followed. Calling a second service means a second client, which is what makes each service's limits, timeouts, retries, and destination policy separately configurable and separately observable.

client = HTTPClient("billing", base_url="https://billing.internal/v1")
async with client:
    response = await client.get("/invoices/42")

The client must be started before it will serve a request; async with does that, and an application normally starts and closes it from the lifespan instead. A request on a client that is not started raises ClientClosed rather than starting one implicitly, so a missing lifespan registration fails loudly.

Everything the caller configures is a frozen dataclass validated at construction (ClientLimits, ClientTimeout, RetryPolicy, RedirectPolicy, DestinationPolicy, RatePolicy), so a misconfiguration raises at boot rather than on the first outbound call.

Responses are read fully into memory under ClientLimits.max_response_bytes; there is no streaming response API. That is a deliberate limitation of this client, which exists to call other services, not to download files.

The public policy and pool are Python; the byte codecs are C, served by the dedicated _client extension when the build has it and by _core otherwise.

ClientClosed#

The client is not started, or was closed while this request was in flight.

Raised rather than starting the client implicitly: a client that is not started is almost always one that was never registered with the lifespan, and silently starting it would move the failure to shutdown.

ClientError#

Base class for outbound client failures.

Every failure this module raises derives from it, so except ClientError is the one catch that covers the client without also catching bugs in the calling code. ClientResponse.raise_for_status raises this base class directly for a >=400 status, which is the only case where the exception type carries no more information than the message.

ClientLimits#

Sizes and counts one client will not exceed. Validated at construction.

These are the client's share of the process, chosen per origin: a slow dependency can only ever hold max_connections sockets and max_waiters parked callers, whatever the rest of the application does.

Args:

• max_connections — Sockets open to the origin at once, idle and in-flight together.
• max_keepalive_connections — Idle sockets retained for reuse. Cannot exceed max_connections.
• max_waiters — Callers parked on a full pool; the next one gets PoolTimeout immediately.
• max_request_header_bytes — Serialized request head ceiling; over it raises ClientError.
• max_response_header_bytes — Response head ceiling, also applied to chunked trailers.
• max_response_bytes — Decoded response-body ceiling; over it raises ResponseTooLarge.
• read_high_water — Transport read buffer, in bytes. Not a response-size limit.
• dns_cache_ttl — Seconds a resolution is reused. 0 re-resolves every connect.

Raises:

• ValueError — Any count is non-positive, the TTL is negative, or keepalive exceeds max.

ClientLimits(max_connections: int = 20, max_keepalive_connections: int = 10, max_waiters: int = 100, max_request_header_bytes: int = 32768, max_response_header_bytes: int = 32768, max_response_bytes: int = 2097152, read_high_water: int = 262144, dns_cache_ttl: float = 30.0) -> None

ClientResponse#

One complete outbound response. Immutable, fully buffered.

body holds the decoded body in full -- the client has no streaming response API, and the read is bounded by ClientLimits.max_response_bytes. Informational (1xx) responses are consumed and discarded during the read, so status is always final.

Headers stay raw bytes pairs in wire order with names lowercased, duplicates included, exactly as the origin sent them. Nothing is combined or decoded, because deciding a charset for someone else's header is how a client starts corrupting values.

Args:

• headers — Response headers, lowercased names, in the order received.
• http_version — "1.0" or "1.1", from the status line.
• reason — The reason phrase. Advisory; never parse it.

ClientResponse(status: int, headers: tuple[tuple[bytes, bytes], ...], body: bytes, http_version: str, reason: bytes = b'') -> None

header#

header(name: bytes) -> bytes | None

The first value for name, or None. Case-insensitive.

A header sent more than once resolves to the first occurrence; iterate headers for the rest, which matters for Set-Cookie.

raise_for_status#

raise_for_status() -> None

Raise ClientError when the status is 400 or above; otherwise return None.

Opt-in: the client itself never treats a status as an error, because a 404 or a 409 is frequently the answer rather than a failure. 3xx does not raise here -- an unfollowed redirect is a response the caller asked for.

Raises:

• ClientError — The status is >= 400. The message carries the status.

ClientSnapshot#

A point-in-time reading of one client's pool. For health and metrics.

Taken without a lock and without awaiting, so the fields are individually accurate and collectively approximate -- fine for a gauge, not a basis for a decision. active + idle is not the pool's open count: a connection being created belongs to neither set yet.

Args:

• active — Connections currently carrying a request.
• idle — Connections held for reuse.
• waiters — Callers parked waiting for a connection right now.
• requests — Requests written since construction; monotonic, never reset.
• reused — Times an idle connection was taken instead of dialling; monotonic.

ClientSnapshot(active: int, idle: int, waiters: int, requests: int, reused: int) -> None

ClientTLS#

Where this client's trust comes from, as paths rather than a context.

Paths, not a built ssl.SSLContext, and that is what makes the fast path reachable. There is no supported way to borrow OpenSSL's SSL_CTX * out of a Python context, so a client handed one can only take asyncio's TLS -- a Python object per read and per write, measured at 2.14x the cost of the native path. Naming the material lets the reactor build its own. TLSConfig and the HTTP/3 backend already answer this the same way.

Args:

• cafile — PEM bundle of trusted roots. None uses the system store.
• capath — Directory of hashed trusted roots.
• verify — Check the peer's chain and host name. Off is a decision that has
• to be typed out, because a client that skips the check is faster
• than one that does not and looks identical until it matters.

ClientTLS(cafile: str | None = None, capath: str | None = None, verify: bool = True) -> None

ClientTimeout#

Deadlines for one outbound request, in seconds. Validated at construction.

total is the outer bound and the only one a caller can rely on: it wraps everything, including retry backoff sleeps and redirect hops, so the inner deadlines cannot compose into something longer. The inner ones exist to fail a stalled stage early enough to be worth retrying, which the outer one cannot distinguish.

Args:

• pool — Waiting for a free connection; over it raises PoolTimeout.
• connect — TCP connect per candidate address; also bounds DNS resolution.
• tls — Handshake, added to connect for the combined per-address deadline.
• response_headers — Silence before the response head completes.
• response_body — Silence during the body, re-armed per read rather than total.
• total — The whole call, retries included. None removes the outer deadline.

Raises:

• ValueError — Any value is non-positive (total may be None, but not 0).

ClientTimeout(pool: float = 1.0, connect: float = 5.0, tls: float = 5.0, response_headers: float = 10.0, response_body: float = 30.0, total: float | None = 45.0) -> None

ConnectError#

Every candidate address failed to connect within the connect timeout.

Carries the last underlying error as its __cause__.

DNSFailure#

The origin host did not resolve, resolution timed out, or yielded nothing.

DestinationPolicy#

Where this client is permitted to connect. The SSRF guard.

Checked in two places, because a name check alone is not a destination check: validate_url runs on the base_url at construction and on any absolute redirect target, and validate_address runs on every address DNS returned, before a connection is attempted. That second check is what a hostname resolving to 127.0.0.1 or 169.254.169.254 has to pass, and it is why a DNS answer is validated in full rather than only the address that happens to win the connection race. Addresses under the well-known NAT64 prefix, IPv4-compatible literals, 6to4, Teredo, and ISATAP are checked again as their translated IPv4 destination, so an IPv6 answer cannot tunnel to loopback or cloud metadata.

The defaults deny every non-global address. Loopback is denied too, so a client aimed at http://localhost in a test needs allow_loopback=True -- the default is chosen for production, and a test opting in is a smaller mistake than a service reaching a metadata endpoint.

Args:

• schemes — URL schemes permitted. Anything outside raises DestinationRejected.
• hosts — Allowed hostnames; *.example.com matches subdomains, not the apex. Empty: any.
• ports — Allowed ports, compared after the scheme default is applied. Empty allows any.
• allow_private — Permit RFC 1918 and other private ranges.
• allow_loopback — Permit 127.0.0.0/8 and ::1.
• allow_link_local — Permit 169.254.0.0/16 and fe80::/10, which include cloud metadata.

DestinationPolicy(schemes: frozenset[str] = frozenset({'https', 'http'}), hosts: tuple[str, ...] = (), ports: frozenset[int] = frozenset(), allow_private: bool = False, allow_loopback: bool = False, allow_link_local: bool = False) -> None

validate_address#

validate_address(value: str) -> None

Check one resolved IP address against the policy.

Deny-by-default on the last rule: an address that is not global and not covered by an explicit allow_* is refused even if it belongs to no named category, so a range this code does not enumerate fails closed. Unspecified (0.0.0.0, ::) and multicast addresses are refused unconditionally -- no flag admits them.

Args:

• value — A textual IP address; an IPv6 %scope suffix is stripped first.

Raises:

• DestinationRejected — The address is outside what the policy permits.
• ValueError — value is not a valid IP address.

validate_url#

validate_url(parsed: SplitResult) -> None

Check a parsed URL's scheme, credentials, host, and port.

Userinfo (https://user:pass@host) is refused outright rather than stripped or forwarded: it would put a credential in a URL that gets logged, and no supported flow needs it.

Hostnames are IDNA-encoded and lowercased before matching, so a Unicode or mixed-case host cannot slip past an allow-list entry.

Args:

• parsed — The result of urllib.parse.urlsplit.

Raises:

• DestinationRejected — Scheme, credentials, host, or port fail the policy.
• UnicodeError — The hostname is not encodable as IDNA.

StreamingClientResponse#

A response whose head has arrived and whose body has not.

Deliberately not a ClientResponse. That class documents itself as "Immutable, fully buffered" and callers rely on .body; giving this one a .body that sometimes worked would be worse than not having one.

Yielded by HttpClient.stream, and only valid inside that block.

StreamingClientResponse(status: int, headers: tuple[tuple[bytes, bytes], ...], http_version: str, reason: bytes, _chunks: Any) -> None

header#

header(name: bytes) -> bytes | None

The first value for name, or None. Names are lowercase on the wire.

iter_bytes#

iter_bytes() -> Any

The body, as it arrives. Iterate once.

DestinationRejected#

A URL or a resolved address is outside DestinationPolicy.

The SSRF guard. Raised from the constructor for a bad base_url, and per request from DNS resolution for an address the policy does not allow.

HTTPClient#

One bounded pool for a configured HTTP origin.

The origin comes from base_url and is fixed for the client's life, as is every policy passed here. Any path in base_url becomes a prefix on every request target. Requests are origin-relative and validated as such: a target that is not /-prefixed, or is protocol-relative (//host), or is not ASCII, raises ValueError before anything is sent.

Concurrency is bounded by ClientLimits, and connections are reused when the response was framed and the peer did not ask to close. Happy Eyeballs is applied across the addresses DNS returned: attempts are staggered 250 ms apart, the first to connect wins, and the losers are closed.

Not started at construction. start admits requests, close drains them, and the object is an async context manager over both. Every method is safe to call from many tasks at once.

Args:

• name — Identifies this client in errors and diagnostics. Cannot be empty.
• base_url — Scheme, host, optional port, optional path prefix. No query or fragment.
• limits — Pool sizes and byte ceilings.
• timeout — Per-request deadlines.
• retry — When a failed attempt is repeated. Disabled by default.
• redirect — Whether 3xx is followed. Disabled by default.
• destination — Where connecting is permitted. Applied to base_url here.
• rate — Outbound throttle. Disabled by default.

Raises:

• ValueError — name is empty, or base_url carries a query or fragment.
• DestinationRejected — base_url is outside destination.

HTTPClient(name: str, *, base_url: str, limits: ClientLimits = ClientLimits(max_connections=20, max_keepalive_connections=10, max_waiters=100, max_request_header_bytes=32768, max_response_header_bytes=32768, max_response_bytes=2097152, read_high_water=262144, dns_cache_ttl=30.0), timeout: ClientTimeout = ClientTimeout(pool=1.0, connect=5.0, tls=5.0, response_headers=10.0, response_body=30.0, total=45.0), retry: RetryPolicy = RetryPolicy(attempts=1, idempotent_only=True, statuses=frozenset({500, 408, 502, 503, 504, 425, 429}), backoff_base=0.05, backoff_cap=1.0, jitter=True, respect_retry_after=True), redirect: RedirectPolicy = RedirectPolicy(enabled=False, max_hops=0), destination: DestinationPolicy = DestinationPolicy(schemes=frozenset({'https', 'http'}), hosts=(), ports=frozenset(), allow_private=False, allow_loopback=False, allow_link_local=False), rate: RatePolicy = RatePolicy(enabled=False, capacity=0.0, rate=0.0, max_wait=30.0), trace: TracePolicy = TracePolicy(propagate=True, tracestate=False), tls: ClientTLS | None = None) -> None

close#

async close() -> None

Stop admitting requests, drop idle connections, wait for in-flight ones.

Refusal is immediate: parked callers are woken with ClientClosed and a connect that completes after this point is closed rather than pooled. Idle sockets are closed at once; in-flight requests are then given timeout.total (or timeout.response_body when total is None) to finish on their own, after which their transports are closed underneath them and they fail rather than hang.

A socket that errors while closing is ignored rather than propagated -- shutting down is not a place to acquire a new failure. Idempotent, and safe on a client that was never started.

counters#

counters() -> Any

This client's pool counters, for wreath.metrics.collect.

get#

get(target: str, *, headers: tuple[tuple[bytes, bytes], ...] = ()) -> Coroutine[Any, Any, ClientResponse]

GET through request. Retried under an idempotent-only policy.

Returning the request coroutine directly preserves deferred execution while avoiding a forwarding coroutine frame around every response.

name (property)#

name: str

The name this client was constructed with. Appears in its error messages.

origin (property)#

origin: str

The immutable scheme, host, and port this client connects to.

post#

post(target: str, *, headers: tuple[tuple[bytes, bytes], ...] = (), body: bytes | bytearray | memoryview = b'', idempotency_key: str | None = None) -> Coroutine[Any, Any, ClientResponse]

POST through request.

POST is not idempotent, so an idempotency_key is what makes this retryable under the default policy: it is sent as an Idempotency-Key header and is the caller's assertion that the origin collapses duplicates. Without one, a POST is attempted exactly once.

query#

query(target: str, *, headers: tuple[tuple[bytes, bytes], ...] = (), body: bytes | bytearray | memoryview = b'', content_type: str = 'application/json') -> Coroutine[Any, Any, ClientResponse]

Send a safe, idempotent QUERY with its mandatory Content-Type.

request#

async request(method: str, target: str, *, headers: tuple[tuple[bytes, bytes], ...] = (), body: bytes | bytearray | memoryview = b'', idempotency_key: str | None = None) -> ClientResponse

Send one request to this client's origin and return the full response.

The whole lifecycle happens here: rate-limit token, pool acquisition, DNS with policy validation, connect, TLS, write, read, retries, and redirects -- all inside ClientTimeout.total.

The response is returned for any status. Nothing raises on a 4xx or 5xx; call ClientResponse.raise_for_status when that is wanted. A 3xx is returned unchanged unless RedirectPolicy is enabled.

The request is recorded as one phase in the flight recorder when the request that triggered it was armed, including when it fails or times out -- the duration of a failed call is precisely what the Inspector needs. An unarmed request pays one ContextVar read.

Args:

• method — Case-insensitive; uppercased before it goes on the wire.
• target — Origin-relative path, /-prefixed, ASCII/percent-encoded.
• headers — Raw byte pairs, sent in order after the client's own head.
• body — Sent verbatim. Copied to bytes once, so a buffer may be reused after.
• idempotency_key — Non-empty value sent as Idempotency-Key; makes a
• non-idempotent method retryable. Do not also supply that header.

Returns:

• The response, body included.

Raises:

• ClientClosed — The client is not started, or was closed mid-request.
• ValueError — target is not an origin-relative ASCII path.
• ClientError — The request headers exceed ClientLimits.max_request_header_bytes.
• PoolTimeout — No connection became free within ClientTimeout.pool.
• DNSFailure — The origin did not resolve.
• DestinationRejected — A resolved address is outside DestinationPolicy.
• TLSFailure — The handshake failed.
• ConnectError — Every candidate address failed to connect.
• ResponseTimeout — The origin stalled mid-response.
• RequestTimeout — The call exceeded ClientTimeout.total.
• ProtocolError — The response could not be framed.
• ResponseTooLarge — The body exceeded ClientLimits.max_response_bytes.
• RedirectError — A redirect could not be followed under RedirectPolicy.

snapshot#

snapshot() -> ClientSnapshot

A ClientSnapshot of the pool right now. Synchronous, never blocks.

start#

async start() -> None

Admit requests. Idempotent, and cheap enough to call from a lifespan hook.

Opens no connections and resolves no names -- both happen on the first request. Starting is only a permission, so a client for an origin that is currently down still starts, and the failure surfaces at the call that needs it.

A closed client can be started again; the pool is empty at that point, since close discarded it.

started (property)#

started: bool

Whether the client is accepting requests. False before start and after close.

stream#

stream(method: str, target: str, *, headers: tuple[tuple[bytes, bytes], ...] = (), body: bytes | bytearray | memoryview = b'') -> _StreamContext

Send one request and hand back the response before its body arrives.

The half request cannot do. request reads the whole body into memory under ClientLimits.max_response_bytes, which is right for an API call and impossible for a proxy, a large download, or an SSE stream that never ends.

async with client.stream("GET", "/big") as response:
    async for chunk in response.iter_bytes():
        await sink.write(chunk)

The connection is held for the life of the block, which is why this is a context manager and not a coroutine returning a response: the body is still on the socket, and releasing the connection early would hand the next caller a pool entry with someone else's bytes in front of it.

A body left unread is not a bug, and the exit path handles it the only safe way: the connection is closed rather than pooled. Draining an unknown remainder to make it reusable would let an upstream decide how long a client blocks.

No retries and no redirects, deliberately. Once a byte of the body has been delivered the request cannot be replayed, and a streaming call that silently restarted would deliver a prefix twice. Use request when those matter; they are what it is for.

Args:

• method — Case-insensitive; uppercased before it goes on the wire.
• target — Origin-relative path, /-prefixed, ASCII/percent-encoded.
• headers — Raw byte pairs, sent in order after the client's own head.
• body — The request body, sent before the response is read.

Returns:

• An async context manager yielding a StreamingClientResponse.

PoolTimeout#

No connection became available within ClientTimeout.pool.

Also raised immediately, without waiting, when ClientLimits.max_waiters callers are already queued -- shedding at a known bound beats an unbounded queue that turns one slow origin into the whole process's latency.

ProtocolError#

The origin's bytes are not a response this client can frame.

Covers an unparseable head, headers past the configured limit, a malformed chunked body, a 101 (protocol switching is not supported), and bytes still buffered after a fully framed body -- the last being a request-smuggling shape, which is why it is refused rather than ignored.

RedirectError#

A redirect could not be followed under RedirectPolicy.

Raised when the hop limit is exhausted, when the Location is not ASCII, when it leaves the origin, and when a 301/302 on a method other than GET, HEAD, or QUERY would require rewriting the method -- refused rather than silently turned into a GET, because that would drop the body the caller asked to send.

RatePolicy#

Client-side outbound throttle. Off by default; validated at construction.

One continuous token bucket shared by every request from this client (rate tokens per second, bursting to capacity), reusing the same TokenBucket primitive as the inbound rate limiter rather than a second implementation. A request that cannot take a token parks rather than being rejected, so this shapes traffic to a rate-limited dependency instead of turning its limit into the caller's errors -- but only up to max_wait, past which the wait becomes a ClientError.

Retries spend tokens too: each attempt acquires before it is sent.

Args:

• enabled — When False every other field is ignored and no bucket exists.
• capacity — Burst size in tokens. One request costs one token.
• rate — Sustained refill in tokens per second.
• max_wait — Seconds a request may park. Exceeding it raises ClientError.

Raises:

• ValueError — enabled with a non-positive rate, capacity, or max_wait.

RatePolicy(enabled: bool = False, capacity: float = 0.0, rate: float = 0.0, max_wait: float = 30.0) -> None

RedirectPolicy#

Whether 3xx responses are followed. Off by default; validated at construction.

With redirects disabled the 3xx response is returned to the caller as-is, which is a valid way to use this client -- nothing is raised. Enabling them requires a hop limit, because the redirect loop is the client's own and an unbounded one is a hang.

Method rewriting follows RFC 9110: a 303 becomes a GET with an empty body, while 307 and 308 preserve both. A 301 or 302 on anything other than GET or HEAD raises RedirectError instead of being rewritten, because rewriting would silently drop the body the caller asked to send.

Cross-origin redirects are never followed, and there is no option to. The client is pinned to one origin: its pool, its TLS context, its destination policy, and its rate and retry budgets all belong to that origin, and following a hop off it would run the new origin's requests under the old one's everything. RedirectPolicy has no cross-origin flag; reaching another origin requires a client configured for that origin.

Args:

• enabled — Follow 3xx responses. When False they are returned unchanged.
• max_hops — Redirects followed before RedirectError. Must be positive when enabled.

Raises:

• ValueError — max_hops is negative, or redirects are enabled with max_hops=0.

RedirectPolicy(enabled: bool = False, max_hops: int = 0) -> None

RequestTimeout#

The whole request exceeded ClientTimeout.total.

The outermost deadline: it covers pool waiting, DNS, connect, TLS, retries, backoff sleeps, and redirect hops together, so no combination of the inner timeouts can outlast it. total=None removes it.

ResponseTimeout#

The origin stopped sending mid-response.

Distinct from RequestTimeout: this one is an inner deadline (ClientTimeout.response_headers or response_body), and it is retryable because it means the connection stalled rather than that the caller's budget ran out.

ResponseTooLarge#

The response body exceeded ClientLimits.max_response_bytes.

A declared Content-Length over the limit is refused before a byte of body is read; chunked and close-delimited bodies are counted as they arrive. A missing or lying length therefore cannot get past it either.

RetryPolicy#

When and how often a failed attempt is repeated. Validated at construction.

Retries are off by default (attempts=1 means one attempt, not one retry). Turning them on requires deciding that repeating the request is safe, which is what idempotent_only encodes: with it set, only DELETE, GET, HEAD, OPTIONS, PUT, QUERY and TRACE are repeated -- plus any request carrying an idempotency_key, because that key is the caller's promise that the origin will collapse a duplicate.

Two kinds of failure are retried: a transport-level one (connect, DNS, response timeout, a connection that died mid-exchange) and a response whose status is in statuses. A ProtocolError, a ResponseTooLarge, or any other ClientError is final -- repeating it would fail identically.

Backoff between attempts is min(backoff_base * 2**attempt, backoff_cap). A Retry-After on a retryable response replaces that delay entirely, in either RFC 9110 §10.2.3 form (delta-seconds or an HTTP-date), but is clamped to backoff_cap * 16 so one absurd header cannot park the caller. Each attempt also spends a token of the client's RatePolicy, so a retry storm is throttled by the same bucket as ordinary traffic.

Args:

• attempts — Total attempts, not extra ones. 1 disables retrying.
• idempotent_only — Restrict retries to idempotent methods and keyed requests.
• statuses — Response statuses treated as retryable.
• backoff_base — First delay in seconds, doubled per attempt.
• backoff_cap — Ceiling for the computed delay, in seconds.
• jitter — Scale each delay by a uniform factor in [0.5, 1.0) to spread a thundering herd.
• respect_retry_after — Honour a Retry-After header in place of the computed delay.

Raises:

• ValueError — attempts is non-positive, or either backoff value is non-positive.

RetryPolicy(attempts: int = 1, idempotent_only: bool = True, statuses: frozenset[int] = frozenset({500, 408, 502, 503, 504, 425, 429}), backoff_base: float = 0.05, backoff_cap: float = 1.0, jitter: bool = True, respect_retry_after: bool = True) -> None

TLSFailure#

The TLS handshake failed. Preferred over ConnectError whenever any candidate address raised an ssl.SSLError, so a certificate problem is never reported as an unreachable host.

wreath.bff#

OAuth browser backend-for-frontend primitives from RFC 10017.

A browser BFF keeps OAuth tokens in a server-side session and exposes only application-shaped routes to the browser. Each proxied resource is compiled to one fixed-origin outbound client and one fixed path prefix; no request value can select a scheme, host, port, or resource client.

BFFResource#

One fixed OAuth resource server exposed through the BFF.

client is normally a wreath.http_client.HTTPClient. Its HTTPS origin is inspected once here; outbound requests receive only origin-relative targets under target_prefix.

BFFResource(client: Any, target_prefix: str = '/', methods: Set[str] = frozenset({'GET', 'DELETE', 'PATCH', 'PUT', 'QUERY', 'POST'})) -> None

bff_access_token#

bff_access_token(request: Request) -> str | None

Return the unexpired access token held in the active BFF session.

bff_router#

bff_router(resources: Mapping[str, BFFResource], *, prefix: str = '/bff', token: TokenResolver = <function bff_access_token>, csrf_header: tuple[bytes, bytes] = (b'x-wreath-bff', b'1')) -> Router

Compile fixed-resource browser routes for an OAuth backend-for-frontend.

Every proxy request must carry csrf_header exactly once. Since a custom header is not CORS-safelisted, a cross-origin browser request must first pass the application's CORS preflight policy. Cookie, Authorization, Host, hop-by-hop fields, and unlisted extension fields never reach the resource server; its Set-Cookie and hop-by-hop fields never reach the browser.

bff_session_policy#

bff_session_policy(secret: str, *, store: Any, max_age: int = 1209600, previous_secrets: Any = ()) -> SessionPolicy

Build the strict server-side session required by an OAuth browser BFF.

The cookie is host-only, scoped to /, Secure, HttpOnly, SameSite=Strict, and carries only an opaque signed session id. A missing store is refused because Wreath's ordinary fallback stores session contents in the cookie; signing OAuth tokens is not the encryption RFC 10017 requires for that deployment shape.

clear_bff_tokens#

clear_bff_tokens(request: Request) -> None

Remove the BFF token set and rotate the active server-side session.

set_bff_tokens#

set_bff_tokens(request: Request, *, access_token: str, refresh_token: str | None = None, expires_at: int | float | None = None) -> None

Put an OAuth token set in the active server-side BFF session.

Call this after the authorization-code exchange. The session id is rotated at the privilege change, and no token is returned to the browser.

wreath.service_client#

A thin, auth-aware wrapper around wreath.http_client.HTTPClient.

Calling another service usually means the same two chores on every request: put a base path in front of the target, and attach a bearer token that has to be kept fresh. ServiceClient binds both once so call sites read like the API:

from wreath.service_client import ServiceClient
from wreath._auth.oauth2 import ClientCredentials

billing = ServiceClient(
    http_client,                                   # an origin-pinned HTTPClient
    base_path="/billing/v1",
    token=ClientCredentials(http_client=http_client, token_path="/oauth/token",
                            client_id=..., client_secret=...),
)
response = await billing.get(f"/invoices/{id}")
invoice = loads(response.body)

The token source is auto-refreshing: a ClientCredentials caches and renews the M2M token before it expires, so ServiceClient just asks for the current one per request. A plain str (static token) or a zero-arg async callable (your own refresh logic) work too.

The response is whatever the wrapped client returns -- for HTTPClient that is a ClientResponse, which holds the body as bytes and does not decode it.

ServiceClient#

An HTTP client bound to a base path and a (refreshing) bearer token.

Every method delegates to the wrapped client's request, so the pool, timeouts, retries, and destination policy are the wrapped client's -- this adds path prefixing and an Authorization header and nothing else. It holds no state of its own and is safe to share across concurrent requests to exactly the degree the wrapped client is.

Three token sources are accepted, resolved in this order: a plain str (static), any object with an awaitable .token() (a wreath._auth.oauth2.ClientCredentials, which caches and renews), and a zero-argument async callable. Anything else raises TypeError on the first request rather than at construction. A resolved token that is not printable ASCII is refused with ValueError naming the character and its position -- a token carrying \r or \n would otherwise splice a second header into the request.

Args:

• client — An origin-pinned client, normally a wreath.http_client.HTTPClient.
• token — Bearer-token source, or None to send no Authorization header.
• base_path — Prefix put in front of every path; a trailing / is stripped.
• default_headers — Sent on every request, after the bearer header and before the caller's.

ServiceClient(client: Any, *, token: Any = None, base_path: str = '', default_headers: _Headers = ()) -> None

delete#

async delete(path: str, *, headers: _Headers = ()) -> Any

DELETE through request. Sends no body.

get#

async get(path: str, *, headers: _Headers = ()) -> Any

GET through request.

patch#

async patch(path: str, *, headers: _Headers = (), body: bytes | bytearray | memoryview = b'') -> Any

PATCH through request.

post#

async post(path: str, *, headers: _Headers = (), body: bytes | bytearray | memoryview = b'', idempotency_key: str | None = None) -> Any

POST through request. The only verb here taking an idempotency key.

put#

async put(path: str, *, headers: _Headers = (), body: bytes | bytearray | memoryview = b'') -> Any

PUT through request.

request#

async request(method: str, path: str, *, headers: _Headers = (), body: bytes | bytearray | memoryview = b'', idempotency_key: str | None = None) -> Any

Send one request to base_path + path with the bearer header attached.

The token is resolved per call, so a refreshing source hands over a current credential on every request. Ambiguous Authorization combinations are refused; a lone caller-supplied value is accepted only when the client has no configured credential.

Args:

• path — Origin-relative; a leading / is optional and one is inserted if absent.
• idempotency_key — Forwarded to the wrapped client, which sends it as Idempotency-Key.

Returns:

• Whatever the wrapped client returns -- a ClientResponse for HTTPClient.

Raises:

• TypeError — The configured token source is none of the three supported shapes.
• ValueError — The resolved token is not printable ASCII, so it cannot be
• carried in an Authorization header.

wreath.webhooks#

Signed inbound and outbound webhook primitives.

A webhook is somebody else's HTTP request arriving with a claim about who sent it, or one of yours leaving with the same claim attached. Both halves are here, and both are built around the same signature profile.

The profile. Wreath signs wreath-v2: HMAC-SHA256 over the exact request body joined with the event's timestamp, id, type, version and lineage. The MAC covers the bytes that were sent, not a re-serialization of them, so a verifier that reformats the JSON before checking would fail -- which is the point. Signed fields are refused if they contain a control character, because the signature base joins them with newlines and an embedded newline would make the split ambiguous. Verification also bounds the timestamp (HMACWebhookVerifier.max_age), so a captured request cannot be replayed forever. Keys are a mapping of key id to secret, and the id travels in a header, which is what makes rotation an ordinary deployment rather than a cutover.

Receiving. WebhookHub registers a WebhookSource per sender and a handler per event type. A request is bounded, verified, deduplicated and dispatched, in that order. Deduplication has two forms and they are not equivalent: LocalReplayStore is per process, and PostgresWebhookInbox is a claim in a table every replica shares, with a fencing token so a lease that expired cannot complete over the top of the worker that took it over.

Sending. WebhookDestination signs and posts. send does it now, and reports the outcome as delivered, failed or unknown -- three states because a transport failure genuinely does not say whether the peer processed the request. enqueue instead commits the intent to PostgresWebhookOutbox inside the caller's transaction, so the delivery exists exactly when the change that caused it does, and WebhookDispatcher sends it afterwards with leases, retries and backoff.

Relaying. An event received can cause an event sent. relay and enqueue_relay carry correlation and causation ids forward and append this service to the envelope's relay path, which is itself signed. A path that would repeat a service, or grow past the hop limit, is refused -- that is what stops two services that subscribe to each other from generating traffic forever.

DispatcherReadiness#

A health-endpoint view of one dispatcher.

ready is the composite: running and no recorded error. The parts are kept separate because they fail differently -- a dispatcher that is not running at all is a startup problem, while one running with last_error set is a dispatcher that died and was restarted, or is about to stop.

Args:

• ready — Running with no recorded error. The single value a probe should read.
• running — Whether the delivery loop is currently executing.
• in_flight — Deliveries currently awaiting a peer response.
• last_error — "Type: message" for the failure that ended the last run, or None.

DispatcherReadiness(ready: bool, running: bool, in_flight: int, last_error: str | None) -> None

GitHubWebhookVerifier#

Verify GitHub's SHA-256 webhook signature and delivery identity.

GitHub signs no timestamp or delivery id, so freshness comes from the replay ledger keyed by a digest of the authenticated body. X-GitHub-Delivery remains the event id exposed to handlers. max_age is therefore the retention time for that ledger rather than a signature check.

GitHubWebhookVerifier(secret: bytes | str, *, replay_ttl: float = 86400.0) -> None

max_age (property, inherited from _NormalizedWebhookVerifier)#

max_age: float

verify (inherited from _NormalizedWebhookVerifier)#

verify(*, body: bytes, headers: Mapping[bytes, bytes], now: datetime | None = None) -> WebhookEnvelope

HMACWebhookSigner#

Sign Wreath's versioned exact-body HMAC-SHA256 profile.

Holds every key it may ever have to sign with, and one id to sign with by default. The others are not spares: a redelivery from the outbox must be signed with the key the row was enqueued under, or a receiver that has since rotated would reject a delivery it had already accepted. That is why headers takes a key_id at all.

Keys are copied at construction, so the caller's mapping may change afterwards without affecting an in-flight signer.

Args:

• keys — Key id to secret. Must contain key_id, and that secret must be non-empty.
• key_id — The key used when a caller does not name one.

Raises:

• ValueError — key_id is not in keys, or its secret is empty.

HMACWebhookSigner(keys: Mapping[str, bytes], *, key_id: str) -> None

headers#

headers(envelope: WebhookEnvelope, *, key_id: str | None = None, signature_profile: str = 'wreath-v2-hmac-sha256') -> tuple[tuple[bytes, bytes], ...]

The signed header set for envelope. Does not include Content-Type.

Always carries the id, type, version, timestamp, key id and signature; correlation, causation and relay-path headers appear only when the envelope has them. The timestamp is normalized to UTC with microsecond precision and a Z suffix, and it is the normalized form that is signed -- so the header must be transmitted byte-for-byte as returned.

Args:

• key_id — Sign with this key instead of the default, e.g. redelivering an old row.
• signature_profile — Reproduce the profile recorded with an outbox row.

Returns:

• Raw (name, value) byte pairs, ready to pass to the HTTP client.

Raises:

• ValueError — The key is unavailable or the profile cannot safely sign the envelope.

key_id (property)#

key_id: str

The default signing key id. Recorded on an outbox row at enqueue time.

HMACWebhookVerifier#

Verify Wreath's HMAC profile with a bounded timestamp window.

Holding several keys is how rotation works: a sender switches key id when it is ready, and both are accepted until the old one is dropped. The key id arrives in a header and selects the secret, so an unknown id is rejected before any MAC is computed.

max_age bounds replay in both directions -- a request whose timestamp is too far in the future is refused too, so a clock-skewed or forged forward timestamp cannot buy an attacker an arbitrarily long window. It does not stop a replay inside the window; that is what a replay store or the inbox is for.

Args:

• keys — Key id to secret. At least one, and none may be empty.
• max_age — Seconds a timestamp may differ from now, either way.

Raises:

• ValueError — keys is empty or holds an empty secret, or max_age is non-positive.

HMACWebhookVerifier(keys: Mapping[str, bytes], *, max_age: float = 300.0) -> None

max_age (property, inherited from _NormalizedWebhookVerifier)#

max_age: float

verify (inherited from _NormalizedWebhookVerifier)#

verify(*, body: bytes, headers: Mapping[bytes, bytes], now: datetime | None = None) -> WebhookEnvelope

InboxClaim#

The result of trying to claim an inbound event in the shared inbox.

Four outcomes, and only one of them means "run the handler":

• claimed -- this worker owns the event; process it and complete it.
• duplicate -- it was already processed; replay result_status.
• active -- another worker holds an unexpired lease; answer 409 and let the sender retry rather than processing it twice.
• failed -- a previous attempt recorded a failure and the row is not reclaimable; a human decides what happens next.
• conflict -- the event id was reused for different payload bytes or a different payload version; never process or replay it.

Args:

• fencing_token — Rises on every claim; PostgresWebhookInbox.complete refuses a stale one.
• result_status — The status a completed attempt returned, when one is recorded.

InboxClaim(outcome: "Literal['claimed', 'duplicate', 'active', 'failed', 'conflict']", fencing_token: int, result_status: int | None = None) -> None

LocalReplayStore#

Bounded webhook replay protection in one process.

Enough for a single worker. Behind more than one it is a fast path rather than the guarantee: each worker has its own view, so the same event delivered twice to two workers is claimed twice and handled twice. Use PostgresWebhookInbox when the deduplication has to hold across replicas -- it is the same claim in a table every worker shares.

Doubly bounded, because a replay store that grows without limit is a memory leak an unauthenticated sender controls: entries expire after ttl, and a new claim is refused once max_entries live claims are held. Existing claims are never evicted early, because key spraying must not make a replay acceptable before its verification window closes.

ttl should be at least the verifier's max_age: a request older than that window is rejected on the signature anyway, so a shorter TTL only creates a gap in which a replay is neither too old nor remembered.

Args:

• max_entries — Live claims retained before new claims are refused.
• ttl — Seconds a claim is remembered, on the monotonic clock.

Raises:

• ValueError — Either bound is non-positive.

LocalReplayStore(*, max_entries: int, ttl: float) -> None

claim#

async claim(source: str, event_id: str, *, now: float | None = None) -> bool

Claim (source, event_id), returning whether this caller won it.

The whole check-and-insert happens under one lock, so two concurrent deliveries of the same event cannot both be told they won. Expiry and capacity admission run here, which is why the store needs no sweeper task.

Args:

• source — Namespace, so two senders may use the same event id.
• now — Monotonic reference time. Defaults to time.monotonic().

Returns:

• True if this caller claimed it; False if it was already claimed.

complete#

async complete(source: str, event_id: str, outcome: str) -> None

Record how a claimed event turned out. Does not extend or release the claim.

The claim keeps its original expiry either way, so a handler that failed does not make the event redeliverable -- a retry from the sender is still refused until the TTL passes. Recording the outcome is for observability, not for control flow. A claim that has already expired is silently not updated.

Args:

• outcome — A free-form label; the receiver writes "completed" or "failed".

max_entries (property)#

max_entries: int

size (property)#

size: int

Live claims held right now, expired ones included until they are swept.

ttl (property)#

ttl: float

OutboxDelivery#

One leased outbox row, as the dispatcher sees it. Immutable.

Carries both keys and both is deliberate: delivery_id identifies the attempt record, event_id identifies the event, and a receiver deduplicates on the latter. fencing_token is what every later transition is matched on, so passing a stale copy of this object is refused rather than acted on.

Args:

• delivery_id — Primary key of the outbox row.
• event_id — The envelope id, and the receiver's deduplication key.
• destination — Names the registered WebhookDestination to send through.
• attempts — Attempts made including this one; incremented as the lease was taken.
• fencing_token — Rises on every claim; every transition is matched on it.
• key_id — The signing key recorded at enqueue, so rotation cannot orphan the row.

OutboxDelivery(delivery_id: str, event_id: str, destination: str, event_type: str, timestamp: datetime, version: str, body: bytes, content_type: str, key_id: str, attempts: int, fencing_token: int, ordering_key: str | None = None, correlation_id: str | None = None, causation_id: str | None = None, relay_path: tuple[str, ...] = (), signature_profile: str = 'wreath-v2-hmac-sha256') -> None

PostgresWebhookInbox#

Transactional cross-replica webhook deduplication and fencing.

A row per (source, message_id) in a table every replica shares, so "already handled" is a fact in the database rather than a fact in one process's memory. Claiming, handling and completing happen in the caller's transaction, which is what makes them atomic with whatever the handler writes: an event cannot be marked processed unless its effects committed, and its effects cannot commit unless it was marked processed.

A crashed worker's lease expires and the event becomes claimable again, with the fencing token incremented. The old worker coming back cannot then complete the row over the top of its successor -- complete matches on the token and raises instead.

Args:

• table — Table name. Must be a plain SQL identifier; it is interpolated, not bound.
• session_factory — Opens the session owned by claim_and_enqueue.
• lease_owner — Stable worker identity recorded by an atomic claim.
• lease_seconds — Positive claim lease used by claim_and_enqueue.
• retention_seconds — How long terminal rows remain replay evidence.

Raises:

• TypeError — session_factory is not callable.
• ValueError — table is not a plain SQL identifier, or atomic configuration is invalid.

PostgresWebhookInbox(table: str = 'wreath_webhook_inbox', *, session_factory: Callable[[], AbstractAsyncContextManager[Any]] | None = None, lease_owner: str | None = None, lease_seconds: float | None = None, retention_seconds: float = 2592000) -> None

claim#

async claim(session: Any, *, source: str, envelope: WebhookEnvelope, lease_owner: str, lease_seconds: float) -> InboxClaim

Claim envelope for processing, or report why it cannot be claimed.

One statement does the whole decision: an insert that, on conflict, takes the row over only when the previous lease has actually expired. There is no read-then-write, so two replicas racing on the same event cannot both be told they claimed it -- the loser reads the existing row and gets active.

Runs in the caller's transaction and commits with it. A handler that raises therefore rolls the claim back too, leaving the event redeliverable, which is the behaviour you want when the failure was transient.

Args:

• session — An open session inside a transaction. Not committed here.
• lease_owner — Identifies the claiming worker; recorded on the row.
• lease_seconds — How long the claim is exclusive before another worker may take it.

Returns:

• An InboxClaim; only claimed authorises running the handler.

Raises:

• ValueError — lease_seconds is not positive.
• RuntimeError — The row vanished mid-transaction, which should not be reachable.

claim_and_enqueue#

async claim_and_enqueue(*, source: str, envelope: WebhookEnvelope, enqueue: Callable[..., Awaitable[Any]], result_status: int = 202) -> bool

Atomically claim one delivery, enqueue its work, and complete the claim.

The configured session owns one transaction containing all three writes. enqueue receives that open session as transaction= so a compatible job owner can place its row in the same commit. Existing, active, and failed deliveries return False without invoking it. Any error or cancellation propagates through the transaction context and rolls the claim back.

complete#

async complete(session: Any, *, source: str, message_id: str, fencing_token: int, result_status: int) -> None

Mark a claimed event completed, if this worker still holds the claim.

The fencing check is the point. A worker whose lease expired while it was working has already been superseded, and letting it write completed would mark an event done that its successor is still processing -- exactly the split-brain the lease exists to prevent. It raises instead, and the caller's transaction rolls back with it.

Args:

• session — The same transaction the claim was made in.
• fencing_token — The token from the InboxClaim. A stale one is refused.
• result_status — Status recorded for replay on a later duplicate.

Raises:

• RuntimeError — The lease was taken over, or the row is no longer processing.

component#

component() -> Any

The inbox's claim on the wreath schema.

purge#

async purge(session: Any, *, limit: int = 1000) -> int

Delete up to limit rows past their retention, in the caller's transaction.

One bounded chunk with no cursor, no resumption, and no pacing. It is the right tool when you already hold a session and want a bounded amount of work done right now; for keeping the table small forever, use purge_pass.

Locks with SKIP LOCKED, so two callers running it at once delete disjoint rows instead of blocking on each other.

Args:

• session — An open session inside a transaction. Not committed here.
• limit — Maximum rows deleted in this call.

Returns:

• How many rows were deleted; 0 when nothing was past retention.

Raises:

• ValueError — limit is not positive.

purge_pass#

purge_pass(*, chunk: int = 1000, **options: Any) -> Any

A recurring pass that drops inbox rows past their retention.

The supported way to keep the inbox small:

jobs.drive(inbox.purge_pass(), cron="23 * * * *")

purge has a chunk size and nothing else -- no cursor, so it starts from the beginning of the index every time; no resumption, so a redeploy loses where it was; and no pacing, so it competes with delivery for the same pool. The pass supplies all three, and keeps one transaction per chunk. See wreath.passes.

Only rows with a retention_until in the past are eligible; a row whose retention was never stamped is never purged by either form.

Takes no database: the pass is a declaration, and it is handed one when the scheduler drives it.

Args:

• chunk — Rows per transaction.
• options — Forwarded to the pass -- within, shift, pace, schema.

Returns:

• An unstarted wreath.passes.ChunkedPass; drive it from the scheduler.

retention_seconds (property)#

retention_seconds: float

schema_sql#

schema_sql() -> str

The inbox DDL, semicolon-joined. A derivation of component().

statements#

statements() -> tuple[str, ...]

DDL creating the inbox table and its retention index. Idempotent.

IF NOT EXISTS throughout, so it is safe to run at every boot.

The table name is unqualified, so it lands in whatever search_path resolves to rather than in the wreath schema. That is where every existing deployment's rows already are, and moving them would not be additive. See wreath.schema.Component.qualified.

Returns:

• One statement per element, in order.

table (property)#

table: str

transactional (property)#

transactional: bool

Whether this inbox can own an atomic claim-and-enqueue transaction.

PostgresWebhookOutbox#

Transactional durable intent for a supervised webhook dispatcher.

The transactional outbox pattern. A delivery is inserted in the same transaction as the change that caused it, so the two commit together: there is no state in which the order shipped but the notification was lost, and none in which it was sent for an order that rolled back. Sending happens afterwards, out of band, by WebhookDispatcher.

Rows move through pending -> leased -> sending -> one of delivered, retry_wait, failed, unknown or cancelled. Every transition is fenced on (delivery_id, fencing_token), so a worker whose lease expired mid-send cannot write over the worker that took the row from it -- it raises instead. Claiming uses FOR UPDATE SKIP LOCKED, so workers do not contend.

Because the transport can fail without an answer, unknown is a terminal state distinct from failed: the dispatcher does not retry it, since a request that may already have been processed must not be sent twice on a guess.

Args:

• table — Table name. Must be a plain SQL identifier; it is interpolated, not bound.
• retention_seconds — How long terminal rows remain available for audit and purging.

Raises:

• ValueError — table is not a plain SQL identifier.

PostgresWebhookOutbox(table: str = 'wreath_webhook_outbox', *, retention_seconds: float = 2592000) -> None

claim_due#

async claim_due(session: Any, *, lease_owner: str, lease_seconds: float) -> OutboxDelivery | None

Lease the single most overdue delivery, or return None when none is due.

Two kinds of row are due: one that is pending or retry_wait with its next_attempt_at in the past, and one whose worker died holding it -- leased or sending with an expired lease. Recovering the second kind is why a crashed dispatcher does not strand deliveries.

FOR UPDATE SKIP LOCKED on one row means concurrent dispatchers each get a different delivery rather than blocking, and the attempt counter and fencing token both advance as the lease is taken, so the row this returns is already fenced against its previous owner.

Args:

• session — An open session inside a transaction. Not committed here.
• lease_owner — Identifies the claiming worker; recorded on the row.
• lease_seconds — How long the claim holds before another worker may take it.

Returns:

• The leased delivery, or None when nothing was due.

Raises:

• ValueError — lease_seconds is not positive.

component#

component() -> Any

The outbox's claim on the wreath schema.

enqueue#

async enqueue(session: Any, *, destination: str, envelope: WebhookEnvelope, key_id: str) -> str

Insert one pending delivery in the caller's transaction.

Writes only; nothing is sent here and no connection is opened. The row becomes visible to the dispatcher when the caller's transaction commits, and disappears if it rolls back -- which is the whole point of the pattern.

The signing key id is recorded on the row rather than resolved at send time, so a delivery that sits in the outbox across a key rotation is still signed with the key its receiver was expecting.

Args:

• session — An open session inside a transaction. Not committed here.
• destination — The registered destination name the dispatcher routes on.
• key_id — The signing key id to use when this row is eventually sent.

Returns:

• The generated delivery id, distinct from the event id.

mark_delivered#

async mark_delivered(session: Any, delivery: OutboxDelivery, *, status: int) -> None

Settle a delivery as delivered and release its lease. Terminal.

Args:

• status — The 2xx the peer answered with; recorded for audit.

Raises:

• RuntimeError — The fencing token is stale, or the row is not leased or sending.

mark_failed#

async mark_failed(session: Any, delivery: OutboxDelivery, *, status: int | None, failure: str | None) -> None

Settle a delivery as failed: the peer answered and refused. Terminal.

Reached when the attempt budget is spent, when the status is not retryable, or when the row names a destination this dispatcher does not have. Distinct from unknown because here the peer's answer is known.

Args:

• status — The refusing status, or None when there was no response to record.
• failure — A short failure code, truncated to 256 characters.

Raises:

• RuntimeError — The fencing token is stale, or the row is not leased or sending.

mark_retry#

async mark_retry(session: Any, delivery: OutboxDelivery, *, delay: float, status: int | None, failure: str | None) -> None

Return a delivery to retry_wait, due again after delay seconds.

Not terminal: the row releases its lease and becomes claimable again once next_attempt_at passes. The attempt counter is not touched here -- it advanced when the lease was taken -- so a row cannot be retried forever by a worker that keeps failing before it claims.

Args:

• delay — Seconds from now until the row is due again. May be 0.
• status — The response status that triggered the retry, or None on a transport failure.
• failure — A short failure code, truncated to 256 characters.

Raises:

• ValueError — delay is negative.
• RuntimeError — The fencing token is stale, or the row is not leased or sending.

mark_sending#

async mark_sending(session: Any, delivery: OutboxDelivery) -> None

Move a leased delivery to sending, just before the request goes out.

The state a recovering worker reads as "this may already have reached the peer". Like every transition here it is fenced, so a worker that lost its lease cannot re-enter sending.

Raises:

• RuntimeError — The fencing token is stale, or the row is no longer leased.

mark_unknown#

async mark_unknown(session: Any, delivery: OutboxDelivery, *, failure: str | None) -> None

Settle a delivery as unknown: nobody knows whether it arrived. Terminal.

Deliberately not retried. The request may have been processed by the peer, and resending on that guess is how a transport failure becomes a duplicate charge. These rows are for a human or a reconciliation job to resolve against the receiver.

Args:

• failure — A short failure code, truncated to 256 characters.

Raises:

• RuntimeError — The fencing token is stale, or the row is not leased or sending.

purge#

async purge(session: Any, *, limit: int = 1000) -> int

Delete up to limit settled rows past retention, in the caller's transaction.

One bounded chunk with no cursor, no resumption, and no pacing; see purge_pass for the form that keeps the table small forever.

A delivery still waiting on a retry is never eligible, whatever its retention stamp says -- only delivered, failed, cancelled and unknown rows are rubbish.

Args:

• session — An open session inside a transaction. Not committed here.
• limit — Maximum rows deleted in this call.

Returns:

• How many rows were deleted; 0 when nothing was eligible.

Raises:

• ValueError — limit is not positive.

purge_pass#

purge_pass(*, chunk: int = 1000, **options: Any) -> Any

A recurring pass that drops settled deliveries past their retention.

The supported way to keep the outbox small:

jobs.drive(outbox.purge_pass(), cron="43 * * * *")

Only rows in a settled state are eligible, exactly as purge does it -- a delivery still waiting on a retry is not rubbish. What the pass adds is the cursor, the resumption, and the pacing that a bare LIMIT does not have. See wreath.passes.

Takes no database: the pass is a declaration, and it is handed one when the scheduler drives it.

Args:

• chunk — Rows per transaction.
• options — Forwarded to the pass -- within, shift, pace, schema.

Returns:

• An unstarted wreath.passes.ChunkedPass; drive it from the scheduler.

renew_lease#

async renew_lease(session: Any, delivery: OutboxDelivery, *, lease_seconds: float) -> None

Extend the lease on a delivery that is still sending.

For a slow peer: without renewal a request that outlasts the lease would be handed to a second worker while the first is still waiting on it. Only sending rows renew, and the fencing token must still match, so a worker that has already been superseded cannot claw the row back.

Args:

• session — A session for this renewal, separate from the one carrying the send.
• lease_seconds — New lease length, measured from now.

Raises:

• ValueError — lease_seconds is not positive.
• RuntimeError — The fencing token is stale, or the row is no longer sending.

retention_seconds (property)#

retention_seconds: float

schema_sql#

schema_sql() -> str

The outbox DDL, semicolon-joined. A derivation of component().

statements#

statements() -> tuple[str, ...]

DDL creating the outbox table and its two indexes. Idempotent.

IF NOT EXISTS throughout, plus an ADD COLUMN IF NOT EXISTS for relay_path, so running it against an older deployment upgrades it in place. One index serves the dispatcher's ready-row query and one serves the retention purge -- the latter was added because the chunked pass refuses to walk an unindexed key, which is how a purge that had always been a sequential scan came to light.

The table name is unqualified, so it lands in whatever search_path resolves to rather than in the wreath schema -- where existing rows already are. See wreath.schema.Component.qualified.

Returns:

• One statement per element, in order.

table (property)#

table: str

StandardWebhookVerifier#

Verify the Standard Webhooks HMAC-SHA256 profile.

whsec_ secrets are decoded as their standard base64 payload; raw bytes are accepted for secret stores that already decoded them. Multiple secrets permit rotation, and multiple v1 signatures in the header are checked. Event type/version come from the verified JSON body because the standard header set carries only id, timestamp, and signature.

StandardWebhookVerifier(secrets: bytes | str | tuple[bytes | str, ...], *, max_age: float = 300.0) -> None

max_age (property, inherited from _NormalizedWebhookVerifier)#

max_age: float

verify (inherited from _NormalizedWebhookVerifier)#

verify(*, body: bytes, headers: Mapping[bytes, bytes], now: datetime | None = None) -> WebhookEnvelope

StripeWebhookVerifier#

Verify Stripe's Stripe-Signature t=...,v1=... profile.

StripeWebhookVerifier(secrets: bytes | str | tuple[bytes | str, ...], *, max_age: float = 300.0) -> None

max_age (property, inherited from _NormalizedWebhookVerifier)#

max_age: float

verify (inherited from _NormalizedWebhookVerifier)#

verify(*, body: bytes, headers: Mapping[bytes, bytes], now: datetime | None = None) -> WebhookEnvelope

WebhookContext#

What a receiving handler is told about the event it is handling.

The second argument to a handler is the validated payload; this is the first, and carries everything around it.

session is the distinction that matters. On a durable source it is the same transaction the inbox claim was made in, so anything the handler writes commits atomically with "this event was processed" -- there is no window in which the work landed but the deduplication did not. On a non-durable source it is None and the handler must open its own.

Args:

• source — The registered source name exposed to the handler.
• request — The live request, for headers the envelope does not carry.
• session — The claim's transaction on a durable source, else None.

WebhookContext(source: str, envelope: WebhookEnvelope, request: Request, session: Any | None = None) -> None

WebhookDeliveryResult#

The outcome of one delivery attempt.

Three outcomes, not two, because a transport failure is genuinely not a failed delivery: a request that timed out may well have been processed. Only failed means the peer answered and refused; unknown means nobody knows, which is why the dispatcher settles those rows separately instead of retrying them into a duplicate.

Args:

• outcome — delivered (2xx), failed (a non-2xx answer), or unknown (no answer).
• event_id — The envelope id this attempt was for.
• status — The response status; None when no response arrived.
• failure — A short failure code, e.g. a client exception name; None when the peer answered.

WebhookDeliveryResult(outcome: "Literal['delivered', 'failed', 'unknown']", event_id: str, status: int | None = None, failure: str | None = None) -> None

WebhookDestination#

One receiver: a client, a path on it, a signing key, and optionally an outbox.

Created through WebhookHub.destination. The client is origin-pinned, so the destination owns only the path -- which is why the path must be origin-relative and is refused otherwise.

Two ways to send, and they are not interchangeable. send posts immediately and hands back the outcome, which is right for something that has no durable consequence. enqueue writes the intent into the caller's transaction and returns; the delivery then survives a crash and is sent by WebhookDispatcher with leases and retries. If losing the webhook would leave the receiver's state wrong, it belongs in the outbox.

relay_id names this service in a relayed envelope's path, and the path is signed. It defaults to the destination name.

Args:

• client — An origin-pinned HTTP client, normally a wreath.http_client.HTTPClient.
• path — Origin-relative receiver path on that client. Must start with a single /.
• outbox — Enables enqueue and enqueue_relay. Without it they raise.
• relay_id — This service's name in a relay path. Must match the relay-id syntax.
• max_relay_hops — Hops permitted before a relay is refused. Between 1 and 32.

Raises:

• ValueError — The path is not origin-relative, or relay_id or the hop limit is invalid.

WebhookDestination(name: str, *, client: Any, path: str, signer: HMACWebhookSigner, outbox: PostgresWebhookOutbox | None, relay_id: str, max_relay_hops: int) -> None

enqueue#

async enqueue(session: Any, event_type: str, payload: Any, *, event_id: str | None = None, version: str = '1', timestamp: datetime | None = None, correlation_id: str | None = None, causation_id: str | None = None, ordering_key: str | None = None, relay_path: tuple[str, ...] = ()) -> str

Insert durable delivery intent in the caller's transaction.

Nothing is sent and no connection is opened; this writes a row. The delivery becomes real exactly when the caller's transaction commits and vanishes if it rolls back, so the notification and the change that caused it can never disagree. WebhookDispatcher sends it afterwards.

The signing key id is captured now, from the destination's signer, so a rotation between enqueue and send cannot orphan the row.

Args:

• session — An open session inside a transaction. Not committed here.
• payload — Bytes-like is stored verbatim; anything else is JSON-encoded.
• event_id — Defaults to a fresh UUID4 hex; also becomes the idempotency key.
• ordering_key — Stored for grouping. Advisory -- nothing orders delivery by it.

Returns:

• The delivery id of the new outbox row.

Raises:

• RuntimeError — This destination was configured without an outbox.
• ValueError — The envelope fields or the relay path are invalid.

enqueue_relay#

async enqueue_relay(session: Any, inbound: WebhookEnvelope, event_type: str, payload: Any, *, version: str = '1', timestamp: datetime | None = None, ordering_key: str | None = None) -> str

Commit a loop-protected relay intent in the caller transaction.

The durable form of relay, and the same lineage rules apply: the outbound event gets a new id, inherits inbound's correlation id (or its id when it starts the chain), records inbound.id as its cause, and appends this service to the signed relay path.

Args:

• session — An open session inside a transaction. Not committed here.
• inbound — The verified envelope this event is caused by.

Returns:

• The delivery id of the new outbox row.

Raises:

• ValueError — This service is already in the relay path, or the hop limit is reached.
• RuntimeError — This destination was configured without an outbox.

relay#

async relay(inbound: WebhookEnvelope, event_type: str, payload: Any, *, version: str = '1', timestamp: datetime | None = None) -> WebhookDeliveryResult

Emit a separately identified event caused by an inbound event.

A relay is a new event, not a forward: it gets its own id and its own signature, and the receiver deduplicates it independently. What connects it to its cause is the correlation id (inherited, or inbound.id when this starts the chain), the causation id, and the relay path.

The relay path is signed and checked before anything is sent, which is what makes a cycle of services impossible rather than merely unlikely: a service already in the path refuses, and so does a path at the hop limit.

Args:

• inbound — The verified envelope this event is caused by.

Returns:

• The outcome of the immediate send.

Raises:

• ValueError — This service is already in the relay path, or the hop limit is reached.

send#

async send(event_type: str, payload: Any, *, event_id: str | None = None, version: str = '1', timestamp: datetime | None = None, correlation_id: str | None = None, causation_id: str | None = None, relay_path: tuple[str, ...] = ()) -> WebhookDeliveryResult

Sign and post one event now, returning the outcome rather than raising.

A transport failure is reported, not raised: ClientError becomes an unknown result, because a request that failed without an answer may still have been processed. A non-2xx answer becomes failed. Nothing here retries, and nothing here is durable -- if the process dies between the caller's commit and this call, the event is simply gone. Use enqueue when that matters.

The event id doubles as the client's idempotency key, so a receiver that honours Idempotency-Key collapses a duplicate on its own.

Args:

• payload — Bytes-like is sent verbatim; anything else is JSON-encoded.
• event_id — Defaults to a fresh UUID4 hex. Supply one to make the send repeatable.
• version — Payload schema version, signed and sent.
• timestamp — Defaults to now, UTC. Must be timezone-aware.
• relay_path — Prior hops. Prefer relay, which builds this correctly.

Returns:

• The outcome -- delivered, failed, or unknown.

Raises:

• ValueError — The envelope fields or the relay path are invalid.

WebhookDispatcher#

Fenced delivery loop with lease renewal and lifespan supervision hooks.

Drains one outbox, routing each row to the destination it names. One dispatcher per worker; several may run against the same table without coordinating, because claiming uses FOR UPDATE SKIP LOCKED and every transition is fenced on the row's token.

The loop is deliberately serial: one delivery at a time, per dispatcher. Throughput comes from running more workers rather than more concurrency inside one, which keeps the ordering and the lease reasoning simple.

Retries are bounded by max_attempts and back off exponentially from retry_delay. Only a status in retry_statuses is retried; a transport failure with no answer settles as unknown and is not retried, because the peer may already have processed it.

Args:

• outbox — The table to drain.
• destinations — Name to destination. A row naming an absent one settles as failed.
• worker_id — Identifies this worker on a lease. Cannot be empty.
• lease_seconds — Claim length; the lease is renewed at a third of this while sending.
• max_attempts — Attempts before a retryable failure settles as failed.
• retry_delay — Base backoff in seconds, doubled per attempt already made,
• capped at retry_cap and jittered ±20% by wreath._jobcore.compute_backoff
• -- the same calculation the job runner and the message bus retry on.
• retry_cap — Longest a retry may be deferred, however many attempts precede it.
• retry_statuses — Response statuses treated as retryable.

Raises:

• ValueError — worker_id is empty, or a lease, attempt, or delay bound is invalid.

WebhookDispatcher(outbox: PostgresWebhookOutbox, destinations: Mapping[str, WebhookDestination], *, worker_id: str, lease_seconds: float = 30.0, max_attempts: int = 6, retry_delay: float = 1.0, retry_cap: float = 3600.0, retry_statuses: frozenset[int] = frozenset({500, 408, 502, 503, 504, 425, 429})) -> None

manage#

manage(app: Any, session_factory: Callable[[], AbstractAsyncContextManager[Any]], *, idle_delay: float = 0.25) -> None

Attach the delivery loop to the application's lifespan.

Registers the loop with the application's ordinary service supervisor, so startup, bounded drain, task ownership, and failure accounting use the same lifecycle as jobs, messaging, entities, and WebSockets. A loop that fails immediately is re-awaited during startup, turning "never started" into a failed boot instead of a quiet absence.

The dispatcher is also published on app.state as webhook_dispatcher_<worker_id> (non-identifier characters replaced with _), so a health endpoint can find it without a global.

A delivery loop that dies stays dead until the process restarts, with readiness reporting it. Call this once per dispatcher.

Args:

• app — The application whose lifespan and state this binds to.
• session_factory — Opens a session per loop iteration and per lease renewal.
• idle_delay — Seconds to wait before re-polling when the outbox is empty.

Raises:

• RuntimeError — This dispatcher is already managed.

readiness (property)#

readiness: DispatcherReadiness

A DispatcherReadiness snapshot. Synchronous, never blocks.

last_error is not cleared by reading it; a new run clears it.

run#

async run(session_factory: Callable[[], AbstractAsyncContextManager[Any]], stopping: asyncio.Event, *, idle_delay: float = 0.1) -> None

Run until stopped; the application supervisor owns this coroutine.

Each iteration opens a session, attempts one delivery, and closes it. An empty outbox waits on stopping for up to idle_delay rather than sleeping, so shutdown is prompt instead of costing a poll interval.

Failures are not absorbed. Anything the loop raises is recorded in readiness and re-raised unchanged, because a delivery loop that keeps running while silently failing is worse than one that stops visibly. CancelledError propagates untouched.

Args:

• session_factory — Opens one session per iteration; also used for lease renewal.
• stopping — Set it to end the loop after the current delivery settles.
• idle_delay — Seconds to wait before re-polling when the outbox is empty.

Raises:

• ValueError — idle_delay is not positive.
• Exception — Whatever a delivery or the database raised, after recording it.

run_once#

async run_once(session: Any, *, renewal_session_factory: Callable[[], AbstractAsyncContextManager[Any]] | None = None) -> WebhookDeliveryResult | None

Claim, send, and settle at most one delivery. The unit run repeats.

Returns None when nothing is due, which is the caller's signal to idle. Otherwise it claims one row, marks it sending, posts it, and settles it into exactly one terminal or retry state before returning -- a row is never left leased by a successful call.

Directly useful in tests and in a one-shot drain: it needs no event loop of its own and no lifespan.

Lease renewal is opt-in via renewal_session_factory. Without it a response slower than lease_seconds lets another worker claim the row while this send is still outstanding, and the settle then fails on the fencing token. Pass a factory whenever peers can be slow. The renewal task is cancelled and awaited in a finally, so it cannot outlive the send it was protecting.

Args:

• session — An open session inside a transaction. Not committed here.
• renewal_session_factory — Opens a separate session to renew the lease while sending.

Returns:

• The attempt's outcome, or None when no delivery was due.

Raises:

• RuntimeError — The lease was taken over mid-send, so the settle was refused.

WebhookEnvelope#

One event, in the form the signature covers. Immutable; validated at construction.

Both directions use it: a signer turns one into headers, and a verifier returns one after the MAC checks out. Holding the raw body rather than a decoded payload is deliberate -- the signature is over those exact bytes, and re-encoding them would break it.

Construction refuses anything that would make the signature base ambiguous or the relay path unusable, so an envelope that exists is one that can be signed.

Args:

• id — Unique per event, and the deduplication key on the receiving side.
• type — Event type; selects the handler on the receiving side.
• version — Payload schema version, carried through and signed.
• timestamp — Must be timezone-aware. Bounds replay via the verifier's window.
• content_type — Sent as Content-Type; not covered by the signature.
• body — The exact bytes signed and sent.
• correlation_id — Ties an event to the chain it belongs to.
• causation_id — The id of the event that caused this one.
• ordering_key — Advisory grouping stored on an outbox row. Not signed, not enforced.
• relay_path — Services this event has already passed through. Signed, and loop-checked.
• deduplication_id — Authenticated identity used by replay stores when provider event ids
• are not signed. Defaults to id.

Raises:

• ValueError — An empty or control-character field, a naive timestamp, or a bad relay path.

WebhookEnvelope(id: str, type: str, version: str, timestamp: datetime, content_type: str, body: bytes, correlation_id: str | None = None, causation_id: str | None = None, ordering_key: str | None = None, relay_path: tuple[str, ...] = (), deduplication_id: str | None = None) -> None

WebhookHub#

The registry an application declares its webhook sources and destinations on.

Reached as app.webhooks(name) rather than constructed directly. Its job is to be the one place that knows every receiver route, which is what makes csrf_exempt answerable at all: a webhook is authenticated by its signature, not by a browser-origin token, so its route must be exempt from CSRF -- and an exemption is only safe when the set of exempt paths is closed and known.

Names are unique in both directions and a duplicate raises, so a second registration cannot quietly shadow the first.

Args:

• app — The application to register receiver routes on.
• name — Identifies this hub. Cannot be empty.

Raises:

• ValueError — name is empty.

WebhookHub(app: Any, name: str) -> None

csrf_exempt#

csrf_exempt(request: Request) -> bool

Return true only for registered unsafe webhook receiver routes.

Pass it to the CSRF middleware as its exemption predicate. It is narrow on purpose: POST only, and only for a path this hub registered as a source. A machine sender cannot hold a CSRF token, and the request is already authenticated by its HMAC signature, so the exemption costs nothing -- but it must not extend one path further than that.

Returns:

• True only for a POST to a registered source path.

destination#

destination(name: str, *, client: Any, path: str, signer: HMACWebhookSigner, outbox: PostgresWebhookOutbox | None = None, relay_id: str | None = None, max_relay_hops: int = 8) -> WebhookDestination

Register one receiver to send to.

name is what an outbox row stores, so a dispatcher routes on it -- renaming a destination orphans the rows already enqueued under the old name, which settle as failed with UnknownDestination.

Args:

• name — Unique per hub. Stored on outbox rows and used as the default relay id.
• client — An origin-pinned HTTP client for the receiver's origin.
• path — Origin-relative receiver path. Must start with a single /.
• signer — Holds the signing keys and the default key id.
• outbox — Enables durable enqueueing. Without it only the immediate send works.
• relay_id — This service's name in a relay path. Defaults to name.
• max_relay_hops — Hops permitted before a relay is refused. Between 1 and 32.

Returns:

• The registered destination.

Raises:

• ValueError — The name is already registered, or the destination config is invalid.

schema_owners (property)#

schema_owners: tuple[Any, ...]

The inbox and outbox stores this hub's sources and destinations hold.

A hub owns no tables itself; its stores do, and they are optional -- a hub with only a signing destination and no durable outbox has none. So the application asks the hub which of its parts have a claim rather than assuming a fixed pair, and a source registered without an inbox contributes nothing to bootstrap.

source#

source(name: str, *, path: str, verifier: WebhookVerifier, replay: LocalReplayStore | None = None, limits: WebhookLimits = WebhookLimits(max_body_bytes=1048576, max_headers=32, max_header_bytes=16384, max_event_id_bytes=256), inbox: PostgresWebhookInbox | None = None, session_factory: Callable[[], AbstractAsyncContextManager[Any]] | None = None, lease_owner: str = 'webhook-receiver', lease_seconds: float = 30.0) -> WebhookSource

Register one sender's receiver route and record it as CSRF-exempt.

The supported way to build a WebhookSource: constructing one directly registers the route but leaves this hub -- and therefore csrf_exempt -- unaware of the path.

Args:

• name — Unique per hub. Combined with the hub name for replay isolation.
• path — Route path for the receiver, registered on the application now.
• verifier — Holds the sender's keys and the replay window.
• replay — In-process replay store; one sized to the verifier's window is made if None.
• inbox — Cross-replica deduplication. Requires session_factory.
• session_factory — Opens the session the claim and handler share. Requires inbox.
• lease_owner — Identifies this worker on an inbox claim.
• lease_seconds — Inbox claim lease length.

Returns:

• The registered source, to hang WebhookSource.event handlers on.

Raises:

• ValueError — The name is already registered, or the source configuration is invalid.

WebhookLimits#

What one inbound webhook request may cost. Validated at construction.

Applied before the signature is checked, and that order matters: verifying a MAC over a body an unauthenticated caller controls the size of is work the caller chose for you. Every breach answers 413 -- except an oversized event id, which answers 400, because it is a valid signed request that this receiver will not store.

Args:

• max_body_bytes — Body ceiling, checked once the body is read and before the MAC.
• max_headers — Header fields accepted; over it the request is refused unread.
• max_header_bytes — Total header name+value bytes, accumulated as they are scanned.
• max_event_id_bytes — UTF-8 bytes in the event id, checked after verification.

Raises:

• ValueError — Any limit is non-positive.

WebhookLimits(max_body_bytes: int = 1048576, max_headers: int = 32, max_header_bytes: int = 16384, max_event_id_bytes: int = 256) -> None

WebhookSource#

One sender's webhooks: a route, a verifier, and a handler per event type.

Created through WebhookHub.source, which also records the path for CSRF exemption -- constructing one directly registers the route but leaves the hub unaware of it. Construction registers a POST route immediately, so it must happen while the application is still accepting routes.

A request is processed in a fixed order, and the order is the security property: bound the request, verify the signature, bound the event id, resolve the handler, deduplicate, then run. Nothing an unauthenticated caller sends reaches a handler, and nothing expensive happens before the MAC checks out.

Responses are deliberately terse and carry no detail: 401 for any verification failure, 413 for anything over a limit, 400 for an unregistered event type, an over-long id, or a body that is not JSON, 409 when another worker holds the event, and 204 on success. A sender learns whether to retry and nothing else.

A body that passes the signature check and then fails to parse is the sender's error, not this service's: the MAC proves they sent exactly those bytes. It answers 400, so a retry of the same bytes is not invited.

Deduplication is whichever of the two was configured. With inbox and session_factory the handler runs inside the claim's transaction, so its writes and the "processed" record commit together. Without them the claim is the in-process LocalReplayStore, which is a fast path rather than a guarantee behind more than one worker.

Args:

• path — Route path for the receiver. Registered on app at construction.
• replay — In-process replay store; one sized to the verifier's window is made if None.
• inbox — Cross-replica deduplication. Requires session_factory.
• session_factory — Opens the session the claim and handler share. Requires inbox.
• lease_owner — Identifies this worker on an inbox claim. Only used with inbox.
• lease_seconds — Inbox claim lease. Only used with inbox.

Raises:

• ValueError — inbox and session_factory were not both given, or the lease is invalid.

WebhookSource(app: Any, name: str, *, path: str, verifier: WebhookVerifier, replay: LocalReplayStore | None, limits: WebhookLimits, inbox: PostgresWebhookInbox | None, session_factory: Callable[[], AbstractAsyncContextManager[Any]] | None, lease_owner: str, lease_seconds: float, replay_namespace: str | None = None) -> None

event#

event(event_type: str, *, payload: Any) -> Callable[[WebhookHandler], WebhookHandler]

Decorator registering the handler for one event type.

The handler is called as handler(context, payload) with the payload already decoded and validated against payload; a request whose body does not fit raises out of the receiver rather than reaching the handler. An event type with no registered handler is answered 400, so a source never silently ignores an event it was sent.

The handler's return value is discarded -- success is signalled by returning, failure by raising. Raising on a durable source rolls the inbox claim back with the handler's own writes; on a non-durable source the replay claim stands, so the sender's retry is refused until the TTL passes.

Args:

• event_type — Matched against the wreath-webhook-type header. Cannot be empty.
• payload — A validation target; the same shapes a route body annotation accepts.

Returns:

• The registering decorator, which returns the handler unchanged.

Raises:

• ValueError — event_type is empty, or already has a handler.

WebhookVerifier#

A provider signature profile consumed by WebhookSource.

The hub has already lower-cased and bounded the headers before calling the verifier. A verifier returns a complete trusted envelope or raises ValueError; boolean and partially verified results are not part of the contract. max_age also sizes the default in-process replay ledger.

WebhookVerifier(*args, **kwargs)

verify#

verify(*, body: bytes, headers: Mapping[bytes, bytes], now: datetime | None = None) -> WebhookEnvelope

wreath.signatures#

HTTP Message Signatures (RFC 9421), in both directions, at the boundary.

One module, because signing and verifying share the signature base -- the canonical byte string both sides hash -- and two implementations of that string disagreeing is where every interoperability failure in this protocol comes from. _signature_base is written once and called by both.

from wreath import Wreath
from wreath.signatures import Signatures

app = Wreath(signatures=Signatures(
    directories=("https://openai.com/.well-known/http-message-signatures-directory",),
    max_age=60.0,
))

The boundary establishes a fact; the policy set decides what it means. This module never allows or denies. Verification publishes signature_verified and signature_agent for Signatures.cedar_context, and a Cedar policy says what an unverified caller may reach:

permit (principal, action == Action::"read", resource is Article) unless { context.signature_verified == false && resource.paid };

Verified is not trusted. A valid signature proves which agent sent the request. It says nothing about whether that agent is welcome, and conflating the two rebuilds the User-Agent allow-list with extra steps and more code.

What "this request" means, exactly#

A signature covers the components the caller listed, and anything it does not list rides along unauthenticated. That is the protocol working as designed, and it is why the required set here is not the minimum the RFC permits:

• @method, @authority, @path and @query are demanded of every signature. Without the query the signature is over an endpoint, so ?x=1 and ?admin=1 produce identical bases and anyone who observes one signed request replays it with the parameters of their choosing.
• content-digest is demanded of any request that carries a body, and the digest is recomputed from the bytes. Covering the header alone is worth nothing: it canonicalizes as a string like any other header, so a signature over it proves only that the sender typed it. The body has not arrived when this middleware runs, so the expectation is parked on request.state and Request.body()/Request.stream() spend it -- a mismatch is a 400 there.

SignatureFacts.covered and the signature_covered context key publish the component list, so a policy can require an end-to-end signature (context.signature_covered.contains("content-digest")) instead of reading signature_verified and assuming what it covered.

What it does not do on the request path#

It never fetches a key. Key material comes from a directory refreshed off the request path (lifespan startup, then refresh() on whatever schedule the deployment chooses). A signature naming a key this process has not got is simply unverified -- the fact is absent and the policy decides, exactly as for a request that carried no signature at all.

That is a deliberate refusal, not an omission. Resolving an unknown keyid by fetching turns one inbound request from an unauthenticated crawler into one outbound request to a host the crawler named, which is an amplifier with a signature header for a trigger.

It is also why this does not reuse wreath._auth.jwks.JwksCache, which is otherwise the obvious candidate. That class refreshes on an unknown kid -- correct for a JWT verifier, whose callers hold a token from a provider it already trusts, and exactly the amplifier here, whose callers are anonymous. JwksCache guards it with a single-flight lock and a negative cache; this module does not need a guard because it does not have the behaviour. What is reused is the parsing: wreath._auth.jwt.key_from_jwk reads the JWK and wreath._auth._ecverify.verify_ed25519 checks the signature, so there is no second JWK reader and no second Ed25519 implementation in the tree.

Profile versioning#

Web Bot Auth is an application profile of RFC 9421, standardised through an IETF working group chartered in 2026, with documents still moving through 2026-27. Signatures(profile=...) names which profile is being enforced and WEB_BOT_AUTH_2026 is the only one implemented; the well-known directory path is expected to change, so it is a constant here rather than a literal spread through the code.

Relationship to wreath.webhooks#

Both sign and verify HTTP requests, and they are deliberately not one profile. webhooks is a symmetric HMAC over an exact body with wreath's own versioned base and a key id chosen by the sender; this is asymmetric Ed25519 over a component list the caller declares per request. Neither base can be expressed in the other's terms, so folding them into one signer would produce a function with two disjoint halves and a mode flag.

Their replay ledgers stay separate too, and that one is a near-miss worth naming: webhooks.LocalReplayStore and NonceLedger are both bounded TTL claim ledgers. They differ in the property that matters -- a full store evicts there and refuses here (see NonceLedger for why) -- and in shape: one is async for a claim taken under a lock, this one is synchronous because it runs on the ingress path where an await is not available. Merging them would hand one caller a lock it cannot take or the other a race it does not have. The shared part that was extractable already exists and is used: wreath.kv.KV is the bounded, lazily-expiring table underneath this one. LocalReplayStore predates it and still hand-rolls a heap; converting it would change webhook eviction order, which is a behaviour change for a live subsystem and belongs in its own change with its own measurement.

NonceLedger#

Bounded single-use nonce memory that refuses when it is full.

The one semantic difference from webhooks.LocalReplayStore, and the reason this is not that class with a flag. A webhook sender has already proved a shared secret before its event id reaches the replay store, so evicting under load costs a duplicate delivery from a party you authenticated. Here the nonce space is controlled by an unauthenticated caller: eviction under load is an attacker flushing the ledger and replaying whatever it displaced, which is the exact attack the ledger exists to stop.

So a full ledger refuses the request instead. That is a denial-of-service trade made deliberately in the safe direction, it is counted rather than silent (refusals), and wreath doctor has a number to look at. Size max_entries for the burst you expect.

One worker's memory. Two workers each remember their own nonces, so a replay routed to the other worker is not caught. That is a fast path, not a guarantee, and it is why the timestamp window is the primary bound and this is the secondary one.

Args:

• max_entries — Nonces retained. Reaching it refuses rather than evicts.
• ttl — Seconds a nonce is remembered. Should be at least max_age, or a
• replay is neither too old nor remembered.

Raises:

• ValueError — Either bound is non-positive.

NonceLedger(*, max_entries: int = 16384, ttl: Any = 300.0) -> None

claim#

claim(nonce: str, *, now: float | None = None) -> bool

Whether nonce is fresh. Records it when it is.

Returns:

• True if this is the first time; False if it was seen, or if the
• ledger is full -- the caller cannot tell those apart, and must not,
• because both mean "this request is not verified".

seen#

seen(nonce: str, *, now: float | None = None) -> bool

Whether nonce is already spent. Records nothing, counts a replay.

The read half of claim, split out so a verifier can refuse a replay before paying for a signature check without letting an unverified caller write to the ledger. A full ledger is not a hit here: refusing on fullness is claim's job, and doing it in the lookup would restore the exhaustion this split exists to remove.

size (property)#

size: int

Live nonces, expired ones excluded.

PaymentRequired#

Answer an agent with a price rather than a wall.

A response shape, deliberately not a payment integration. Pay-Per-Crawl normalised answering an unpaid agent with a 402 and terms; four competing agent-payment protocols were still live as of mid-2026 (x402 under a Linux Foundation body from July 2026, Google's AP2, Stripe's MPP, ACP), and blessing one in a framework would age badly within a year. scheme is the application's choice and settlement is entirely its business.

if not paid(request):
    raise PaymentRequired(
        amount="0.002", currency="USD", pay_to="https://pay.example/x"
    )

An HTTPException, so it travels the ordinary error boundary, renders as an RFC 9457 problem document like every other refusal, and gets its challenge header copied on by the same code that copies a 401's -- rather than a second response path that would have to learn all of that again.

PaymentRequired(*, amount: str, currency: str, pay_to: str, scheme: str = 'http-402', detail: str = 'This resource requires payment.') -> None

terms#

terms() -> bytes

The Accept-Payment challenge, as a structured-field item.

SignatureError#

A signature was present and did not verify.

Raised by the signing and verifying helpers. The middleware never lets one escape: at ingress a failed verification is absence of the fact, not a refusal, because refusing is the policy set's job.

SignatureFacts#

What verification established about one request. Never a decision.

Attributes: verified: A signature was present and verified against a known key. agent: The Signature-Agent the caller named, when it did. key_id: The keyid the signature named. reason: Why an unverified request was not verified. For logs and doctor, never for the client -- telling a caller which check failed hands them an oracle. covered: The component identifiers the signature covered, lowercased and in the order the caller listed them. A policy that cares what was signed has to be able to ask: verified alone says a signature checked out, not what it was a signature over, and the two answers differ by exactly the components a caller chose to leave out.

SignatureFacts(verified: bool = False, agent: str | None = None, key_id: str | None = None, reason: str | None = None, covered: tuple[str, ...] = ()) -> None

SigningKey#

A private Ed25519 key, for the outbound direction.

sign is supplied by the caller rather than implemented here: wreath's zero-dependency crypto is verify-only (wreath._auth._ecverify says so in its first paragraph, and the reason is that verification has no nonce and no private key to get wrong). A deployment that signs outbound requests has a signing key somewhere already -- in cryptography, in an HSM, in a KMS -- and this takes the callable it already has.

Args:

• key_id — The keyid parameter receivers will see, and look up.
• sign — bytes -> bytes, producing a 64-byte Ed25519 signature.
• agent — The Signature-Agent value, i.e. where this key is published.

SigningKey(key_id: str, sign: Callable[[bytes], bytes], agent: str | None = None) -> None

Signatures#

Verify inbound RFC 9421 signatures at ingress; publish the outcome.

Registered as global middleware, so it covers route misses, static files and authorization failures alike -- which is the point, because the traffic this is for is trying to make the server pay for request construction before anything says no.

Args:

• directories — Well-known Web Bot Auth directory URLs, https only.
• max_age — Seconds a created timestamp may differ from now, either way.
• Bounding the future too, so a forged forward timestamp cannot buy an
• arbitrarily long replay window.
• profile — The application profile to enforce. Only WEB_BOT_AUTH_2026.
• required — Components a signature must cover to be accepted.
• nonces — The replay ledger, or None to rely on max_age alone.
• refresh_on_startup — Fetch the directories during lifespan startup.
• Turn it off for a deployment that refreshes on its own schedule, or
• a test that installs keys directly -- it is the difference between
• App(signatures=...) reaching the network at boot and not.
• clock — A time source, for tests.

Raises:

• ValueError — max_age is non-positive, profile is unknown, or a
• directory URL is not https.

Signatures(*, directories: Iterable[str] = (), max_age: Any = 60.0, profile: str = 'web-bot-auth-2026', required: Sequence[str] = ('@method', '@authority', '@path', '@query'), nonces: NonceLedger | None = None, refresh_on_startup: bool = True, clock: Callable[[], float] | None = None) -> None

before#

async before(request: Any) -> None

Compatibility wrapper; the compiled tape uses before_sync.

before_sync#

before_sync(request: Any) -> None

Resolve the facts for this request. Never short-circuits.

Synchronous and non-blocking by construction: every input is already in memory, because key resolution never touches the network here.

A request with no Signature-Input costs one lookup in the header index the request already built for CSRF and the auth backend, and returns.

cedar_context#

cedar_context(request: Any) -> dict[str, object]

The facts, shaped for a Cedar context.

Compose into an application's context provider:

def context(request):
    return {"method": request.method, **signatures.cedar_context(request)}

authorizer = CedarAuthorizer(engine, context=context)

signature_verified is always present and always a boolean, so both when and unless shapes read the same on an unsigned request; signature_agent is absent when there is none, so a policy testing it with has fails closed rather than matching an empty string.

signature_covered is the covered component set, present whenever the signature verified. A policy that needs a request signed end to end -- context.signature_covered.contains("content-digest") -- can say so, rather than reading signature_verified and hoping.

directories (property)#

directories: tuple[str, ...]

The configured directory URLs.

facts#

facts(request: Any) -> SignatureFacts

This request's verification outcome, resolved once per request.

Cached on request.state the way wreath._auth.cedar.request_flags is, and for the same reason: several policies asking one question inside one decision must get one answer.

install#

install(url: str, document: Mapping[str, Any]) -> int

Install a fetched directory document. Returns usable key count.

Separate from fetching so the refresh transport is the deployment's choice and the tests need no network.

Raises:

• KeyError — url is not a configured directory.

refresh#

async refresh(client_factory: Any = None) -> int

Fetch every configured directory. Returns keys installed.

Off the request path by construction. Call it from lifespan startup -- App(signatures=...) does -- and from a schedule if the deployment wants rotation picked up sooner than a restart.

A directory that fails to refresh leaves its previous keys in place and increments refresh_errors: dropping them would turn a transient network fault into every agent silently becoming unverified.

crawler_policy#

crawler_policy(app: Any) -> CrawlerPolicy

Split an application's routes into public and not.

"Public" is AuthRequirement.access_level == 0 -- the framework's own single definition of "this endpoint asks nothing of the caller", already read by both dispatchers, by Wreath._authorize_request and by MCP. Using it here rather than a second predicate is the whole point: a hand-maintained robots.txt drifts from the route table on the first refactor, and this one cannot.

A path with parameters is reduced to its static prefix, because robots.txt has no notion of {id} and a literal /photos/{id} matches nothing.

llms_txt#

llms_txt(app: Any, *, title: str, summary: str | None = None) -> str

An llms.txt body naming what this application offers a model.

The same derivation as robots_txt and the same honesty: a declaration, not an enforcement. Public routes carrying a summary become the listing, because a route with no description is not something a model can use.

robots_txt#

robots_txt(app: Any, *, sitemap: str | None = None, crawl_delay: int | None = None) -> str

A robots.txt body derived from app's routes.

An honour-system control, and worth saying so where somebody will read it: a crawler that ignores this is not stopped by it. The enforced half is a signature check and a policy. This is for the crawlers that do behave, and its value is that it cannot drift from what the application actually serves.

sign_request#

sign_request(key: SigningKey, *, method: str, url: str, headers: Mapping[str, str] | None = None, body: bytes | None = None, components: Sequence[str] = ('@method', '@authority', '@path', '@query'), created: int | None = None, expires_in: int | None = None, nonce: str | None = None, label: str = 'sig1', tag: str | None = None) -> dict[str, str]

Sign an outbound request. Returns the headers to add.

The mirror of verification, over the same signature_base, which is what makes the pair testable against each other and against the RFC's vectors.

headers = sign_request(key, method="GET", url="https://api.example.com/v1/x")
await client.get("/v1/x", headers=headers)

Pass body when there is one. It is hashed into a Content-Digest header, which is added to both the returned headers and the covered set, and that is the only thing that makes the signature a signature over the payload rather than over the endpoint. A verifier configured with this module's default required set refuses a bodied request that omits it.

Args:

• body — The request body, hashed into Content-Digest and covered.
• components — What to cover. Defaults to the required set; content-digest
• is appended when body is given and is not already listed.

Raises:

• SignatureError — A component cannot be built from the given request.
• ValueError — url is not absolute.

signature_base#

signature_base(message: RequestMessage, components: Sequence[tuple[str, Mapping[str, Any]]], params: Mapping[str, Any]) -> bytes

The canonical bytes a signature covers, per RFC 9421 §2.5.

One line per covered component -- the quoted, lowercased identifier with its parameters, then ": ", then the value -- and a final "@signature-params" line carrying the serialized component list and the signature parameters. Lines are joined with \n and there is no trailing newline.

Pinned against the RFC's own Ed25519 vector (§B.2.6) in tests/test_signatures_base.py, byte for byte, which is the only way to know a base builder agrees with anyone else's.

Raises:

• SignatureError — An unknown component, an unknown component parameter, a
• covered header the message does not carry, or a duplicated component.

wreath.digest#

RFC 9530 integrity fields and algorithm preference negotiation.

Digest#

One parsed or computed Content-Digest/Repr-Digest field value.

Digest(values: Mapping[str, bytes]) -> None

algorithms (property)#

algorithms: tuple[str, ...]

Algorithms carried by the field, in wire order.

compute (classmethod)#

compute(content: bytes, *algorithms: str) -> Digest

Hash content with each named active algorithm and serialize the field.

expectation#

expectation() -> tuple[str, bytes]

The strongest active algorithm this field carries.

parse (classmethod)#

parse(value: bytes | str) -> Digest

Parse a bounded RFC 9530 integrity field.

verify#

verify(content: bytes) -> str

Verify content against the strongest supported member and name it.

DigestError#

An integrity field is malformed, unusable, or does not match its data.

DigestPreferences#

Want-Content-Digest or Want-Repr-Digest weighted preferences.

DigestPreferences(weights: Mapping[str, int]) -> None

parse (classmethod)#

parse(value: bytes | str) -> DigestPreferences

Parse a bounded RFC 9530 integrity preference field.

preferred#

preferred(*supported: str) -> str | None

Highest acceptable weight, breaking ties by server preference order.

wreath.xml#

A strict XML reader for documents that carry a signature.

Wreath owns this because the alternatives are not available to it: defusedxml and xmlsec are third-party runtime dependencies, which src/wreath does not take, and the stdlib has no exclusive canonicalization -- ElementTree ships C14N 2.0, a different algorithm from the Exclusive XML Canonicalization 1.0 that XML Signature actually requires.

This is a profile, not an XML parser#

It refuses far more than it accepts, and the refusals are the feature:

<!DOCTYPE, in any form. The only way to declare an entity, so the only route to an expansion bomb or an external resolver. Removing it removes XXE and the billion-laughs family outright.

Entity references beyond the five predefined. Nothing else is declarable, so anything else is a caller expecting an expansion that will not happen.

Comments. A comment splits a text node. Two readings of one value is how a <NameID> gets truncated and a login lands as the wrong person.

CDATA sections. A second spelling of text, and one value with two spellings is an ambiguity a signature cannot settle.

Processing instructions. No processor exists for them here, so they can only be a channel to one that does.

A byte order mark. The bytes a signature covers must be the bytes that were parsed.

Any encoding but UTF-8, and malformed, overlong or surrogate UTF-8. One byte sequence, one meaning: declared-versus-actual encoding mismatch is a whole family of confusion attacks.

Unbounded depth, size, element and attribute counts. An unauthenticated boundary reads whatever it is given.

Everything it accepts, it accepts the same way every time: tests/test_xml_parse.py, test_xml_refusals.py, test_xml_c14n.py and test_xml_wrapping.py hold the parser to the whole corpus, including every exploit, and test_xml_fuzz.py drives it with input nobody designed.

Byte provenance is the point#

Every element records the byte range it was parsed from, and canonicalization re-reads those bytes rather than re-serializing the tree:

from wreath.xml import parse

doc = parse(raw)
assertion = doc.find_id("_a1")        # refuses if two elements claim "_a1"
signed = doc.canonicalize(assertion)  # exclusive c14n of the original bytes
subject = assertion.children[0].children[0].text

A verifier canonicalizes assertion and a consumer reads values out of that same element. There is no second lookup to divert, so XML Signature Wrapping -- the SAML vulnerability class, where the verifier checks one subtree and the consumer reads another -- has nowhere to happen. A repeated identifier is a refusal rather than a resolution order, which closes the other half.

What this is not#

It does not verify signatures, and it knows nothing about SAML. It produces the canonical bytes a signature is computed over and the provenance that makes the result meaningful; the crypto is wreath._auth's and the assertion semantics belong to the layer above.

Document#

A parsed document, and the bytes it was parsed from.

Document(root: Element, source: bytes, canonicalizer: Callable[..., bytes] | None = None) -> None

canonicalize#

canonicalize(element: Element | None = None, inclusive_prefixes: Sequence[str] = ()) -> bytes

Exclusive XML Canonicalization 1.0 of element (default: root).

inclusive_prefixes is the InclusiveNamespaces PrefixList: those prefixes are rendered when in scope even if the subtree does not visibly utilize them. #default names the default namespace.

The canonical form is computed by re-reading the subtree's own source bytes, not by re-serializing the tree this document already holds. That is what makes the bytes a signature covers and the bytes a consumer reads provably the same: there is no reconstruction step in between for the two to disagree across.

find_id#

find_id(value: str) -> Element | None

The element whose ID is value, or None.

Refuses when more than one element carries the identifier. A document with two candidates for one ID is the precondition for every signature- wrapping attack, so it is rejected rather than resolved by document order.

subtree_bytes#

subtree_bytes(element: Element) -> bytes

The original bytes of element, start tag through end tag.

This is the provenance that makes signature wrapping a non-issue: a caller verifies over these bytes and reads values from this same element, so there is no second lookup for an attacker to divert.

Element#

One element, and the byte range of the source it was parsed from.

tag and the keys of attrib use the {uri}local spelling xml.etree.ElementTree uses, so a caller reading values out of this tree reads the same strings it would from the stdlib.

Element(tag: str, attrib: dict[str, str], text: str, tail: str, children: tuple[Element, ...], span: tuple[int, int], nsdeclarations: tuple[tuple[str, str], ...], nsscope: tuple[tuple[str, str], ...] = (), nsinherited: tuple[tuple[str, str], ...] = (), qualified: tuple[tuple[str, str, str, str], ...] = (), prefix: str = '', local: str = '') -> None

Limits#

Bounds every parse is checked against.

The defaults suit a SAML assertion or an S3 response. There is no unbounded setting: a parser on an unauthenticated boundary that will read whatever it is given has no answer to a document built to exhaust memory.

Limits(max_bytes: int = 1048576, max_depth: int = 100, max_elements: int = 65536, max_attributes: int = 256, max_attribute_bytes: int = 65536) -> None

XMLRefusal#

A document outside the accepted profile.

reason is a stable machine-readable code; the message says which construct was refused and where. Both are part of the contract, because a refusal test that asserts only the exception type passes on whichever branch happened to fire.

XMLRefusal(reason: str, message: str, offset: int) -> None

canonicalize_span#

canonicalize_span(data: bytes, start: int, end: int, inherited: Sequence[tuple[str, str]] = (), inclusive_prefixes: Sequence[str] = (), limits: Limits | None = None) -> bytes

Exclusive c14n of data[start:end], seeded with inherited.

Document.canonicalize is the ordinary entry point; this is the one underneath it, exposed for a caller that has bytes and a span from somewhere else.

parse#

parse(data: bytes, limits: Limits | None = None) -> Document

Parse data under limits, or raise XMLRefusal.

limits defaults to Limits, whose bounds suit a SAML assertion or an S3 response. There is no unbounded setting.

wreath.email#

First-party email declarations, transports, suppression, and DKIM.

CapturingEmailSender#

Test sender — records (email, link) pairs instead of sending.

CapturingEmailSender(verifications: list[tuple[str, str]] = <factory>, resets: list[tuple[str, str]] = <factory>, messages: list[Message] = <factory>, suppression: SuppressionList | None = None) -> None

send#

async send(message: Message) -> None

Append message to messages, applying suppression if one is set.

The suppression check is duplicated here rather than inherited so that a test using this sender exercises the same refusal a real one would -- a capturing double that accepts what production refuses is how a suppression bug reaches production green.

send_password_reset#

async send_password_reset(email: str, link: str) -> None

Append (email, link) to resets. Nothing is delivered.

send_verification#

async send_verification(email: str, link: str) -> None

Append (email, link) to verifications. Nothing is delivered.

DkimError#

A key could not be read, or a signature failed its own verification.

DkimSigner#

Signs outgoing mail for one domain/selector pair.

The public half goes in DNS at <selector>._domainkey.<domain> as a TXT record; wreath.doctor.check_email_deliverability reports whether it is actually there, which is the failure this class is otherwise silent about.

Args:

• domain — The signing domain, which must align with the From address
• for DMARC to pass.
• selector — Names the DNS record holding the public key.
• key — An RsaKey or Ed25519Key, usually from load_private_key.
• headers — Which header fields to sign, lower-cased. Defaults to
• DEFAULT_SIGNED_HEADERS; from is added if a caller omits it.

DkimSigner(domain: str, selector: str, key: RsaKey | Ed25519Key, headers: tuple[str, ...] = ('from', 'to', 'cc', 'subject', 'date', 'message-id', 'mime-version', 'content-type', 'content-transfer-encoding', 'list-unsubscribe', 'list-unsubscribe-post')) -> None

algorithm (property)#

algorithm: str

sign#

sign(message: bytes, *, now: float | None = None) -> str

Return the DKIM-Signature header value for a serialised message.

message is the full RFC 5322 message with CRLF line endings, exactly as it will be handed to the MTA. Sign the bytes you send: re-serialising after signing is the most common way a valid signature becomes an invalid one.

Raises:

• DkimError — the message has no header/body separator, or the
• signature failed its own verification.

Ed25519Key#

An Ed25519 private key: the 32-byte seed, per RFC 8032.

Ed25519Key(seed: bytes) -> None