Receiving webhooks

Receive events from external systems the moment they happen, with no polling anywhere.

Webhooks are HTTP callbacks that external systems send to your application when events occur. Instead of polling an API repeatedly, you register a URL and the external system POSTs data to you when something happens.

You receive webhooks through a REST channel that points to a service. The service receives the incoming JSON payload, processes it and triggers other actions - updating a database, calling another API or publishing a message.

Create a webhook channel

To create the channel, go to Connections > Channels > REST in the Dashboard, click Create a new channel and fill in the form:

Name: Jira Webhooks
URL path: the endpoint the external system calls, e.g. /webhooks/jira
Service: the service that processes incoming events
Data format: JSON
Click OK

Whether the channel needs a Security definition depends on the external system - systems that sign their events instead of authenticating are verified in the service, as shown below.

Basic webhook handler

The service receives the event in self.request.input. The structure is defined by the external system - the example below handles Jira issue events:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class JiraWebhook(Service):
    """ Processes issue events that Jira sends.
    """
    name = 'webhooks.jira.receive'

    def handle(self) -> 'None':

        # Every Jira issue event carries the event type and the issue key
        event = self.request.input
        event_type = event['webhookEvent']
        issue_key = event['issue']['key']

        self.logger.info('Received Jira event: %s', event_type)

        # Each event type has its own handler
        if event_type == 'jira:issue_created':
            self.handle_issue_created(issue_key, event)

        elif event_type == 'jira:issue_updated':
            self.handle_issue_updated(issue_key, event)

    def handle_issue_created(self, issue_key, event) -> 'None':
        self.logger.info('New issue created: %s', issue_key)

    def handle_issue_updated(self, issue_key, event) -> 'None':
        self.logger.info('Issue updated: %s', issue_key)

To test the handler before connecting Jira, send what Jira would:

curl -X POST http://localhost:11223/webhooks/jira \
  -d '{"webhookEvent": "jira:issue_created", "issue": {"key": "PROJ-123"}}'

Map webhook data to your models

A thin webhook handler transforms the event to your internal format and delegates the processing to a service that can be tested independently:

# -*- coding: utf-8 -*-

# stdlib
from dataclasses import dataclass

# Zato
from zato.server.service import Service

@dataclass
class AccessRequest:
    issue_key: str
    email: str
    company_name: str
    request_type: str

class AccessRequestWebhook(Service):
    name = 'webhooks.access-request.receive'

    def handle(self) -> 'None':

        # The event in the external system's format ..
        data = self.request.input

        # .. becomes our internal model ..
        request = AccessRequest(
            issue_key=data['issueKey'],
            email=data['email'],
            company_name=data['companyName'],
            request_type=data['requestType'],
        )

        # .. and a separate service does the actual work.
        self.invoke('access.request.process', request)

Respond quickly, process asynchronously

Webhook providers expect a fast 200 response and retry or disable endpoints that are slow. Hand the heavy processing to self.invoke_async and answer at once:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class ServiceNowWebhook(Service):
    name = 'webhooks.servicenow.receive'

    def handle(self) -> 'None':

        event = self.request.input
        event_type = event['event_type']

        # The processing runs in the background ..
        self.invoke_async('servicenow.incident.process', {
            'event_type': event_type,
            'payload': event
        })

        # .. and the provider receives its 200 immediately.
        self.response.payload = {'status': 'received'}

Verify webhook signatures

Systems that sign their events send a signature header instead of credentials - verify it against the raw request body before processing. The example below receives events from Salesforce:

# -*- coding: utf-8 -*-

# stdlib
import hashlib
import hmac
from http import HTTPStatus

# Zato
from zato.server.service import Service

class SalesforceWebhook(Service):
    """ Verifies and processes signed Salesforce events.
    """
    name = 'webhooks.salesforce.receive'

    def handle(self) -> 'None':

        # The signature arrives in a header and may be absent ..
        signature = self.request.http.headers.get('x-salesforce-signature')

        # .. a request without one is rejected before any processing ..
        if not signature:
            self.response.status_code = HTTPStatus.UNAUTHORIZED
            self.response.payload = {'error': 'Missing signature'}
            return

        # .. the verification runs over the raw request body ..
        raw_body = self.request.raw
        secret = self.config.salesforce.webhook_secret

        if not self.verify_signature(raw_body, signature, secret):
            self.response.status_code = HTTPStatus.UNAUTHORIZED
            self.response.payload = {'error': 'Invalid signature'}
            return

        # .. and only a verified event is processed.
        event = self.request.input
        self.process_salesforce_event(event)

    def verify_signature(self, payload, signature, secret) -> 'bool':
        expected = hmac.new(
            secret.encode(),
            payload.encode(),
            hashlib.sha256
        ).hexdigest()
        return hmac.compare_digest(expected, signature)

    def process_salesforce_event(self, event) -> 'None':
        event_type = event['type']
        self.logger.info('Processing Salesforce event: %s', event_type)

Test webhooks locally

Each handler can be exercised with curl before the real external system is connected:

# Simulate a ServiceNow incident event
curl -X POST http://localhost:11223/webhooks/servicenow \
  -d '{"event_type": "incident.created", "incident": {"number": "INC0010001", "priority": "2"}}'

# Simulate a Salesforce opportunity event
curl -X POST http://localhost:11223/webhooks/salesforce \
  -H "X-Salesforce-Signature: your-test-signature" \
  -d '{"type": "opportunity.closed_won", "data": {"opportunity_id": "006xxx", "amount": 50000}}'

Page	What it covers
REST channels	The channels that webhook events arrive through
Custom authentication	The HMAC verification patterns signed webhooks use
Error handling	What the provider receives when processing fails

Learn more

REST integration overview Start here to see how teams build REST integrations

In-depth API tutorial Master architectural patterns for production-ready integrations

Modern REST API tutorial Design, build and secure REST APIs with Python