Webhook Setup

Receive incoming Peppol documents in real-time by configuring webhooks.

How It Works

┌───────────────┐   ┌───────────────┐   ┌───────────────┐   ┌───────────────┐
│  Sender's     │──▶│   GoRoute     │──▶│   Webhook     │──▶│  Your App     │
│  Access Point │   │   Receives    │   │   Delivery    │   │  Processes    │
└───────────────┘   └───────────────┘   └───────────────┘   └───────────────┘

1. A sender transmits a document via their Access Point
2. GoRoute receives the document at your registered Peppol ID
3. GoRoute validates and processes the document
4. GoRoute sends the document to your webhook endpoint
5. Your application processes the incoming document

Register a Webhook

import requests

webhook_config = {
    "url": "https://your-app.com/webhooks/peppol",
    "events": ["document.received"],
    "secret": "whsec_your_random_secret_here",
    "active": True
}

response = requests.post(
    "https://app.goroute.ai/peppol-api/api/v1/webhooks",
    headers={
        "X-API-Key": "your_api_key",
        "Content-Type": "application/json"
    },
    json=webhook_config
)

result = response.json()
print(f"Webhook ID: {result['id']}")
print(f"Endpoint: {result['url']}")

Webhook Configuration

Field	Type	Required	Description
url	string	Yes	HTTPS URL to receive webhooks
events	array	Yes	List of events to receive
secret	string	Recommended	Secret for signature verification
active	boolean	No	Enable/disable webhook (default: true)
headers	object	No	Custom headers to include

Webhook Events

Event	Description
document.received	New document received via Peppol
document.validated	Document passed validation
document.failed_validation	Document failed validation
transaction.delivered	Outgoing document delivered
transaction.failed	Outgoing document failed
participant.updated	Participant registration changed

Webhook Payload

When a document is received, GoRoute sends:

{
    "id": "evt_abc123",
    "type": "document.received",
    "created_at": "2024-01-15T10:30:45Z",
    "data": {
        "document_id": "doc_xyz789",
        "document_type": "Invoice",
        "sender": {
            "scheme": "0106",
            "identifier": "12345678",
            "name": "Sender Company BV"
        },
        "receiver": {
            "scheme": "0106",
            "identifier": "87654321",
            "name": "Your Company BV"
        },
        "received_at": "2024-01-15T10:30:45Z",
        "metadata": {
            "invoice_id": "INV-2024-00123",
            "issue_date": "2024-01-15",
            "due_date": "2024-02-15",
            "currency": "EUR",
            "total_amount": "121.00"
        }
    }
}

Implement Your Endpoint

Basic Handler

from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)
WEBHOOK_SECRET = "whsec_your_random_secret_here"

@app.route("/webhooks/peppol", methods=["POST"])
def handle_webhook():
    # Verify the webhook signature
    signature = request.headers.get("X-GoRoute-Signature")
    if not verify_signature(request.data, signature):
        return jsonify({"error": "Invalid signature"}), 401
    
    # Parse the event
    event = request.json
    event_type = event["type"]
    
    # Handle different event types
    if event_type == "document.received":
        handle_incoming_document(event["data"])
    elif event_type == "transaction.delivered":
        handle_delivery_confirmation(event["data"])
    elif event_type == "transaction.failed":
        handle_delivery_failure(event["data"])
    
    # Return 200 to acknowledge receipt
    return jsonify({"status": "received"})


def verify_signature(payload: bytes, signature: str) -> bool:
    """Verify the webhook signature."""
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, f"sha256={expected}")


def handle_incoming_document(data: dict):
    """Process an incoming Peppol document."""
    document_id = data["document_id"]
    document_type = data["document_type"]
    
    print(f"Received {document_type}: {document_id}")
    
    # Fetch the full document content
    document = fetch_document(document_id)
    
    # Process in your system
    process_document(document)


def fetch_document(document_id: str) -> dict:
    """Fetch the full document from GoRoute."""
    response = requests.get(
        f"https://app.goroute.ai/peppol-api/api/v1/documents/{document_id}",
        headers={"X-API-Key": "your_api_key"}
    )
    return response.json()

Async Handler (Recommended)

from fastapi import FastAPI, Request, BackgroundTasks
import httpx

app = FastAPI()

@app.post("/webhooks/peppol")
async def handle_webhook(request: Request, background_tasks: BackgroundTasks):
    # Verify signature
    signature = request.headers.get("X-GoRoute-Signature")
    body = await request.body()
    
    if not verify_signature(body, signature):
        return {"error": "Invalid signature"}, 401
    
    event = await request.json()
    
    # Process in background (respond quickly)
    background_tasks.add_task(process_event, event)
    
    # Return immediately
    return {"status": "received"}


async def process_event(event: dict):
    """Process the webhook event in the background."""
    event_type = event["type"]
    
    if event_type == "document.received":
        await process_incoming_document(event["data"])


async def process_incoming_document(data: dict):
    """Fetch and process the incoming document."""
    document_id = data["document_id"]
    
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://app.goroute.ai/peppol-api/api/v1/documents/{document_id}",
            headers={"X-API-Key": "your_api_key"}
        )
        document = response.json()
    
    # Store in your database
    await save_to_database(document)
    
    # Trigger your business logic
    await notify_accounts_payable(document)

Security Best Practices

1. Verify Signatures

Always verify webhook signatures:

def verify_signature(payload: bytes, signature: str) -> bool:
    """Verify the webhook signature using HMAC-SHA256."""
    if not signature or not signature.startswith("sha256="):
        return False
    
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    received = signature[7:]  # Remove "sha256=" prefix
    return hmac.compare_digest(expected, received)

2. Use HTTPS

Your webhook endpoint MUST use HTTPS with a valid SSL certificate.

3. Implement Idempotency

Handle duplicate deliveries gracefully:

def handle_webhook(event: dict):
    event_id = event["id"]
    
    # Check if already processed
    if db.event_exists(event_id):
        return {"status": "already_processed"}
    
    # Process the event
    process_event(event)
    
    # Mark as processed
    db.mark_event_processed(event_id)
    
    return {"status": "processed"}

4. Respond Quickly

Return a 2xx response within 30 seconds:

# ❌ Don't do heavy processing in the request
@app.post("/webhooks/peppol")
def webhook():
    event = request.json
    process_everything_synchronously(event)  # Slow!
    return {"status": "ok"}

# ✅ Process in background, respond immediately
@app.post("/webhooks/peppol")
def webhook():
    event = request.json
    queue.enqueue(process_event, event)  # Background task
    return {"status": "ok"}  # Fast response

Retry Policy

GoRoute retries failed webhook deliveries:

Attempt	Delay
1st retry	1 minute
2nd retry	5 minutes
3rd retry	30 minutes
4th retry	2 hours
5th retry	12 hours

After 5 failed attempts, the webhook is marked as failed and requires manual review.

Retry Headers

X-GoRoute-Delivery-Attempt: 2
X-GoRoute-First-Attempt-At: 2024-01-15T10:30:45Z

Testing Webhooks

Send Test Event

# Trigger a test webhook
response = requests.post(
    "https://app.goroute.ai/peppol-api/api/v1/webhooks/{webhook_id}/test",
    headers={"X-API-Key": "your_api_key"}
)

print(f"Test event sent: {response.json()}")

Local Development

Use ngrok or similar to expose your local server:

# Install ngrok
npm install -g ngrok

# Expose your local server
ngrok http 3000

# Use the ngrok URL as your webhook endpoint
# https://abc123.ngrok.io/webhooks/peppol

Managing Webhooks

List Webhooks

response = requests.get(
    "https://app.goroute.ai/peppol-api/api/v1/webhooks",
    headers={"X-API-Key": "your_api_key"}
)

webhooks = response.json()
for wh in webhooks["items"]:
    print(f"{wh['id']}: {wh['url']} - {wh['status']}")

Update Webhook

response = requests.patch(
    f"https://app.goroute.ai/peppol-api/api/v1/webhooks/{webhook_id}",
    headers={
        "X-API-Key": "your_api_key",
        "Content-Type": "application/json"
    },
    json={
        "url": "https://new-endpoint.com/webhooks/peppol",
        "events": ["document.received", "transaction.delivered"]
    }
)

Delete Webhook

response = requests.delete(
    f"https://app.goroute.ai/peppol-api/api/v1/webhooks/{webhook_id}",
    headers={"X-API-Key": "your_api_key"}
)

Webhook Logs

View recent webhook deliveries:

response = requests.get(
    f"https://app.goroute.ai/peppol-api/api/v1/webhooks/{webhook_id}/logs",
    params={"limit": 50},
    headers={"X-API-Key": "your_api_key"}
)

logs = response.json()
for log in logs["items"]:
    print(f"{log['created_at']}: {log['status_code']} - {log['event_type']}")

Next Steps

• Processing Documents
• Responding to Invoices
• Error Handling