Receive Webhook Events in Python with aiohttp

Direct answer: create an aiohttp.web.Application, register a POST route, read and authenticate the raw request body, parse it according to the provider’s content type, dispatch by verified event metadata, and return an explicit response. The small server below receives GitHub-style JSON deliveries and leaves provider-specific signature verification in a clearly marked function.

aiohttp is an asynchronous HTTP client/server framework for asyncio. Its web server passes a Request object to each handler, so a webhook endpoint is an async coroutine that returns an aiohttp response. See the aiohttp web reference and your provider’s current webhook documentation before deploying.

1. Create a minimal aiohttp webhook receiver

Install aiohttp in a virtual environment:

python -m venv .venv
. .venv/bin/activate
python -m pip install aiohttp

Save this as app.py:

import json
import os
from aiohttp import web

MAX_BODY_BYTES = 25 * 1024 * 1024

async def receive_webhook(request: web.Request) -> web.Response:
    """Receive one provider delivery and acknowledge it explicitly."""
    # Read bytes before parsing. Signature verification must use these exact bytes.
    try:
        raw_body = await request.read()
    except web.HTTPRequestEntityTooLarge:
        raise web.HTTPRequestEntityTooLarge(
            max_size=MAX_BODY_BYTES,
            actual_size=MAX_BODY_BYTES + 1,
        )

    if len(raw_body) > MAX_BODY_BYTES:
        raise web.HTTPRequestEntityTooLarge(
            max_size=MAX_BODY_BYTES,
            actual_size=len(raw_body),
        )

    # Authenticate here, before trusting the decoded event.
    # For GitHub, verify X-Hub-Signature-256 with your configured secret.
    # See the signature section below for a complete verifier.
    if not verify_provider_signature(request, raw_body):
        raise web.HTTPUnauthorized(text="Invalid webhook signature")

    content_type = request.headers.get("Content-Type", "")
    if content_type.split(";", 1)[0].lower() != "application/json":
        raise web.HTTPUnsupportedMediaType(
            text="This endpoint expects application/json"
        )

    try:
        event = json.loads(raw_body)
    except (UnicodeDecodeError, json.JSONDecodeError):
        raise web.HTTPBadRequest(text="Expected valid JSON")

    delivery_id = request.headers.get("X-GitHub-Delivery")
    event_name = request.headers.get("X-GitHub-Event")

    if not isinstance(event, dict):
        raise web.HTTPBadRequest(text="Webhook payload must be a JSON object")

    # Persist delivery_id before performing side effects when idempotency matters.
    # Dispatch only after authentication and basic shape validation.
    await dispatch_event(event_name, delivery_id, event)
    return web.json_response({"received": True})


def verify_provider_signature(request: web.Request, raw_body: bytes) -> bool:
    """Replace this with the selected provider's official verification procedure."""
    # A missing secret means this example cannot authenticate the sender.
    # Return False until a real provider verifier is configured.
    return bool(os.environ.get("WEBHOOK_SECRET"))


async def dispatch_event(event_name: str | None,
                         delivery_id: str | None,
                         event: dict) -> None:
    if event_name == "push":
        # Queue work or call your push handler here.
        return
    if event_name == "issues":
        return
    # Ignore subscribed events that your application does not use.


app = web.Application(client_max_size=MAX_BODY_BYTES)
app.add_routes([web.post("/webhooks/github", receive_webhook)])

if __name__ == "__main__":
    web.run_app(app, host="127.0.0.1", port=8080)

The placeholder verifier deliberately fails closed unless a secret is configured. Do not copy it as a complete GitHub verifier. The important ordering is: read bytes, authenticate bytes, validate the content type, decode JSON, validate fields, then dispatch.

2. Verify GitHub deliveries correctly

GitHub sends an X-Hub-Signature-256 header containing an HMAC-SHA-256 hex digest of the request body when a webhook secret is configured. GitHub recommends this header over the legacy X-Hub-Signature SHA-1 header. The signature covers the raw bytes, so re-serializing JSON before verification can invalidate it. GitHub also supplies X-GitHub-Delivery, a globally unique delivery identifier, and X-GitHub-Event, the event name. These headers help routing and deduplication; they are not authentication by themselves. See GitHub webhook events and payloads.

A reliable webhook pipeline authenticates the raw body before parsing and dispatching the event.

Add this verifier to the previous file:

import hashlib
import hmac


def verify_provider_signature(request: web.Request, raw_body: bytes) -> bool:
    secret = os.environ.get("WEBHOOK_SECRET")
    supplied = request.headers.get("X-Hub-Signature-256", "")
    if not secret or not supplied.startswith("sha256="):
        return False

    expected_digest = hmac.new(
        secret.encode("utf-8"), raw_body, hashlib.sha256
    ).hexdigest()
    expected = f"sha256={expected_digest}"
    return hmac.compare_digest(supplied, expected)

Set the secret in the process environment rather than committing it:

export WEBHOOK_SECRET='replace-with-the-secret-from-github'
python app.py

Only enable events your application handles. GitHub documents a 25 MB payload cap; an event larger than that is not delivered. Set aiohttp’s client_max_size to a limit appropriate for your provider and reject oversized requests early.

3. Parse JSON and URL-encoded webhook formats

await request.json() is convenient when the provider sends application/json. aiohttp checks the content type by default and caches the body, so later reads are possible. Reading await request.read() first is useful when a signature must be checked against the original bytes.

async def json_handler(request: web.Request) -> web.Response:
    try:
        payload = await request.json()
    except (web.HTTPBadRequest, ValueError):
        raise web.HTTPBadRequest(text="Expected application/json")
    return web.json_response({"keys": list(payload)})

GitHub can be configured for JSON or application/x-www-form-urlencoded delivery. Do not send a URL-encoded delivery to a JSON-only handler. For form data, use aiohttp’s parser:

async def form_handler(request: web.Request) -> web.Response:
    form = await request.post()
    action = form.get("action")
    payload_text = form.get("payload")
    return web.json_response({
        "action": action,
        "has_payload": payload_text is not None,
    })

Multipart forms are also handled by request.post(). Keep the configured body limit in mind: aiohttp raises an entity-too-large error when the limit is exceeded. If a provider uses another media type, follow that provider’s parser and signature procedure rather than coercing it into JSON.

4. Route events safely

Authenticate first, then use the verified event header as a dispatch hint. Validate required fields for each event type before acting:

async def dispatch_event(event_name, delivery_id, event):
    if not delivery_id:
        raise web.HTTPBadRequest(text="Missing delivery identifier")

    if event_name == "push":
        repository = event.get("repository", {}).get("full_name")
        commits = event.get("commits", [])
        if not repository or not isinstance(commits, list):
            raise web.HTTPBadRequest(text="Invalid push payload")
        await enqueue("push", delivery_id, event)
        return

    if event_name == "ping":
        await mark_seen(delivery_id)
        return

    # Unknown but authenticated events can be recorded and acknowledged,
    # or rejected according to your provider's policy.
    await record_unhandled(delivery_id, event_name)

Implement enqueue, mark_seen, and record_unhandled with your database or queue. Store the delivery identifier with a uniqueness constraint when duplicate side effects would be harmful. A redelivery can otherwise create duplicate tickets, emails, deployments, or payments. Retry schedules and required acknowledgement status vary by provider, so confirm them in the provider’s documentation.

5. Decide when to acknowledge

Aiohttp lets a handler select the status and body. A successful web.json_response is an explicit 200 response. Keep the synchronous portion short: authenticate, validate, persist or enqueue, and acknowledge. Long API calls inside the request can cause provider timeouts and redelivery.

Use a durable queue when processing is slow or has multiple failure points:

Verify the signature and parse the payload.
Insert the delivery ID and payload in durable storage with a uniqueness constraint.
Return a success response after the insert succeeds.
Let a worker process the event and record success or failure.

If persistence fails, return an error that matches the provider’s retry behavior. Never acknowledge an event that has not been stored when losing it would matter.

6. Test locally with cURL

Start the server, then send a JSON request. This example is for parser testing only; it does not produce a valid GitHub signature:

curl -i http://127.0.0.1:8080/webhooks/github \
  -H 'Content-Type: application/json' \
  -H 'X-GitHub-Delivery: local-001' \
  -H 'X-GitHub-Event: ping' \
  -H 'X-Hub-Signature-256: sha256:invalid' \
  --data '{"zen":"Keep it simple"}'

Because the verifier compares the HMAC, this request should be rejected. For an end-to-end test, calculate the digest over the exact file bytes with the same secret:

body='{"zen":"Keep it simple"}'
signature=$(printf %s "$body" | openssl dgst -sha256 -hmac "$WEBHOOK_SECRET" | awk '{print $2}')
curl -i http://127.0.0.1:8080/webhooks/github \
  -H 'Content-Type: application/json' \
  -H "X-Hub-Signature-256: sha256=$signature" \
  -H 'X-GitHub-Delivery: local-002' \
  -H 'X-GitHub-Event: ping' \
  --data "$body"

7. Send a test delivery from Python or Node.js

Python’s standard library can generate a signed test request. This is useful for integration tests that exercise the same raw bytes your handler receives:

import hashlib
import hmac
import json
import requests

secret = b"replace-with-the-secret"
body = json.dumps({"zen": "Keep it simple"}, separators=(",", ":")).encode()
digest = hmac.new(secret, body, hashlib.sha256).hexdigest()
response = requests.post(
    "http://127.0.0.1:8080/webhooks/github",
    data=body,
    headers={
        "Content-Type": "application/json",
        "X-Hub-Signature-256": f"sha256={digest}",
        "X-GitHub-Delivery": "python-test-001",
        "X-GitHub-Event": "ping",
    },
    timeout=10,
)
print(response.status_code, response.text)

Node.js 18 or newer can do the same with built-in fetch and crypto:

import crypto from 'node:crypto';

const secret = 'replace-with-the-secret';
const body = JSON.stringify({ zen: 'Keep it simple' });
const digest = crypto.createHmac('sha256', secret).update(body).digest('hex');

const response = await fetch('http://127.0.0.1:8080/webhooks/github', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'X-Hub-Signature-256': `sha256=${digest}`,
    'X-GitHub-Delivery': 'node-test-001',
    'X-GitHub-Event': 'ping'
  },
  body
});
console.log(response.status, await response.text());

8. Production checklist

Terminate TLS at a trusted proxy or load balancer and forward requests to aiohttp securely.
Keep the webhook secret in a secret manager or environment variable.
Verify the raw body with the provider’s current algorithm before parsing.
Use constant-time comparison for MACs, such as hmac.compare_digest.
Set a body-size limit and reject unexpected content types.
Validate event-specific fields and ignore events you did not subscribe to.
Persist delivery IDs and make side effects idempotent.
Bound request work with timeouts and enqueue slow processing.
Log delivery ID, event name, status, and processing duration without logging secrets or sensitive payloads.
Monitor authentication failures, parse failures, queue depth, and worker errors.

9. Troubleshooting common failures

Symptom	Likely cause	Fix
401 or 403	Wrong secret, missing prefix, or body changed before verification.	Verify `sha256=`, use the exact raw bytes, and load the configured secret in the running process.
400 invalid JSON	Malformed body, form delivery sent to JSON route, or non-object payload.	Inspect `Content-Type`, select the matching parser, and validate the decoded shape.
415 unsupported media type	Provider sends URL-encoded or another content type.	Configure the provider for JSON or add a dedicated form/multipart route.
413 entity too large	Payload exceeds aiohttp’s `client_max_size` or provider limit.	Reduce subscribed events, raise the limit only when justified, and honor the provider’s documented cap.
Duplicate processing	Provider redelivery or handler timeout after side effects.	Persist the delivery ID and make inserts and side effects idempotent.
Provider reports timeout	Handler performs slow work before responding.	Store or enqueue quickly, acknowledge after durable acceptance, and process asynchronously.
Events never arrive	Route, public URL, TLS, firewall, or provider subscription is wrong.	Check the exact path, proxy logs, provider delivery history, and a local cURL request.

10. Performance, reliability, and cost considerations

For throughput, keep handlers allocation-light, avoid blocking filesystem or database calls in the event loop, and move CPU-heavy work to workers. A durable queue adds operational cost but protects you from losing events during downstream outages. Connection reuse and a reverse proxy can reduce handshake overhead. Measure your own latency and queue behavior; the supplied sources do not establish universal aiohttp benchmarks.

Webhook delivery itself is normally an inbound HTTP concern. Your costs come from the process, database, queue, logging, and downstream APIs. Subscribe only to events you handle, because unnecessary deliveries increase parsing and storage work. GitHub’s documented payload limit is 25 MB; other providers may use different limits, signatures, retries, and acknowledgement rules.

11. Or skip the browser setup

If a webhook triggers a page capture, you can call ScreenshotNeo instead of maintaining browser automation, consent handling, and screenshot workers. One GET request returns a PNG, JPEG, WebP, or PDF:

ScreenshotNeo can handle browser cleanup before a webhook-triggered capture.

cURL (see the ScreenshotNeo API docs):

curl -G "https://api.screenshotneo.com/v1/shot" -d access_key=YOUR_API_KEY --data-urlencode url=https://stripe.com -o shot.webp

Python:

import requests
r = requests.get("https://api.screenshotneo.com/v1/shot", params={"access_key": "YOUR_API_KEY", "url": "https://stripe.com"}, timeout=90)
open("shot.webp", "wb").write(r.content)

Node.js:

const q = new URLSearchParams({ access_key: 'YOUR_API_KEY', url: 'https://stripe.com' }); const res = await fetch(`https://api.screenshotneo.com/v1/shot?${q}`);

ScreenshotNeo removes cookie banners, newsletter popups, and chat widgets before capture. Bot checks, blank pages, failed loads, timeouts, and cache hits are not billed, and the response identifies the result with X-Page-Verdict and X-Billed headers. Its MCP server provides take_screenshot, get_page_info, and capture_pdf for Claude, Cursor, and other MCP clients. The Free plan includes 1,000 screenshots per month with no card; paid plans start at $5 for 3,000 shots. Create a free ScreenshotNeo account.

12. FAQ

Should I use `request.json()` or `request.read()`?

Use request.read() when you must verify a signature over exact bytes, then decode those bytes. Use request.json() for already-authenticated JSON when its content-type behavior fits your endpoint.

Is `X-GitHub-Event` enough to authenticate a request?

No. It identifies the event type. Authenticate the body with GitHub’s configured HMAC signature, then validate the event fields.

What status should every provider receive?

There is no universal status or retry policy. Follow the provider’s current documentation and acknowledge only after the delivery is durably accepted.

Can one aiohttp application serve several providers?

Yes. Use separate routes or provider-aware middleware so each provider gets its own signature algorithm, content-type parser, size limit, and dispatch rules.

How do I handle a provider that sends URL-encoded payloads?

Use await request.post(), inspect the provider’s documented fields, and verify the signature over the representation required by that provider before acting.