Build a Customer Chat Triage Agent With Claude + Intercom Webhooks: End-to-End Python Tutorial (May 2026)

Image: Intercom marketing site og:image (intercom.com), used for editorial coverage of the customer-service platform the FastAPI handler below integrates with.

What you will build

This tutorial walks through a production-shaped triage agent that sits between a customer chat tool and a human support team. Intercom emits a webhook every time a new conversation opens, the FastAPI service verifies the signature, asks Claude to classify the message into one of four buckets, and then either replies with a holding message, assigns the conversation to a team, or closes it as spam.

Per the Anthropic messages API reference, claude-haiku-4-5 is the fastest current model with near-frontier intelligence ¹ , which is the right tier for a classification call that needs to return inside the webhook’s response window. Anthropic also surfaces claude-sonnet-4-6 as the best speed-intelligence balance and claude-opus-4-7 as the frontier tier ² . Haiku is the source-consensus pick for low-latency routing.

The full code is around 120 lines. The pieces are: FastAPI route, HMAC verification helper, Claude classification call, and Intercom Admin API actions.

The four buckets

Bucket	What lands here	Action taken
urgent	Outage reports, security incidents, payment failures blocking access	Assign to the on-call team, reply with a 15-minute SLA acknowledgement
billing	Invoice questions, refund requests, plan changes, tax forms	Assign to the billing team, reply with a holding message
general	Product questions, feature requests, how-to questions	Assign to the support team, reply with a holding message
spam	Promotional pitches, link spam, off-topic blasts	Close the conversation, do not reply

A four-bucket schema is the minimum useful split: narrower and you cannot route, wider and the model loses precision on the edges. Tune the buckets to your team’s actual rotas before shipping.

Prerequisites

• Python 3.11 or newer.
• An Anthropic API key. Sign in at the Anthropic Console to mint one.
• An Intercom workspace with developer access. The Intercom Developer Hub exposes webhook subscriptions and the Admin API token under the same workspace ³ .
• A public HTTPS endpoint to receive webhooks. For development, use a tunnel such as ngrok or Cloudflare Tunnel; for production, deploy the FastAPI app behind a Uvicorn worker per the Uvicorn deployment guide ⁴ .

Install the dependencies:

pip install fastapi uvicorn[standard] anthropic httpx python-dotenv

fastapi and uvicorn are the web stack. anthropic is the official Python SDK. httpx is the async HTTP client used to call the Intercom Admin API. python-dotenv loads the API keys from a local .env file in development.

Step 1: Project scaffold

Create the project layout:

mkdir chat-triage && cd chat-triage
touch app.py classify.py intercom.py signing.py .env

The split keeps each file responsible for one concern. app.py owns the route, classify.py calls Claude, intercom.py calls Intercom, and signing.py verifies the webhook signature.

Populate .env:

ANTHROPIC_API_KEY=sk-ant-...
INTERCOM_ACCESS_TOKEN=dG9rZW5fdmFsdWU=
INTERCOM_CLIENT_SECRET=your-app-client-secret
INTERCOM_TEAM_ONCALL=12345
INTERCOM_TEAM_BILLING=67890
INTERCOM_TEAM_SUPPORT=11111
INTERCOM_ADMIN_ID=99999

The client secret is the value Intercom surfaces under your app’s Basic Information page; it is the key used for HMAC signature verification, distinct from the access token used for Admin API calls.

Image: Intercom Blog — Introducing the Fin API platform (intercom.com/blog), used for editorial coverage of the platform whose webhook surface the FastAPI receiver below consumes.

Step 2: Verify the Intercom signature

Per the Intercom developer documentation, webhook payloads ship with an X-Hub-Signature header containing a SHA-1 HMAC of the raw request body, keyed by the app’s client secret ⁵ . Verification has to use the raw bytes; re-serialising a parsed JSON object produces a different byte sequence and a different signature.

# signing.py
import hashlib
import hmac
import os


def verify_intercom_signature(raw_body: bytes, header_value: str) -> bool:
    """Constant-time HMAC-SHA1 check against the Intercom client secret."""
    secret = os.environ["INTERCOM_CLIENT_SECRET"].encode("utf-8")
    expected = hmac.new(secret, raw_body, hashlib.sha1).hexdigest()
    if not header_value:
        return False
    prefix, _, supplied = header_value.partition("=")
    if prefix != "sha1" or not supplied:
        return False
    return hmac.compare_digest(expected, supplied)

Three details worth flagging:

1. hmac.compare_digest is constant-time, which closes a timing-attack side channel that a plain == comparison would leave open.
2. The header value is sha1=<hex>; strip the prefix before comparing.
3. The function takes raw bytes, not a parsed model. The FastAPI route below reads the raw body before any JSON parsing.

Step 3: The Claude classification call

The classification prompt is a system message that tells Claude the four buckets, the routing meaning of each, and the exact JSON shape to return. A small JSON envelope is easier to parse reliably than free-form prose, and it forces the model to commit to a single label.

# classify.py
import json
import os
from anthropic import Anthropic

client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

SYSTEM_PROMPT = """You triage incoming customer-support messages into
exactly one of four buckets and return a strict JSON object with two
keys: bucket and reason.

Buckets:
- urgent: outage reports, security incidents, payment failures that
  block access, account lockouts in progress.
- billing: invoice questions, refund requests, plan-change requests,
  tax or compliance documents.
- general: product questions, feature requests, how-to questions,
  onboarding queries.
- spam: promotional pitches, link spam, irrelevant blasts.

Return only the JSON object. No prose, no markdown fences.
Schema: {"bucket": "urgent|billing|general|spam", "reason": "string"}
"""


def classify(message_body: str) -> dict:
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=200,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": message_body}],
    )
    text = response.content[0].text.strip()
    try:
        parsed = json.loads(text)
    except json.JSONDecodeError:
        return {"bucket": "general", "reason": "classifier-fallback"}
    if parsed.get("bucket") not in {"urgent", "billing", "general", "spam"}:
        return {"bucket": "general", "reason": "out-of-schema"}
    return parsed

Two safety nets matter here. First, if Claude returns anything that is not parseable JSON, the function falls back to the general bucket; failing into the safest route prevents a malformed classifier response from dropping a message on the floor. Second, an out-of-schema bucket value triggers the same fallback. The reason field is kept for downstream logging.

Per the Anthropic messages API reference, the system parameter is top-level: a sibling of messages, not a message role ⁶ . The claude-haiku-4-5 identifier is the current low-latency tier per Anthropic’s models overview ⁷ .

Image: Anthropic platform docs — Models overview (platform.claude.com), used for editorial coverage of the model-tier choices discussed in the classifier.

Step 4: Talk to the Intercom Admin API

Per the Intercom REST API reference, replying to a conversation is a POST to https://api.intercom.io/conversations/{id}/reply ⁸ , assignment is a POST to https://api.intercom.io/conversations/{id}/parts with message_type: assignment ⁹ , and closing a conversation uses the same parts endpoint with message_type: close ¹⁰ .

# intercom.py
import os
import httpx

BASE = "https://api.intercom.io"


def headers() -> dict:
    return {
        "Authorization": f"Bearer {os.environ['INTERCOM_ACCESS_TOKEN']}",
        "Content-Type": "application/json",
        "Intercom-Version": "2.13",
    }


async def reply(conversation_id: str, body: str) -> None:
    url = f"{BASE}/conversations/{conversation_id}/reply"
    payload = {
        "message_type": "comment",
        "type": "admin",
        "admin_id": os.environ["INTERCOM_ADMIN_ID"],
        "body": body,
    }
    async with httpx.AsyncClient(timeout=10.0) as http:
        response = await http.post(url, json=payload, headers=headers())
        response.raise_for_status()


async def assign(conversation_id: str, team_id: str) -> None:
    url = f"{BASE}/conversations/{conversation_id}/parts"
    payload = {
        "message_type": "assignment",
        "type": "admin",
        "admin_id": os.environ["INTERCOM_ADMIN_ID"],
        "assignee_id": team_id,
    }
    async with httpx.AsyncClient(timeout=10.0) as http:
        response = await http.post(url, json=payload, headers=headers())
        response.raise_for_status()


async def close(conversation_id: str) -> None:
    url = f"{BASE}/conversations/{conversation_id}/parts"
    payload = {
        "message_type": "close",
        "type": "admin",
        "admin_id": os.environ["INTERCOM_ADMIN_ID"],
    }
    async with httpx.AsyncClient(timeout=10.0) as http:
        response = await http.post(url, json=payload, headers=headers())
        response.raise_for_status()

A few notes on the Admin API surface. Set an explicit Intercom-Version header so a future schema bump does not silently break the integration. Use a short HTTP timeout so a slow Intercom response does not pile up webhook retries on the receiver. The Bearer token is the workspace access token; rotate it through your secrets manager rather than the Developer Hub UI for production environments.

Step 5: Wire the FastAPI route

The route reads the raw bytes for signature verification, then parses the JSON and dispatches the four-bucket routing logic. Per FastAPI’s documentation on using the Request object directly, await request.body() is the canonical way to access the raw request body in an async handler ¹¹ .

# app.py
import json
import os
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request

from classify import classify
from intercom import assign, close, reply
from signing import verify_intercom_signature

load_dotenv()
app = FastAPI()

HOLDING_REPLIES = {
    "urgent": (
        "Thanks for flagging this. The on-call team is paged and will "
        "respond within 15 minutes."
    ),
    "billing": (
        "Thanks for reaching out. The billing team will reply within "
        "one business day."
    ),
    "general": (
        "Thanks for the question. A teammate will reply within four "
        "business hours."
    ),
}

TEAM_FOR_BUCKET = {
    "urgent": os.environ.get("INTERCOM_TEAM_ONCALL"),
    "billing": os.environ.get("INTERCOM_TEAM_BILLING"),
    "general": os.environ.get("INTERCOM_TEAM_SUPPORT"),
}


@app.post("/webhooks/intercom")
async def receive_webhook(request: Request):
    raw = await request.body()
    signature = request.headers.get("X-Hub-Signature", "")
    if not verify_intercom_signature(raw, signature):
        raise HTTPException(status_code=401, detail="bad signature")

    event = json.loads(raw)
    topic = event.get("topic", "")
    if topic != "conversation.user.created":
        return {"status": "ignored", "topic": topic}

    item = event["data"]["item"]
    conversation_id = item["id"]
    source_body = item.get("source", {}).get("body", "")

    verdict = classify(source_body)
    bucket = verdict["bucket"]

    if bucket == "spam":
        await close(conversation_id)
        return {"status": "closed", "bucket": "spam"}

    team_id = TEAM_FOR_BUCKET[bucket]
    if team_id:
        await assign(conversation_id, team_id)
    await reply(conversation_id, HOLDING_REPLIES[bucket])
    return {"status": "routed", "bucket": bucket}

The route’s contract: respond fast (the classification + two Intercom calls finish well inside Intercom’s webhook timeout window in practice), always return a 200 once the signature passes (so Intercom does not retry on transient downstream errors), and degrade safely into the general bucket when the classifier wobbles.

Image: Anthropic — Claude Haiku news page (anthropic.com/news), used for editorial coverage of the model-tier identifier called from the classifier above.

Step 6: Run it locally

Start the receiver:

uvicorn app:app --reload --port 8000

In a second terminal, open an ngrok tunnel:

ngrok http 8000

Take the public HTTPS URL ngrok prints and register it under your Intercom app’s Webhook subscriptions page, with the conversation.user.created topic selected. Intercom will fire a verification ping; if the signature helper rejects it, double-check the client secret value pasted into .env.

To test end-to-end, open your Intercom Messenger preview, send a test message that matches one of the four buckets, and watch the Uvicorn logs. A successful run logs a 200 response, the bucket name, and the conversation id.

Step 7: Deploy

For a small workload, a single Uvicorn worker on a small VM (Fly.io, Railway, Cloudflare Workers via the FastAPI adapter, or a $5 DigitalOcean droplet) is enough. Per the Uvicorn deployment guide, production deployments should run behind a process supervisor with --workers set to roughly 2 \* CPU_count ¹² :

uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4

Three operational details worth tracking once you ship:

• Latency budget. The classification call plus the two Intercom calls should finish well under 10 seconds in practice; if you see timeouts, switch the classifier to claude-haiku-4-5 (already the default here) and shorten the system prompt.
• Retry posture. Intercom retries webhooks on a non-2xx response. Return 200 once the signature passes, even when the downstream Admin API call fails; log the failure and surface it to your monitoring stack rather than letting Intercom re-fire the event.
• Idempotency. Webhook retries can fire after the conversation is already routed. Either track the conversation id in a small Redis set with a short TTL, or check the Intercom conversation’s team_assignee_id before re-assigning.

Image: Python documentation — hmac module (docs.python.org), used for editorial coverage of the constant-time HMAC-SHA1 verification helper used in signing.py.

What to add next

The triage agent above is the minimum viable shape. Common additions that the source documentation supports:

• Multi-turn context. The current classifier reads only the first message body. Pull the full conversation_parts array from the Intercom payload and feed the last three turns into the system prompt for higher precision on edge cases.
• Custom attributes. Intercom’s Admin API supports updating conversation attributes; add the classifier’s reason field as a searchable attribute so support leads can audit routing decisions.
• Confidence threshold. Ask Claude to return a confidence field alongside the bucket. Route low-confidence classifications into a human-review queue rather than auto-assigning.
• Escalation to Sonnet. Keep Haiku for the first-pass bucket but escalate ambiguous responses to claude-sonnet-4-6 for a second opinion when the confidence field drops below a threshold.

The pattern generalises beyond Intercom. Front, Zendesk, Help Scout, and Crisp expose similar webhook + REST surfaces; swap the intercom.py module for the equivalent vendor client and the rest of the code carries over without changes.