Python SDK

The webhooks-cc package brings the TypeScript SDK's testing tools to Python: capture webhooks and email, wait for them, send signed webhooks to your own handler, and verify signatures. It passes the same conformance tests as the TypeScript SDK, so signatures and matchers behave identically in both.

Install

pip install webhooks-cc
pip install "webhooks-cc[crypto]"  # also verify Discord (Ed25519) and PayPal (RSA) signatures

Python 3.10 or later. Create an API key on your API keys page and set WHK_API_KEY, or pass api_key=. An AI agent without an account can start in the agent sandbox.

Wait for a webhook

from webhooks_cc import WebhooksCC, match_body_path, temporary_endpoint
 
with WebhooksCC() as client:  # reads WHK_API_KEY
    # Deleted when the block ends, even when the test fails.
    with temporary_endpoint(client, "checkout test") as endpoint:
        trigger_checkout(webhook_url=endpoint.url)
 
        request = client.requests.wait_for(
            endpoint.slug,
            timeout="30s",
            match=match_body_path("type", "checkout.session.completed"),
        )
        assert request.json()["data"]["object"]["amount_total"] == 2000

wait_for looks back five minutes, so a webhook that arrived just before you started waiting still counts. wait_for_all(slug, count=3) collects several, oldest first.

Durations take "500ms", "30s", "5m", "2h", "7d" or a timedelta. A bare number is milliseconds, as in every webhooks.cc SDK, so prefer a string where you mean seconds. Times such as since and request.received_at are milliseconds since the epoch.

Matchers

Matchers take a captured Request and return a bool. Combine them with match_all() and match_any(), or pass any function of your own.

from webhooks_cc import match_all, match_header, match_method, match_path
 
match = match_all(
    match_method("POST"),
    match_path("/webhooks/**"),
    match_header("x-github-event", "push"),
)
request = client.requests.wait_for(endpoint.slug, match=match)

Available: match_method, match_path (globs: * within a segment, ** across segments, ? one character), match_header, match_query_param, match_body_path, match_body_subset, match_content_type, match_verified, match_unverified and match_email. See Matchers & Helpers for what each one matches.

Preview a response

preview_response shows what a captured request would get back from the endpoint's response rules, with placeholders filled in, without sending anything. Pass response_rules or mock_response to try settings before saving them; leave them out to use the saved ones.

preview = client.endpoints.preview_response(
    endpoint.slug,
    response_rules=[
        {
            "conditions": [{"field": "body_path", "op": "gt", "path": "data.amount", "value": "10000"}],
            "response": {"status": 422, "body": '{"amount": {{ json.data.amount }}}', "headers": {}},
        }
    ],
)
print(preview.rule, preview.response)

Favourite endpoints

Mark an endpoint, your own or one a team shares with you, as a favourite: the dashboard lists favourites first and opens the first one by default.

client.endpoints.favorite(endpoint.slug)    # True
client.endpoints.unfavorite(endpoint.slug)  # False
 
favorites = [item for item in client.endpoints.list() if item.favorite]

Both are idempotent and return the endpoint's favourite state afterwards. A favourite is yours alone, so marking a shared endpoint changes nothing for anyone else. endpoint.favorite is set on what endpoints.list() and endpoints.get() return, and None on what create() and update() return. They need an account: the sandbox client has neither method.

pytest

Installing the package adds two fixtures. Tests that use them are skipped when WHK_API_KEY is not set.

def test_sends_a_receipt_webhook(webhooks_cc, webhooks_cc_endpoint):
    place_order(webhook_url=webhooks_cc_endpoint.url)
 
    request = webhooks_cc.requests.wait_for(webhooks_cc_endpoint.slug, timeout="30s")
    assert request.method == "POST"

webhooks_cc is a client for the whole session. webhooks_cc_endpoint is a new endpoint for each test, named after it and deleted when it ends. The pytest guide walks through a whole suite: handler tests without a network, retries, emails and CI. Outside pytest, temporary_endpoint(client) does the same as a context manager, and capture_during(client, action) runs action(endpoint) and returns what it sent:

from webhooks_cc import capture_during
 
[request] = capture_during(client, lambda endpoint: notify(endpoint.url))
assert request.json()["event"] == "order.paid"

Test email: codes and links

Every endpoint receives email at {slug}@mailhooks.cc, and at {slug}+{tag}@mailhooks.cc for any tag. Use a fresh tag per test run to find your own email. webhooks.cc finds the one-time codes and links in each email, the same ones the dashboard shows, and the SDK reads them from the API.

import uuid
from webhooks_cc import extract_code, extract_link
 
run = f"signup-{uuid.uuid4().hex[:8]}"
sign_up(email=client.emails.address(endpoint, run))
 
email = client.emails.wait_for(endpoint.slug, tag=run, subject="Confirm your email")
code = extract_code(email)   # "482913"
link = extract_link(email)   # "https://app.example.com/confirm?token=..."

email.email holds the parsed message: subject, from_, to, text, html, attachments, and the SPF, DKIM and DMARC results. extract_code() raises ValueError when the endpoint's Show codes and links found in emails setting is off. See Email capture for how capture works.

Email fields and reply text

An endpoint can pick up to 20 named fields out of every email it captures. Each reads the subject, text, html, from, to (To and Cc) or one header, and the first group of its RE2 pattern (the whole match without a group) is the value:

from webhooks_cc import EmailField
 
client.endpoints.update(
    endpoint.slug,
    email_fields=[
        {"name": "invoice", "from": "subject", "pattern": r"(INV-\d+)"},
        EmailField(name="mailer", from_="header", header="X-Mailer", pattern=".+"),
    ],
)
 
email = client.emails.wait_for(endpoint.slug, tag=run)
email.email_fields  # {"invoice": "INV-2041", "mailer": "Acme Billing"}
email.reply_text    # "Thanks, paid." without the quoted message and the signature

Every email the client reads carries the values as email_fields (None for a field that found nothing, and None altogether when the endpoint has no fields), and forwarding templates read them as {{ fields.invoice }}. endpoint.email_fields lists the saved fields, and email_fields=None removes them. The client refuses a list that email_field_issues() finds fault with (names, sources, headers, pattern lengths) before sending it; the API checks the patterns themselves. reply_text is the new text of a reply, or None when the email quotes no earlier message.

Layouts and stored messages

An endpoint's layouts read each email before it is delivered: fields, line items and attachments. Pass layouts= to endpoints.update() (as Layout objects or dicts in the API's shape, None removes them), try them on a captured email first with preview_layouts(), and read the result as request.parsed:

preview = client.endpoints.preview_layouts(endpoint.slug)  # the saved layouts on the newest email
print(preview.problems, preview.parsed.fields if preview.parsed else None)
 
email = client.requests.get(email.id)
print(email.parsed.line_items if email.parsed else None)

The preview saves nothing and uses no parsed pages. For an email kept in storage (email.email.stored), client.requests.raw(id) returns the exact message as bytes and client.requests.attachment(id, index) one attachment (content, filename, content_type).

Test an inbound email handler

If your app already receives email through Postmark, SendGrid, Mailgun or CloudMailin, render a captured email in that provider's shape and post it to the handler you wrote for it, on localhost or anywhere else:

rendered = client.emails.render(email.id, format="postmark")
response = client.send_to(
    "http://localhost:8000/inbound/postmark",
    body=rendered.body,
    headers=rendered.headers,
)
assert response.status_code == 200

format is "postmark" (Postmark's inbound JSON), "sendgrid" (an Inbound Parse form), "mailgun" (a route's form; pass mailgun_signing_key, your Mailgun HTTP webhook signing key, and its signature verifies) or "cloudmailin" (JSON Normalized), or "json" for the email.received JSON, unsigned. rendered.headers holds the content type and user agent to send with rendered.body. To post every captured email this way, give a destination that format; see In a mail provider's shape.

Send signed webhooks to your handler

Test your own webhook handler, on localhost or anywhere else, with requests signed the way the provider signs them:

response = client.send_to(
    "http://localhost:8000/webhooks/stripe",
    provider="stripe",
    secret="whsec_test_secret",
    template="checkout.session.completed",
)
assert response.status_code == 200

Leave out body to send the provider's sample payload, or pass your own: a string or bytes go as they are, anything else as JSON. timestamp (seconds) and message_id make the signature repeatable. build_request() returns the method, URL, headers and body without sending them. Signed requests never follow redirects, even with an http_client of your own that does, so they only reach the host you named. client.endpoints.send_template(slug, ...) sends the same to a webhooks.cc endpoint, and TEMPLATE_PROVIDERS lists the 34 providers.

Verify signatures

from webhooks_cc import verify_signature
 
valid = verify_signature(
    "stripe",
    body=request.body,        # the exact bytes or text received
    headers=request.headers,  # any mapping; names match in any case
    secret="whsec_...",
)

Twilio, Square and HubSpot also need url; HubSpot also needs method. Discord takes public_key. One function per provider is in webhooks_cc.verify, such as verify_stripe_signature(body, header, secret). See Signature Verification for each provider's scheme.

Forwarding destinations

An endpoint forwards what it captures to up to 10 destinations, each with its own kinds, filter, format, headers, retries, throttle, delay and signing secret. client.destinations manages them:

destination = client.destinations.create(
    endpoint.slug,
    url="https://hooks.example.com/invoices",
    email=False,
    filter={
        "conditions": [{"field": "body_path", "path": "type", "op": "eq", "value": "invoice.paid"}]
    },
    format="template",
    body_template={
        "invoice": "{{ json.data.object.id }}",
        "amount": "{{ json.data.object.amount_paid }}",
    },
)
preview = client.destinations.preview(endpoint.slug, destination_id=destination.id)
print(preview.filter_passes, preview.delivery.body if preview.delivery else preview.error)

A filter takes the conditions of response rules, plus kind, email (with name subject, from, to, text or tag), spam_score, email_auth (with name spf, dkim or dmarc) and signature; see Send only some captures. format="template" sends your own JSON, and "chat" a message for Slack or Discord, with your own message_template if you like. "postmark", "sendgrid", "mailgun" and "cloudmailin" post emails in that mail provider's shape and HTTP requests as received; mailgun_signing_key signs the Mailgun form as Mailgun does. The key is stored encrypted and never returned (has_mailgun_signing_key says whether one is saved).

preview() is a dry run: whether the filter lets a captured request through (filter_passes) and the exact request the destination would send (delivery), or why it would send nothing (error). Pass a saved destination_id, unsaved fields in destination (the names create() takes), or both, and a request_id to run it against a particular request. Nothing is sent.

update(slug, destination_id, ...) changes only the arguments you pass, and None clears name, filter, body_template, message_template, sent_field and mailgun_signing_key. test() sends one delivery now and reports the answer, its filter reported in filter_passes rather than applied. list(slug, summary=True) adds each destination's last 24 hours, and get(), delete(), reorder(), secret() and rotate_secret() do what they say.

Verify what webhooks.cc forwards

When webhooks.cc forwards requests or email to your server, check that they came from it with the destination's signing secret (client.destinations.secret(slug, destination_id), or client.forwarding.secret(slug) for the endpoint's primary destination):

from webhooks_cc import WebhookVerificationError, verify_forwarded_email
 
try:
    event = verify_forwarded_email(body, headers, secret)
except WebhookVerificationError as error:
    return 400 if error.code == "invalid_payload" else 401
print(event["data"]["subject"])

The event's data also carries replyText and, when the endpoint has named fields, fields. verify_forwarded_request() checks a request forwarded as JSON, and verify_relayed_request() one relayed as it was received. A destination that sends your own JSON is checked with verify_forwarded_json(), which makes the same checks and returns the parsed body, whatever its shape:

paid = verify_forwarded_json(body, headers, secret)
print(paid["invoice"], paid["amount"])

Stream requests as they arrive

for request in client.requests.subscribe(endpoint.slug, timeout="10m", reconnect=True):
    print(request.method, request.path)

The stream ends at timeout, when the endpoint is deleted, or when you stop iterating. With reconnect=True it follows the server's 30-minute connection rotation and recovers from dropped connections without missing requests.

No account: the agent sandbox

An AI agent can capture webhooks before anyone signs up. WebhooksCC.sandbox() registers through auth.md, solves a short proof of work (a few seconds of CPU) and returns a sandbox: up to 3 endpoints, 25 captured requests each and 100 in all, for 24 hours. Sandbox endpoints only capture: no mock responses, forwarding or email. See AI agents for the limits and how a person connects an agent.

from webhooks_cc import WebhooksCC
 
with WebhooksCC.sandbox(client_name="my-agent") as sandbox:
    endpoint = sandbox.endpoints.create()
    trigger_checkout(webhook_url=endpoint.url)
    request = sandbox.requests.wait_for(endpoint.slug, timeout="60s")
 
    # To keep the endpoints, a human connects the agent to their account.
    claim = sandbox.claim("dev@example.com")
    print(f"Open {claim.verification_uri} and enter {claim.user_code}")
    client = sandbox.wait_for_claim()  # a WebhooksCC for that account

The human opens the link, signs in (or signs up) as that email and types the code; the endpoints and what they captured move into their account. wait_for_claim() raises AgentAuthError with code expired_token when the 15-minute code runs out (call claim() again) and access_denied when the human declines.

The registration is saved in the agent credential file that the MCP server, the TypeScript and Go SDKs and the CLI share:

WHK_AGENT_FILE names another path. The next WebhooksCC.sandbox(), in this process or another, reuses a live sandbox instead of registering again; pass credentials_file=False to leave the file alone, or a path to use another file. The claim token is never saved, so only the process that registered moves the sandbox endpoints into an account. A sandbox reused from the file starts a new code for the email instead (claim.moves_endpoints is False): the agent is connected to the account, and the sandbox endpoints stay where they are until they expire.

A connection lasts 90 days, and every process picks it up:

client = WebhooksCC.connected()  # None when no agent is connected
client = WebhooksCC()            # without WHK_API_KEY, acts for the connected account

To connect an agent without a sandbox, start with the human's email:

pending = WebhooksCC.connect("dev@example.com", client_name="my-agent")
print(f"Open {pending.verification_uri} and enter {pending.user_code}")
client = pending.wait()

pending.restart() gets a new code when the old one ran out. An API key, passed or in WHK_API_KEY, always wins over the file. A refused sandbox call raises SandboxError with the server's code (sandbox_endpoint_limit, sandbox_full, ...), and a rate limit RateLimitError. Each step of the protocol is a function of its own in webhooks_cc.agent.

Errors

API errors raise WebhooksCCError with a status_code, or one of its subclasses:

TimeoutError is also Python's built-in TimeoutError. Network failures raise httpx's exceptions. Retries are off unless you pass retry=RetryOptions(max_attempts=3). They repeat GET, PUT and DELETE calls after a network error, a timeout or a status in retry_on (429, 500, 502, 503 and 504 by default), and any call after a 429; a POST or PATCH that failed any other way may have gone through, so it is not repeated, and neither is requests.clear() without before.

Configuration

The source is in packages/sdk-python, MIT licensed.