Metadata-Version: 2.5
Name: webhooks-cc
Version: 0.3.0
Summary: Python SDK for webhooks.cc: capture, wait for, sign and verify webhooks and test emails
Project-URL: Homepage, https://webhooks.cc
Project-URL: Documentation, https://github.com/kroqdotdev/webhooks-cc/tree/main/packages/sdk-python#readme
Project-URL: Repository, https://github.com/kroqdotdev/webhooks-cc
Author: webhooks.cc
License-Expression: MIT
License-File: LICENSE
Keywords: email,github,pytest,signature,stripe,testing,webhooks
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: Pytest
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Topic :: Software Development :: Testing
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx<1,>=0.27
Provides-Extra: crypto
Requires-Dist: cryptography>=42; extra == 'crypto'
Description-Content-Type: text/markdown

# webhooks-cc

Python SDK for [webhooks.cc](https://webhooks.cc): capture webhooks and email in your tests, wait for them, send signed webhooks to your own handlers, and verify provider signatures.

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

Python 3.10 or later. Get an API key at [webhooks.cc/account/api-keys](https://webhooks.cc/account/api-keys) and set `WHK_API_KEY`, or pass `api_key=`. An AI agent without an account can start in [the agent sandbox](#no-account-the-agent-sandbox).

## Wait for a webhook

```python
from webhooks_cc import WebhooksCC, match_body_path, temporary_endpoint

with WebhooksCC() as client:
    # 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
```

Durations take `"500ms"`, `"30s"`, `"5m"`, `"2h"`, `"7d"`, a `timedelta`, or a number of milliseconds (the unit every webhooks.cc SDK uses, so prefer a string where you mean seconds). Times such as `since` and `Request.received_at` are milliseconds since the epoch.

Matchers combine with `match_all()` and `match_any()`, and any function that takes a `Request` and returns a bool works too: `match_method`, `match_path` (globs: `*` within a segment, `**` across segments, `?` for one character), `match_header`, `match_query_param`, `match_body_path`, `match_body_subset`, `match_content_type`, `match_verified`, `match_unverified` and `match_email`.

## pytest

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

```python
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 session; `webhooks_cc_endpoint` is a new endpoint for each test, deleted afterwards. Outside pytest, `temporary_endpoint(client)` is the same as a context manager, and `capture_during(client, action)` runs `action(endpoint)` and returns what it sent.

## 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 its dashboard shows.

```python
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.

## Email fields, reply text and provider shapes

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

```python
client.endpoints.update(
    endpoint.slug,
    email_fields=[{"name": "invoice", "from": "subject", "pattern": r"(INV-\d+)"}],
)
email = client.emails.wait_for(endpoint.slug, tag=run)
email.email_fields  # {"invoice": "INV-2041"}, None for a field that found nothing
email.reply_text    # what a reply says above the quoted message, without the signature
```

Forwarding templates read the fields as `{{ fields.invoice }}`. `email_field_issues()` checks a list before you save it (the API checks the patterns themselves).

To test a handler you wrote for an inbound mail provider, render a captured email in that provider's shape and post it there:

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

`format` is `"postmark"`, `"sendgrid"`, `"mailgun"` (pass `mailgun_signing_key` to sign the form as Mailgun does) or `"cloudmailin"`, or `"json"` for the `email.received` JSON.

## 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:

```python
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. Redirects are never followed, even by an `http_client` of your own, so signed requests only reach the host you named. `client.endpoints.send_template(slug, ...)` sends the same to a webhooks.cc endpoint. `TEMPLATE_PROVIDERS` lists the 34 providers.

## Verify signatures

```python
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 `method`. Discord takes `public_key`. One function per provider is in `webhooks_cc.verify`.

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):

```python
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"], event["data"]["replyText"], event["data"].get("fields"))
```

`verify_forwarded_request()` checks a request forwarded as JSON, `verify_forwarded_json()` a destination's own JSON (it returns the parsed body, whatever its shape), and `verify_relayed_request()` one relayed as it was received.

## 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:

```python
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)
```

Filters take 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`. `format` may also be an inbound mail provider's shape (`"postmark"`, `"sendgrid"`, `"mailgun"` or `"cloudmailin"`): emails go out as that provider posts them and HTTP requests as received. `mailgun_signing_key` signs the Mailgun form; it is never returned, and `has_mailgun_signing_key` says whether one is saved. `preview()` is a dry run: pass a saved `destination_id`, unsaved fields in `destination` (the names `create()` takes), or both. `update()` changes only the arguments you pass, `test()` sends one delivery now and reports the answer, and `list()` (with `summary=True` for the last 24 hours), `get()`, `delete()`, `reorder()`, `secret()` and `rotate_secret()` do what they say.

## Mock responses and response rules

An endpoint answers with `mock_response` instead of a plain 200 OK, or with the first of its `response_rules` whose conditions match the request (method, path, header, query, body, body_path, content_type or size). Bodies and header values may hold `{{ placeholders }}` filled in from each request, such as `{{ json.id }}`. Check what a sender would get back without sending anything:

```python
preview = client.endpoints.preview_response(
    endpoint.slug,
    response_rules=[
        {
            "conditions": [{"field": "body_path", "path": "type", "op": "eq", "value": "ping"}],
            "response": {"status": 202, "body": '{"pong": "{{ json.id }}"}', "headers": {}},
        }
    ],
)
print(preview.rule, preview.response.status if preview.response else 200)
```

It runs against the endpoint's newest HTTP request unless you pass `request_id`, and uses the saved settings unless you pass `mock_response` or `response_rules` (`None` tries none at all). `client.endpoints.update(slug, response_rules=[...])` saves them.

## Stream requests as they arrive

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

## No account: the agent sandbox

An AI agent can capture webhooks before anyone signs up. `WebhooksCC.sandbox()` registers through [auth.md](https://webhooks.cc/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.

```python
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 registration is saved in the agent credential file the MCP server, the other SDKs and the `whk` CLI share: `~/.config/whk/agent.json` on Linux (in `$XDG_CONFIG_HOME` when that is set), `~/Library/Application Support/whk/agent.json` on macOS, `%APPDATA%\whk\agent.json` on Windows, or the path in `WHK_AGENT_FILE`. The next `WebhooksCC.sandbox()` reuses a live sandbox instead of registering again, and `credentials_file=False` leaves the file alone. The claim token is never saved, so only the process that registered moves the sandbox endpoints into the account; a sandbox reused from the file starts a new code for the email instead (`claim.moves_endpoints` is false), and its endpoints stay in the sandbox until they expire.

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

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

`WebhooksCC.connect(email)` connects an agent without a sandbox: show the human `pending.verification_uri` and `pending.user_code`, then `pending.wait()` returns the client. An API key always wins over the file. A refused sandbox call raises `SandboxError` with the server's `code`; the protocol steps themselves are in `webhooks_cc.agent`.

## Errors

API errors raise `WebhooksCCError` with a `status_code`, or one of its subclasses: `UnauthorizedError` (401), `NotFoundError` (404), `RateLimitError` (429, with `retry_after`) and `TimeoutError`, which 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`.

## Same behaviour as the TypeScript SDK

This SDK passes the same conformance vectors as [`@webhooks-cc/sdk`](https://www.npmjs.com/package/@webhooks-cc/sdk): signing, signature verification, forwarding signatures, matchers and durations. They live in [`packages/sdk/conformance`](https://github.com/kroqdotdev/webhooks-cc/tree/main/packages/sdk/conformance).

## License

MIT
