Python webhook handler

Example FastAPI and Flask handlers that verify Wocha webhook signatures and dispatch by event type.

Setup

pip install wocha fastapi uvicorn
# or
pip install wocha flask

Set your signing secret:

export WOCHA_WEBHOOK_SECRET=whsec_...

FastAPI handler

import os
from fastapi import FastAPI, Request, Response, HTTPException
from wocha.webhooks import verify_webhook
from wocha.errors import AuthenticationError, GreetError
 
app = FastAPI()
 
 
@app.post("/webhooks/wocha")
async def wocha_webhook(request: Request) -> Response:
    payload = (await request.body()).decode("utf-8")
    signature = request.headers.get("X-Wocha-Signature", "")
 
    try:
        event = verify_webhook(
            payload=payload,
            signature=signature,
            secret=os.environ["WOCHA_WEBHOOK_SECRET"],
        )
    except AuthenticationError:
        raise HTTPException(status_code=401, detail="Invalid signature")
    except GreetError as err:
        raise HTTPException(status_code=400, detail=str(err))
 
    try:
        await handle_webhook_event(event)
    except Exception:
        raise HTTPException(status_code=500, detail="Handler failed")
 
    return Response(content="OK", status_code=200)
 
 
async def handle_webhook_event(event: dict) -> None:
    event_type = event["event_type"]
    data = event["data"]
 
    if event_type == "user.created":
        await provision_account(data)
    elif event_type == "user.updated":
        await sync_user_profile(data)
    elif event_type == "user.deleted":
        await delete_account(data["id"])
    elif event_type == "session.created":
        await track_session(data)
    elif event_type == "session.revoked":
        await invalidate_session_cache(data["id"])
    elif event_type in ("org.created", "organisation.created"):
        await create_organisation_record(data)
    elif event_type in ("org.member.added", "organisation.member_added"):
        await sync_membership(data)
    elif event_type in ("org.member.removed", "organisation.member_removed"):
        await remove_membership(data)
    elif event_type == "connection.created":
        await sync_connection(data)
    elif event_type == "application.created":
        await register_application(data)
    elif event_type == "application.deleted":
        await unregister_application(data["client_id"])
    else:
        print(f"[webhook] unhandled event: {event_type}")
 
 
async def provision_account(data: dict) -> None:
    ...
 
 
async def sync_user_profile(data: dict) -> None:
    ...
 
 
async def delete_account(user_id: str) -> None:
    ...
 
 
async def track_session(data: dict) -> None:
    ...
 
 
async def invalidate_session_cache(session_id: str) -> None:
    ...
 
 
async def create_organisation_record(data: dict) -> None:
    ...
 
 
async def sync_membership(data: dict) -> None:
    ...
 
 
async def remove_membership(data: dict) -> None:
    ...
 
 
async def sync_connection(data: dict) -> None:
    ...
 
 
async def register_application(data: dict) -> None:
    ...
 
 
async def unregister_application(client_id: str) -> None:
    ...
 
 
# uvicorn main:app --reload --port 8000

Flask handler

import os
from flask import Flask, request, Response
from wocha.webhooks import verify_webhook
from wocha.errors import AuthenticationError, GreetError
 
app = Flask(__name__)
 
 
@app.post("/webhooks/wocha")
def wocha_webhook() -> Response:
    payload = request.get_data(as_text=True)
    signature = request.headers.get("X-Wocha-Signature", "")
 
    try:
        event = verify_webhook(
            payload=payload,
            signature=signature,
            secret=os.environ["WOCHA_WEBHOOK_SECRET"],
        )
    except AuthenticationError:
        return Response("Invalid signature", status=401)
    except GreetError as err:
        return Response(str(err), status=400)
 
    try:
        handle_webhook_event(event)
    except Exception:
        return Response("Handler failed", status=500)
 
    return Response("OK", status=200)
 
 
def handle_webhook_event(event: dict) -> None:
    event_type = event["event_type"]
    data = event["data"]
 
    if event_type == "user.created":
        provision_account(data)
    elif event_type == "user.deleted":
        delete_account(data["id"])
    elif event_type == "session.revoked":
        invalidate_session_cache(data["id"])
    elif event_type in ("org.member.added", "organisation.member_added"):
        sync_membership(data)
    else:
        print(f"[webhook] unhandled event: {event_type}")
 
 
def provision_account(data: dict) -> None:
    ...
 
 
def delete_account(user_id: str) -> None:
    ...
 
 
def invalidate_session_cache(session_id: str) -> None:
    ...
 
 
def sync_membership(data: dict) -> None:
    ...
 
 
# flask --app main run --port 8000

Error handling

The verify_webhook helper raises:

Exception	HTTP status	Meaning
`AuthenticationError`	401	Invalid or expired signature
`GreetError` (validation)	400	Malformed payload or unknown event type

Always read the raw request body — do not use Pydantic models or request.json() before verification.

Local testing

wocha webhook listen --forward-to http://localhost:8000/webhooks/wocha
wocha webhook test http://localhost:8000/webhooks/wocha user.created