Generic Webhook Integration

Use this guide when your helpdesk isn't covered by the platform-specific guides, or when you're building a custom integration from scratch.

Webhook receiver pattern

The core pattern is a small FastAPI service — a "webhook bridge" — that receives events from your helpdesk, normalizes them into Simpli's domain model, calls the relevant Simpli services, and pushes results back.

Complete webhook bridge

"""Simpli webhook bridge — connects any helpdesk to Simpli services."""

import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException, Header
from pydantic import BaseModel
import httpx
import structlog

log = structlog.get_logger()

app = FastAPI(title="Simpli Webhook Bridge")

# Service URLs — configure via environment variables in production
TRIAGE_URL = "http://triage:8000"
REPLY_URL = "http://reply:8000"
SENTIMENT_URL = "http://sentiment:8000"
QA_URL = "http://qa:8000"

WEBHOOK_SECRET = "your-webhook-secret"  # validate incoming webhooks


class NormalizedTicket(BaseModel):
    ticket_id: str
    subject: str
    body: str
    customer_id: str
    channel: str = "web"


def normalize_payload(payload: dict) -> NormalizedTicket:
    """Map any helpdesk payload to a standard format."""
    return NormalizedTicket(
        ticket_id=str(payload.get("ticket_id") or payload.get("id") or payload.get("case_id", "")),
        subject=payload.get("subject") or payload.get("title", ""),
        body=payload.get("body") or payload.get("description") or payload.get("content", ""),
        customer_id=str(payload.get("customer_id") or payload.get("requester_id") or payload.get("contact_id", "")),
        channel=payload.get("channel", "web"),
    )


@app.post("/webhook/ticket-created")
async def on_ticket_created(request: Request):
    """Handle new ticket creation."""
    payload = await request.json()
    ticket = normalize_payload(payload)
    log.info("ticket_received", ticket_id=ticket.ticket_id)

    async with httpx.AsyncClient(timeout=10.0) as client:
        classification = (await client.post(f"{TRIAGE_URL}/classify", json={
            "subject": ticket.subject,
            "body": ticket.body,
        })).json()

        routing = (await client.post(f"{TRIAGE_URL}/route", json={
            "ticket_id": ticket.ticket_id,
            **classification,
        })).json()

        sentiment = (await client.post(f"{SENTIMENT_URL}/analyze", json={
            "customer_id": ticket.customer_id,
            "text": f"{ticket.subject}\n{ticket.body}",
        })).json()

    log.info("ticket_processed", ticket_id=ticket.ticket_id,
             category=classification["category"], team=routing["team"],
             sentiment=sentiment["label"])

    # TODO: Push results back to your helpdesk via its API
    return {"ticket_id": ticket.ticket_id, "classification": classification,
            "routing": routing, "sentiment": sentiment}


@app.post("/webhook/ticket-resolved")
async def on_ticket_resolved(request: Request):
    """Handle ticket resolution — trigger QA evaluation."""
    payload = await request.json()
    messages = payload.get("messages") or payload.get("comments", [])

    normalized_messages = [
        {"role": msg.get("role", "customer"), "content": msg.get("content") or msg.get("body", "")}
        for msg in messages
    ]

    async with httpx.AsyncClient(timeout=30.0) as client:
        score = (await client.post(f"{QA_URL}/evaluate", json={
            "conversation_id": str(payload.get("ticket_id", "")),
            "agent_id": str(payload.get("agent_id", "")),
            "messages": normalized_messages,
        })).json()

    return {"score": score}

Payload normalization

The normalize_payload function handles the fact that different helpdesks use different field names for the same concepts. It tries multiple keys in priority order and falls back to empty strings.

Field	Zendesk	Freshdesk	Intercom	Salesforce
Ticket ID	`id`	`id`	`id`	`CaseNumber`
Subject	`subject`	`subject`	`title`	`Subject`
Body	`description`	`description`	`body`	`Description`
Customer	`requester_id`	`requester_id`	`contact_id`	`ContactId`

If your helpdesk uses field names not covered by the defaults, extend the normalize_payload function with your platform's conventions. For Salesforce and other platforms with PascalCase fields, you may want to add a case-normalization step before lookup.

Webhook security

Always validate that incoming webhooks are genuine. Most helpdesk platforms sign their webhook payloads with HMAC-SHA256.

def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify HMAC-SHA256 webhook signature."""
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, signature)


@app.middleware("http")
async def check_webhook_signature(request: Request, call_next):
    if request.url.path.startswith("/webhook/"):
        body = await request.body()
        signature = request.headers.get("x-webhook-signature", "")
        if not verify_signature(body, signature, WEBHOOK_SECRET):
            raise HTTPException(status_code=401, detail="Invalid signature")
    return await call_next(request)

Additional security recommendations:

Use HTTPS for all webhook endpoints.
IP allowlisting: Restrict incoming traffic to your helpdesk's known IP ranges.
Rotate secrets periodically and support multiple active secrets during rotation windows.

Error handling and retries

The bridge code above uses httpx with explicit timeouts (timeout=10.0 for classification, timeout=30.0 for QA scoring). Beyond that, consider these patterns:

Idempotency: Track processed ticket_id values (e.g., in Redis or a database) to avoid duplicate processing when helpdesks retry failed webhook deliveries.
Structured logging: The bridge uses structlog so every log line includes ticket_id, making it easy to trace issues.
Graceful degradation: If one Simpli service is down, catch the exception and continue with the others rather than failing the entire webhook.
Message queue: For high-volume deployments, put incoming webhooks onto a queue (Redis + Celery, or a managed service like SQS) and process them asynchronously. This decouples webhook receipt from processing and gives you automatic retries.

Pushing results back

After Simpli processes a ticket, you'll want to update your helpdesk with the results. The exact API call depends on your platform, but the pattern is the same — an HTTP request to your helpdesk's API:

async def update_helpdesk_ticket(ticket_id: str, classification: dict, sentiment: dict):
    """Push Simpli results back to the helpdesk."""
    async with httpx.AsyncClient() as client:
        await client.put(
            f"https://your-helpdesk.com/api/tickets/{ticket_id}",
            headers={"Authorization": "Bearer YOUR_API_TOKEN"},
            json={
                "tags": [classification["category"], sentiment["label"]],
                "priority": classification.get("priority", "normal"),
                "custom_fields": {
                    "ai_category": classification["category"],
                    "ai_sentiment": sentiment["label"],
                    "ai_confidence": classification.get("confidence"),
                },
            },
        )

Call this function at the end of your on_ticket_created handler, replacing the # TODO comment in the bridge code.

Running the bridge

pip install fastapi uvicorn httpx structlog
uvicorn bridge:app --host 0.0.0.0 --port 9000

For production, run behind a reverse proxy (nginx, Caddy) with TLS termination and use a process manager like PM2, systemd, or Docker.

Next steps

Platform-specific guides for Zendesk, Freshdesk, Intercom, Salesforce, and HubSpot
Docker Compose setup for running the bridge alongside Simpli services
See the Integration Overview for the full architecture diagram

Generic Webhook Integration

On this page