Webhooks

Build webhook receivers in Singularity using the BaseWebhook class — with signature verification, event routing, handler dispatch, and idempotency patterns.

Webhooks let external platforms (Stripe, GitHub, Slack, and others) push events to your application in real time. Singularity provides a BaseWebhook base class that handles the boilerplate of raw body extraction, signature verification, JSON parsing, event type extraction, and handler dispatch. You implement the platform-specific parts: how to verify signatures, how to extract the event type, and what to do for each event.

How It Works

A webhook service is a standard Singularity service that extends BaseWebhook instead of being a plain class. The Manager auto-discovers it like any other service and registers its post method as a POST endpoint. When an external platform sends a webhook, the request flows through a well-defined pipeline.

Processing Pipeline

Raw body extraction -- The post() method reads the raw request body and headers before any JSON parsing. This is critical because signature verification must operate on the exact bytes that were sent.
Signature verification -- Your verify() method receives the headers and raw body. Return True to proceed or False to reject with a 401 response.
JSON parsing -- The body is parsed as JSON. If parsing fails, a 400 response is returned.
Event type extraction -- Your get_event_type() method extracts the event type string from the payload or headers (platform-dependent).
Handler dispatch -- The event type is looked up in the events dictionary. If a matching handler method exists, it is called with (payload, headers). If no mapping exists, on_unhandled_event() is called instead.
Response -- The handler's return value is wrapped in a JSONResponse with status 200. If the handler returns a JSONResponse directly, it is used as-is. Exceptions are caught and logged, returning a 500 response.

The BaseWebhook Class

Here is the interface you implement when creating a webhook service:

Method	Required	Description
`verify(headers, raw_body) -> bool`	Override recommended	Verify the webhook signature. Returns `True` by default (pass-through for development).
`get_event_type(headers, payload) -> str`	Override recommended	Extract the event type string. Returns `payload.get("type", "unknown")` by default.
`events: dict[str, str]`	Yes	Maps event type strings to handler method names.
`on_unhandled_event(event_type, payload, headers)`	Optional	Called when an event type has no mapping. Default: log and return 200 with `{"status": "ignored"}`.
Handler methods `(payload, headers)`	Yes	One async method per event type, named in the `events` dict.

The post() method is already implemented by BaseWebhook and should not be overridden. It handles the full pipeline described above.

The BaseWebhook class sets rpc_exposed = False by default. Webhook services are typically not called via RPC since they are designed to receive external HTTP requests. You can override this if needed.

Stripe Webhook Example

Stripe sends webhook events as JSON with a stripe-signature header containing an HMAC-SHA256 signature. The event type is in payload["type"].

# services/v1/stripe/service.py
"""Stripe webhook receiver."""

import hmac
import hashlib

from singularity.core.webhook import BaseWebhook
from singularity.core.acquire import Acquire


class StripeService(BaseWebhook):
  """Receives and processes Stripe webhook events."""

  events = {
    "invoice.paid": "handle_invoice_paid",
    "invoice.payment_failed": "handle_invoice_failed",
    "customer.created": "handle_customer_created",
    "customer.subscription.deleted": "handle_subscription_deleted",
    "charge.refunded": "handle_refund",
  }

  def __init__(self, acquire: Acquire):
    super().__init__(acquire)
    self.db = acquire.db_session

  def verify(self, headers: dict, raw_body: bytes) -> bool:
    """Verify Stripe's HMAC-SHA256 signature."""
    secret = self.acquire.settings.stripe_webhook_secret
    signature = headers.get("stripe-signature", "")

    # Stripe uses a timestamp + signature format: t=xxx,v1=yyy
    # For simplicity, this example checks the v1 signature
    parts = dict(
      pair.split("=", 1) for pair in signature.split(",") if "=" in pair
    )
    timestamp = parts.get("t", "")
    sig_hash = parts.get("v1", "")

    # Construct the signed payload
    signed_payload = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(
      secret.encode(), signed_payload, hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(sig_hash, expected)

  def get_event_type(self, headers: dict, payload: dict) -> str:
    """Stripe puts the event type in the JSON body."""
    return payload.get("type", "unknown")

  async def handle_invoice_paid(self, payload: dict, headers: dict):
    """Process a successful invoice payment."""
    invoice = payload["data"]["object"]
    invoice_id = invoice["id"]
    customer_id = invoice["customer"]
    amount = invoice["amount_paid"]

    async with self.db() as session:
      session.add(Payment(
        external_id=invoice_id,
        customer_id=customer_id,
        amount=amount,
        status="paid",
      ))
      await session.commit()

    return {"status": "processed", "invoice": invoice_id}

  async def handle_invoice_failed(self, payload: dict, headers: dict):
    """Handle a failed invoice payment."""
    invoice_id = payload["data"]["object"]["id"]
    # Notify the customer, retry logic, etc.
    return {"status": "recorded", "invoice": invoice_id}

  async def handle_customer_created(self, payload: dict, headers: dict):
    """Sync a new Stripe customer to the local database."""
    customer = payload["data"]["object"]
    return {"status": "synced", "customer": customer["id"]}

  async def handle_subscription_deleted(self, payload: dict, headers: dict):
    """Handle subscription cancellation."""
    subscription = payload["data"]["object"]
    return {"status": "cancelled", "subscription": subscription["id"]}

  async def handle_refund(self, payload: dict, headers: dict):
    """Process a charge refund."""
    charge = payload["data"]["object"]
    return {"status": "refunded", "charge": charge["id"]}

This service is auto-discovered at POST /api/v1/stripe. Configure the Stripe dashboard to send events to https://your-domain.com/api/v1/stripe.

GitHub Webhook Example

GitHub sends the event type in the x-github-event header and uses x-hub-signature-256 for HMAC-SHA256 verification.

# services/v1/github/service.py
"""GitHub webhook receiver."""

import hmac
import hashlib

from singularity.core.webhook import BaseWebhook
from singularity.core.acquire import Acquire


class GithubService(BaseWebhook):
  """Receives and processes GitHub webhook events."""

  events = {
    "push": "handle_push",
    "pull_request": "handle_pr",
    "issues": "handle_issue",
    "release": "handle_release",
  }

  def __init__(self, acquire: Acquire):
    super().__init__(acquire)

  def verify(self, headers: dict, raw_body: bytes) -> bool:
    """Verify GitHub's HMAC-SHA256 signature."""
    secret = self.acquire.settings.github_webhook_secret
    signature = headers.get("x-hub-signature-256", "")

    expected = "sha256=" + hmac.new(
      secret.encode(), raw_body, hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(signature, expected)

  def get_event_type(self, headers: dict, payload: dict) -> str:
    """GitHub puts the event type in the X-GitHub-Event header."""
    return headers.get("x-github-event", "unknown")

  async def handle_push(self, payload: dict, headers: dict):
    """Handle a push event."""
    branch = payload["ref"].split("/")[-1]
    commits = len(payload.get("commits", []))
    pusher = payload.get("pusher", {}).get("name", "unknown")

    return {
      "status": "received",
      "branch": branch,
      "commits": commits,
      "pusher": pusher,
    }

  async def handle_pr(self, payload: dict, headers: dict):
    """Handle a pull request event."""
    action = payload.get("action", "unknown")
    pr_number = payload["pull_request"]["number"]
    title = payload["pull_request"]["title"]

    return {
      "status": "received",
      "action": action,
      "pr": pr_number,
      "title": title,
    }

  async def handle_issue(self, payload: dict, headers: dict):
    """Handle an issue event."""
    action = payload.get("action", "unknown")
    issue_number = payload["issue"]["number"]

    return {"status": "received", "action": action, "issue": issue_number}

  async def handle_release(self, payload: dict, headers: dict):
    """Handle a release event."""
    tag = payload["release"]["tag_name"]
    return {"status": "received", "tag": tag}

Handling Unhandled Events

When an incoming event type has no entry in the events dictionary, on_unhandled_event() is called. The default implementation logs the event and returns a 200 status with {"status": "ignored", "event": event_type}.

This default behavior is intentional: most webhook providers expect a 200 response even for events you do not handle. Returning a non-2xx status may cause the provider to retry the delivery repeatedly.

Override on_unhandled_event() if you need custom behavior:

class StripeService(BaseWebhook):
  events = {
    "invoice.paid": "handle_invoice_paid",
  }

  async def on_unhandled_event(self, event_type: str, payload: dict, headers: dict):
    """Store unhandled events for later replay."""
    async with self.db() as session:
      session.add(UnprocessedEvent(
        event_type=event_type,
        payload=payload,
        source="stripe",
      ))
      await session.commit()

    return JSONResponse(
      status_code=200,
      content={"status": "stored_for_replay", "event": event_type},
    )

Storing unhandled events is useful during development. As you add new event handlers, you can replay stored events to test them without triggering real webhook deliveries from the provider.

Idempotency

Webhook providers may deliver the same event multiple times (network retries, provider-side deduplication failures). BaseWebhook does not enforce idempotency -- this is intentionally left to the handler implementation because the deduplication strategy depends on your data model and business logic.

Here is a common pattern using an event ID as a deduplication key:

from fastapi.responses import JSONResponse

class StripeService(BaseWebhook):
  events = {
    "invoice.paid": "handle_invoice_paid",
  }

  def __init__(self, acquire: Acquire):
    super().__init__(acquire)
    self.db = acquire.db_session

  async def handle_invoice_paid(self, payload: dict, headers: dict):
    """Idempotent handler -- skips already-processed events."""
    event_id = payload["id"]  # Stripe event ID, e.g., "evt_1Abc..."

    async with self.db() as session:
      # Check if this event was already processed
      existing = await session.execute(
        select(ProcessedWebhookEvent).where(
          ProcessedWebhookEvent.event_id == event_id
        )
      )
      if existing.scalar_one_or_none():
        return {"status": "already_processed", "event_id": event_id}

      # Process the event
      invoice = payload["data"]["object"]
      session.add(Payment(
        external_id=invoice["id"],
        amount=invoice["amount_paid"],
        status="paid",
      ))

      # Record the event as processed
      session.add(ProcessedWebhookEvent(
        event_id=event_id,
        event_type="invoice.paid",
        processed_at=datetime.utcnow(),
      ))
      await session.commit()

    return {"status": "processed", "event_id": event_id}

Always check for duplicate events before performing side effects (database writes, sending emails, charging payments). The ProcessedWebhookEvent table acts as an idempotency key store. Consider adding a unique constraint on event_id for additional safety.

A more robust approach uses database transactions to make the check-and-process atomic:

async def handle_invoice_paid(self, payload: dict, headers: dict):
  event_id = payload["id"]

  async with self.db() as session:
    async with session.begin():
      # SELECT ... FOR UPDATE to prevent concurrent processing
      result = await session.execute(
        select(ProcessedWebhookEvent)
        .where(ProcessedWebhookEvent.event_id == event_id)
        .with_for_update(skip_locked=True)
      )
      if result.scalar_one_or_none():
        return {"status": "already_processed"}

      # Process and record atomically
      invoice = payload["data"]["object"]
      session.add(Payment(external_id=invoice["id"], amount=invoice["amount_paid"]))
      session.add(ProcessedWebhookEvent(event_id=event_id, event_type="invoice.paid"))

  return {"status": "processed", "event_id": event_id}

Scaffolding Webhook Services

The CLI provides a webhook template that generates a complete BaseWebhook subclass with placeholder implementations:

# Create a Stripe webhook receiver
singularity generate service v1/stripe --template webhook -d "Stripe webhook receiver"

# Create a GitHub webhook receiver
singularity generate service v1/github --template webhook -d "GitHub webhook receiver"

# Webhook with database access (Acquire is included by default for webhooks)
singularity generate service v1/shopify --template webhook --db -d "Shopify webhook receiver"

The generated file includes:

Import of BaseWebhook
Import of Acquire with __init__ calling super().__init__(acquire)
A sample events dictionary with a placeholder mapping
A verify() method stub with commented examples for Stripe and GitHub
A get_event_type() method stub with platform-specific examples
A sample handler method for the placeholder event

After scaffolding, replace the placeholder implementations with your platform-specific logic.

The --rpc mixin has no effect on webhook templates. Webhook services set rpc_exposed = False by default because they are designed to receive inbound HTTP from external platforms, not inter-service calls. You can manually set rpc_exposed = True if you need to expose webhook handlers to other services.

Webhook Security Best Practices

Always verify signatures

Never skip signature verification in production. The default verify() returns True as a convenience for development. Always override it with platform-specific HMAC validation before deploying.

Use HTTPS endpoints

Configure your webhook URLs with HTTPS. Webhook payloads often contain sensitive data (customer IDs, payment amounts) that should be encrypted in transit.

Implement idempotency

Assume every event can be delivered more than once. Use event IDs as deduplication keys and check before processing. Store processed event IDs in your database.

Return 200 quickly

Webhook providers have short timeout windows (typically 5-30 seconds). If your handler needs to do heavy work, acknowledge the event immediately and process it asynchronously via a background task.

Async Processing Pattern

For handlers that perform expensive operations, acknowledge the webhook immediately and offload the work to a background task:

class StripeService(BaseWebhook):
  events = {
    "invoice.paid": "handle_invoice_paid",
  }

  def __init__(self, acquire: Acquire):
    super().__init__(acquire)
    self.tasks = acquire.tasks

  async def handle_invoice_paid(self, payload: dict, headers: dict):
    """Acknowledge immediately, process in background."""
    event_id = payload["id"]

    # Offload to background task
    self.tasks.submit(
      "process_stripe_invoice",
      event_id=event_id,
      payload=payload,
    )

    # Return 200 immediately so Stripe does not retry
    return {"status": "accepted", "event_id": event_id}

Testing Webhook Services

Testing webhooks requires constructing requests with valid signatures. Here is a pattern for unit testing:

import hmac
import hashlib
import json
from unittest.mock import AsyncMock, MagicMock
from fastapi import Request


def build_signed_request(payload: dict, secret: str, platform: str = "stripe"):
    """Build a mock Request with a valid webhook signature."""
    raw_body = json.dumps(payload).encode()

    if platform == "stripe":
      timestamp = "1234567890"
      signed_payload = f"{timestamp}.".encode() + raw_body
      sig = hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()
      headers = {"stripe-signature": f"t={timestamp},v1={sig}"}

    elif platform == "github":
      sig = hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest()
      headers = {
        "x-hub-signature-256": f"sha256={sig}",
        "x-github-event": payload.get("_event_type", "push"),
      }

    request = MagicMock(spec=Request)
    request.headers = headers
    request.body = AsyncMock(return_value=raw_body)
    request.json = AsyncMock(return_value=payload)
    return request


async def test_stripe_invoice_paid():
    """Test that a valid Stripe invoice.paid event is processed."""
    acquire = MagicMock()
    acquire.settings.stripe_webhook_secret = "whsec_test_secret"
    acquire.db_session = AsyncMock()

    service = StripeService(acquire=acquire)

    request = build_signed_request(
      payload={
        "id": "evt_123",
        "type": "invoice.paid",
        "data": {
          "object": {
            "id": "inv_abc",
            "customer": "cus_xyz",
            "amount_paid": 2999,
          }
        },
      },
      secret="whsec_test_secret",
      platform="stripe",
    )

    response = await service.post(request)
    assert response.status_code == 200


async def test_invalid_signature_rejected():
    """Test that an invalid signature returns 401."""
    acquire = MagicMock()
    acquire.settings.stripe_webhook_secret = "whsec_test_secret"

    service = StripeService(acquire=acquire)

    request = build_signed_request(
      payload={"type": "invoice.paid", "data": {"object": {"id": "inv_abc"}}},
      secret="wrong_secret",  # Intentionally wrong
      platform="stripe",
    )

    response = await service.post(request)
    assert response.status_code == 401

Complete Reference

BaseWebhook Source

The full BaseWebhook class handles the pipeline in its post() method:

class BaseWebhook:
  rpc_exposed = False
  events: dict[str, str] = {}

  def __init__(self, acquire: Acquire):
    self.acquire = acquire

  async def post(self, request: Request) -> JSONResponse:
    headers = dict(request.headers)
    raw_body = await request.body()

    # Step 1: Verify signature
    if not self.verify(headers, raw_body):
      return JSONResponse(status_code=401, content={"error": "Verification failed"})

    # Step 2: Parse JSON
    try:
      payload = await request.json()
    except Exception:
      return JSONResponse(status_code=400, content={"error": "Invalid JSON payload"})

    # Step 3: Extract event type
    event_type = self.get_event_type(headers, payload)

    # Step 4: Look up handler
    handler_name = self.events.get(event_type)
    if handler_name is None:
      return await self.on_unhandled_event(event_type, payload, headers)

    handler = getattr(self, handler_name, None)
    if handler is None:
      return JSONResponse(status_code=500, content={"error": "Handler misconfigured"})

    # Step 5: Call handler with error boundary
    try:
      result = await handler(payload, headers)
      if isinstance(result, JSONResponse):
        return result
      return JSONResponse(status_code=200, content=result)
    except Exception:
      return JSONResponse(status_code=500, content={"error": "Internal handler error"})

  def verify(self, headers: dict, raw_body: bytes) -> bool:
    return True  # Override in subclass

  def get_event_type(self, headers: dict, payload: dict) -> str:
    return payload.get("type", "unknown")  # Override in subclass

  async def on_unhandled_event(self, event_type, payload, headers) -> JSONResponse:
    return JSONResponse(status_code=200, content={"status": "ignored", "event": event_type})

Platform Verification Reference

def verify(self, headers: dict, raw_body: bytes) -> bool:
  secret = self.acquire.settings.stripe_webhook_secret
  signature = headers.get("stripe-signature", "")
  parts = dict(p.split("=", 1) for p in signature.split(",") if "=" in p)
  timestamp = parts.get("t", "")
  sig_hash = parts.get("v1", "")
  signed_payload = f"{timestamp}.".encode() + raw_body
  expected = hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()
  return hmac.compare_digest(sig_hash, expected)

def verify(self, headers: dict, raw_body: bytes) -> bool:
  secret = self.acquire.settings.github_webhook_secret
  signature = headers.get("x-hub-signature-256", "")
  expected = "sha256=" + hmac.new(
    secret.encode(), raw_body, hashlib.sha256
  ).hexdigest()
  return hmac.compare_digest(signature, expected)

def verify(self, headers: dict, raw_body: bytes) -> bool:
  secret = self.acquire.settings.slack_signing_secret
  timestamp = headers.get("x-slack-request-timestamp", "")
  sig_basestring = f"v0:{timestamp}:".encode() + raw_body
  expected = "v0=" + hmac.new(
    secret.encode(), sig_basestring, hashlib.sha256
  ).hexdigest()
  signature = headers.get("x-slack-signature", "")
  return hmac.compare_digest(signature, expected)

# Stripe -- event type in body
def get_event_type(self, headers, payload):
  return payload.get("type", "unknown")

# GitHub -- event type in header
def get_event_type(self, headers, payload):
  return headers.get("x-github-event", "unknown")

# Slack -- event type nested in body
def get_event_type(self, headers, payload):
  return payload.get("event", {}).get("type", "unknown")

Always verify signatures

Use HTTPS endpoints

Implement idempotency

Return 200 quickly

On this page