The webhook handler is the most consequential hundred lines in any email agent. Everything downstream — classification, drafting, replying — only runs if this one endpoint answers a handshake correctly, acknowledges fast, and never trusts a payload it hasn't checked. FastAPI happens to be a great fit for all three, so here's the handler done right.

Context first: the agent in question lives on an Agent Account — a Nylas-hosted mailbox your application owns, currently in beta. Inbound mail to that address fires a message.created webhook, one of 43 trigger types spanning messages, threads, calendars, events, contacts, folders, and grants. Your FastAPI app is the receiving end.

The contract your endpoint signs

Two non-negotiables come with the subscription:

1. The challenge handshake. When the webhook is created or reactivated, a GET request arrives with a challenge query parameter. You must echo the exact raw value back in a 200 OK body within 10 seconds — no quotes, no JSON wrapper. Fail any part of that and the endpoint is marked failed immediately, with no retry. Passing it generates the webhook_secret you'll use for signature verification.
2. The 10-second acknowledgment. Every notification expects a 200 OK within the same window. Do your real work after responding, not before.

Payload models

Pydantic models make the nested notification shape explicit instead of a pile of payload["data"]["object"] lookups:

from pydantic import BaseModel

class EmailParticipant(BaseModel):
 email: str
 name: str = ""

class MessageObject(BaseModel):
 id: str
 grant_id: str
 subject: str = ""
 from_: list[EmailParticipant] = []
 snippet: str = ""

 model_config = {"populate_by_name": True}

class NotificationData(BaseModel):
 object: MessageObject

class Notification(BaseModel):
 type: str
 data: NotificationData

One wrinkle: the sender field is literally named from in the JSON, which is a Python keyword — alias it. Also keep the models loose (defaults everywhere) because the .truncated delivery variant strips the body when a payload would exceed 1 MB, and your model shouldn't reject it.

The handler

from fastapi import FastAPI, Request, BackgroundTasks
from fastapi.responses import PlainTextResponse
import hmac, hashlib, os

app = FastAPI()
WEBHOOK_SECRET = os.environ["NYLAS_WEBHOOK_SECRET"]

@app.get("/webhooks/nylas", response_class=PlainTextResponse)
async def challenge(request: Request) -> str:
 # Echo the raw value — PlainTextResponse avoids JSON quoting
 return request.query_params.get("challenge", "")

@app.post("/webhooks/nylas")
async def notification(request: Request, background: BackgroundTasks):
 raw = await request.body()

 # Verify before trusting: HMAC-SHA256 of the raw body
 signature = request.headers.get("x-nylas-signature", "")
 expected = hmac.new(WEBHOOK_SECRET.encode(), raw, hashlib.sha256).hexdigest()
 if not hmac.compare_digest(expected, signature):
 return PlainTextResponse("invalid signature", status_code=401)

 event = Notification.model_validate_json(raw)
 if event.type.startswith("message.created"):
 background.add_task(
 handle_inbound,
 event.data.object.grant_id,
 event.data.object.id,
 )
 return {"ok": True}

BackgroundTasks is the load-bearing piece. The response goes out as soon as the function returns; handle_inbound runs afterward, outside the 10-second window. The startswith check is deliberate too — it catches both message.created and the .truncated variant with one branch, which the docs recommend treating identically as "new mail arrived."

Note the signature check runs over the raw request body, not the parsed JSON. Re-serializing the payload changes whitespace and breaks the HMAC. That's why the handler takes Request directly instead of a Pydantic parameter.

Dispatching to the agent

The notification carries metadata, but for the full message you fetch it — which doubles as your handling of truncated deliveries:

import httpx

API_KEY = os.environ["NYLAS_API_KEY"]
BASE = "https://api.us.nylas.com"

async def handle_inbound(grant_id: str, message_id: str):
 async with httpx.AsyncClient() as client:
 r = await client.get(
 f"{BASE}/v3/grants/{grant_id}/messages/{message_id}",
 headers={"Authorization": f"Bearer {API_KEY}"},
 )
 message = r.json()["data"]
 # From here: classify with your LLM, draft, and send the reply
 # via POST /v3/grants/{grant_id}/messages/send — same grant.

Because the reply goes out through the same grant_id, the conversation stays in one mailbox: inbound mail arrives, the agent reasons, the response leaves from the agent's own address.

Closing the loop: sending the reply

The send half is one more httpx call against the same grant. The payload is just subject, body, and recipients:

async def send_reply(grant_id: str, to_email: str, subject: str, body: str):
 async with httpx.AsyncClient() as client:
 r = await client.post(
 f"{BASE}/v3/grants/{grant_id}/messages/send",
 headers={"Authorization": f"Bearer {API_KEY}"},
 json={
 "subject": subject,
 "body": body,
 "to": [{"email": to_email}],
 },
 )
 return r.json()["data"]

The recipient sees a normal message from the agent's address — no "sent via" branding, no relay footer. If the account also has a calendar role, the same grant drives GET /v3/grants/{grant_id}/events and POST /v3/grants/{grant_id}/events/{id}/send-rsvp, so a scheduling agent can accept invitations from the same handler.

Testing without a tunnel

The challenge echo and the signature check are both pure functions of the request, which means they're testable with FastAPI's TestClient before you expose anything to the internet:

from fastapi.testclient import TestClient
import hmac, hashlib, json

client = TestClient(app)

def test_challenge_echo_is_raw():
 r = client.get("/webhooks/nylas?challenge=abc123")
 assert r.status_code == 200
 assert r.text == "abc123" # not '"abc123"' — JSON quoting fails the handshake

def test_rejects_bad_signature():
 r = client.post(
 "/webhooks/nylas",
 content=b'{"type": "message.created", "data": {"object": {}}}',
 headers={"x-nylas-signature": "0" * 64},
 )
 assert r.status_code == 401

def test_accepts_valid_signature():
 body = json.dumps({
 "type": "message.created",
 "data": {"object": {"id": "m1", "grant_id": "g1"}},
 }).encode()
 sig = hmac.new(WEBHOOK_SECRET.encode(), body, hashlib.sha256).hexdigest()
 r = client.post(
 "/webhooks/nylas", content=body,
 headers={"x-nylas-signature": sig},
 )
 assert r.status_code == 200

That first assertion — r.text == "abc123" with no quotes — is the one that catches the classic mistake of returning the challenge through a JSON response class.

Three things that bite later

• Duplicates. Delivery is at-least-once, so identical notifications can arrive twice. Dedupe on the message id before your agent acts, or it'll reply twice.
• Ordering. A message.created and a message.updated for the same message can land out of order. The fetched message is the source of truth, not the notification sequence.
• Local testing. The callback URL must be public HTTPS — Nylas calls it from the internet to verify it. Tunnel to localhost during development and recreate the webhook when the tunnel URL changes.

The new email webhook recipe covers the full delivery semantics if you want the details behind these rules.

Wire this up, email your agent's address from your phone, and watch handle_inbound fire within seconds. Then add the actual brain. What does your handle_inbound do first — classify, or fetch the whole thread?