Verifiers

once.verifiers.stripe_params

stripe_params(key: str, params: Mapping[str, Any], *, metadata_key: str = METADATA_KEY) -> dict[str, Any]

Tag create params with the once key and use it as the Idempotency-Key.

Returns keyword arguments for a StripeClient create call::

client.v1.payment_intents.create(**stripe_params(idempotency_key, {...}))

Existing metadata is kept. The Idempotency-Key makes Stripe replay the first response if the same request is sent again within 24 hours.

once.verifiers.stripe_verifier

stripe_verifier(kind: Kind, *, client: Any = None, api_key: str | None = None, metadata_key: str = METADATA_KEY, list_params: Callable[[dict[str, Any]], Mapping[str, Any]] | None = None, scan: int = 100, result: Callable[[dict[str, Any]], Any] | None = None) -> Verifier

Build a verifier that finds the Stripe object tagged with the once key.

kind is "charge", "payment_intent", or "refund". client is a stripe.StripeClient, or anything shaped like one (tests pass a fake). If it is None, a StripeClient is made on first use from api_key (or stripe.api_key), which needs the stripe package (pip install agent-once[stripe]).

Charges and PaymentIntents are looked up with the search API first. Search can lag new objects by a minute or so, so if it finds nothing, the newest scan objects from the list API are checked too. Refunds have no search, so only the list is used. list_params maps the tool args to list filters such as {"customer": args["customer"]}, to narrow the scan.

If search isn't available to the account (Stripe doesn't offer it in some regions, or a restricted key can't use it), the list is used instead and search is skipped from then on. With scan=0 that error is raised.

The found object's status decides the outcome. Charges: succeeded is returned, failed is NOT_HAPPENED. PaymentIntents: succeeded and requires_capture are returned, canceled is NOT_HAPPENED. Refunds: succeeded and pending are returned, failed and canceled are NOT_HAPPENED. Any other status (processing, requires_action, requires_payment_method, ...) raises StripeInconclusive, so the record stays unknown until it settles.

Returns the object as a plain dict (or result(obj) if given), or NOT_HAPPENED. NOT_HAPPENED needs a list that was read to the end: if more than scan objects match the filters and none of the first scan is the one, StripeInconclusive is raised and the record stays unknown. With scan=0 a search miss alone returns NOT_HAPPENED, so only use it with stripe_params, whose Idempotency-Key makes a rerun safe for 24 hours. Other Stripe errors propagate, which also leaves the record unknown.

once.verifiers.http_verifier

http_verifier(url: str, *, method: str = 'GET', headers: Mapping[str, str] | None = None, happened: Collection[int] = OK, not_happened: Collection[int] = (404,), timeout: float = 10.0, result: Callable[[Any], Any] | None = None) -> Verifier

Build a verifier that asks an HTTP endpoint whether the effect happened.

url is a template. {key} is the once key and any other {name} is the tool argument of that name, each percent-encoded, for example "https://api.example.com/orders/{order_id}/emails/{key}". Strings go in as they are; other values as JSON, so None is null and True is true. A tool argument named key with {key} in the template raises ValueError, since the template can't tell the two apart.

A status in happened returns the response body: parsed JSON, the text if it is not JSON, or None if it is empty. A status in not_happened returns NOT_HAPPENED, so the call runs again. Any other status, or a network error, raises, which leaves the record unknown.

result maps the body to what the tool itself returns, so a verified result looks like a normal one. It may also return NOT_HAPPENED, for APIs that answer 200 with something like {"found": false}.