Step-by-step guide to creating a webhook endpoint and receiving AgentPost events

Webhook Setup Guide

This guide walks you through setting up a webhook receiver to handle AgentPost events in your application.

Step 1: Create a Webhook Receiver

First, create an HTTP endpoint in your application that can receive POST requests. Your endpoint must return a 2xx status code within 30 seconds to acknowledge receipt.

import { Hono } from 'hono';
import { createHmac, timingSafeEqual } from 'crypto';

const app = new Hono();

app.post('/webhooks/agentpost', async (c) => {
  const payload = await c.req.text();
  const signature = c.req.header('x-agentpost-signature');
  const timestamp = c.req.header('x-agentpost-timestamp');

  // Verify signature (see Verifying Webhooks guide)
  if (!verifySignature(payload, signature, timestamp)) {
    return c.json({ error: 'Invalid signature' }, 401);
  }

  const event = JSON.parse(payload);

  switch (event.type) {
    case 'message.received':
      await handleIncomingEmail(event.data);
      break;
    case 'message.bounced':
      await handleBounce(event.data);
      break;
    case 'message.complained':
      await handleComplaint(event.data);
      break;
    default:
      console.log(`Unhandled event type: ${event.type}`);
  }

  return c.json({ received: true });
});

async function handleIncomingEmail(data: any) {
  console.log(`New email from ${data.from_address}: ${data.subject}`);
  // Process the email with your AI agent
}

async function handleBounce(data: any) {
  console.log(`Bounce: ${data.bounce_type} for ${data.bounced_recipients[0].email_address}`);
  // Remove bounced address from your contact lists
}

async function handleComplaint(data: any) {
  console.log(`Complaint from ${data.complained_recipients[0]}`);
  // Stop sending to this recipient
}

export default app;

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

app = Flask(__name__)

@app.route('/webhooks/agentpost', methods=['POST'])
def handle_webhook():
    payload = request.get_data(as_text=True)
    signature = request.headers.get('x-agentpost-signature')
    timestamp = request.headers.get('x-agentpost-timestamp')

    # Verify signature (see Verifying Webhooks guide)
    if not verify_signature(payload, signature, timestamp):
        return jsonify({"error": "Invalid signature"}), 401

    event = json.loads(payload)

    if event["type"] == "message.received":
        handle_incoming_email(event["data"])
    elif event["type"] == "message.bounced":
        handle_bounce(event["data"])
    elif event["type"] == "message.complained":
        handle_complaint(event["data"])
    else:
        print(f"Unhandled event type: {event['type']}")

    return jsonify({"received": True})


def handle_incoming_email(data):
    print(f"New email from {data['from_address']}: {data['subject']}")
    # Process the email with your AI agent


def handle_bounce(data):
    recipient = data["bounced_recipients"][0]["email_address"]
    print(f"Bounce: {data['bounce_type']} for {recipient}")
    # Remove bounced address from your contact lists


def handle_complaint(data):
    print(f"Complaint from {data['complained_recipients'][0]}")
    # Stop sending to this recipient

# Test your endpoint with a simulated event payload
curl -X POST http://localhost:3000/webhooks/agentpost \
  -H "Content-Type: application/json" \
  -H "x-agentpost-signature: test_signature" \
  -H "x-agentpost-timestamp: 1709910600" \
  -d '{
    "id": "evt_01JQ8X5K2M3N4P5R6S7T8V9W",
    "type": "message.received",
    "timestamp": "2026-03-08T14:30:00.000Z",
    "org_id": "org_01JQ8X5K2M3N4P5R6S7T8V9W",
    "env_id": "env_01JQ8X5K2M3N4P5R6S7T8V9W",
    "data": {
      "message_id": "msg_01JQ8X5K2M3N4P5R6S7T8V9W",
      "thread_id": "thr_01JQ8X5K2M3N4P5R6S7T8V9W",
      "inbox_id": "inb_01JQ8X5K2M3N4P5R6S7T8V9W",
      "from_address": "customer@acmeco.com",
      "to_addresses": ["support@agent-post.dev"],
      "subject": "Need help with billing",
      "extracted_text": "Hi, I need help with my billing.",
      "has_attachments": false,
      "received_at": "2026-03-08T14:30:00.000Z"
    }
  }'

Step 2: Register the Endpoint with AgentPost

Once your receiver is deployed and accessible via a public URL, register it with AgentPost:

import AgentPost from '@agentpost/sdk';

const client = new AgentPost({ apiKey: 'ap_sk_live_your_key_here' });

const endpoint = await client.webhookEndpoints.create({
  url: 'https://your-app.com/webhooks/agentpost',
  events: ['message.received', 'message.bounced', 'message.complained'],
  description: 'Production email event handler',
});

// Store the secret securely -- it is only shown once
console.log('Webhook secret:', endpoint.secret);
// Output: whsec_abc123def456...

from agentpost import AgentPost

client = AgentPost(api_key="ap_sk_live_your_key_here")

endpoint = client.webhook_endpoints.create(
    url="https://your-app.com/webhooks/agentpost",
    events=["message.received", "message.bounced", "message.complained"],
    description="Production email event handler",
)

# Store the secret securely -- it is only shown once
print(f"Webhook secret: {endpoint.secret}")
# Output: whsec_abc123def456...

curl -X POST https://api.agent-post.dev/api/v1/webhooks/endpoints \
  -H "Authorization: Bearer ap_sk_live_your_key_here" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-app.com/webhooks/agentpost",
    "events": ["message.received", "message.bounced", "message.complained"],
    "description": "Production email event handler"
  }'

# Response includes the webhook secret (shown only once):
# { "id": "whe_...", "secret": "whsec_...", ... }

Step 3: Handle Events in Your Application

Structure your webhook handler to route events efficiently. For AI agent workflows, the message.received event is typically your primary trigger:

import AgentPost from '@agentpost/sdk';

const client = new AgentPost({ apiKey: 'ap_sk_live_your_key_here' });

async function handleIncomingEmail(data: {
  message_id: string;
  thread_id: string;
  inbox_id: string;
  from_address: string;
  subject: string;
  extracted_text: string;
}) {
  // 1. Get the full message details
  const message = await client.messages.get(data.inbox_id, data.message_id);

  // 2. Process with your AI agent
  const agentResponse = await yourAIAgent.process({
    from: message.from_address,
    subject: message.subject,
    body: message.extracted_text ?? message.text_body,
    threadId: message.thread_id,
  });

  // 3. Send a reply
  await client.messages.reply(data.inbox_id, data.message_id, {
    text_body: agentResponse.text,
    html_body: agentResponse.html,
  });
}

from agentpost import AgentPost

client = AgentPost(api_key="ap_sk_live_your_key_here")


def handle_incoming_email(data):
    # 1. Get the full message details
    message = client.messages.get(data["inbox_id"], data["message_id"])

    # 2. Process with your AI agent
    agent_response = your_ai_agent.process(
        from_addr=message.from_address,
        subject=message.subject,
        body=message.extracted_text or message.text_body,
        thread_id=message.thread_id,
    )

    # 3. Send a reply
    client.messages.reply(
        data["inbox_id"],
        data["message_id"],
        text_body=agent_response.text,
        html_body=agent_response.html,
    )

# After receiving a message.received webhook, fetch the full message:
curl https://api.agent-post.dev/api/v1/inboxes/inb_01JQ8X/messages/msg_01JQ8X \
  -H "Authorization: Bearer ap_sk_live_your_key_here"

# Then reply to it:
curl -X POST https://api.agent-post.dev/api/v1/inboxes/inb_01JQ8X/messages/msg_01JQ8X/reply \
  -H "Authorization: Bearer ap_sk_live_your_key_here" \
  -H "Content-Type: application/json" \
  -d '{
    "text_body": "Thanks for reaching out! Let me look into that for you."
  }'

Step 4: Monitor Delivery Logs

Check the delivery status of webhook events to debug issues and verify your endpoint is working correctly:

// List recent deliveries for an endpoint
const deliveries = await client.webhookEndpoints.listDeliveries(endpoint.id, {
  limit: 20,
});

for (const delivery of deliveries.data) {
  console.log(`${delivery.event_type}: ${delivery.status} (${delivery.http_status})`);
}

// Example output:
// message.received: delivered (200)
// message.bounced: delivered (200)
// message.received: failed (500)

# List recent deliveries for an endpoint
deliveries = client.webhook_endpoints.list_deliveries(
    endpoint_id=endpoint.id,
    limit=20,
)

for delivery in deliveries.data:
    print(f"{delivery.event_type}: {delivery.status} ({delivery.http_status})")

# Example output:
# message.received: delivered (200)
# message.bounced: delivered (200)
# message.received: failed (500)

# List recent deliveries for a webhook endpoint
curl "https://api.agent-post.dev/api/v1/webhooks/endpoints/whe_01JQ8X/deliveries?limit=20" \
  -H "Authorization: Bearer ap_sk_live_your_key_here"

Best Practices

Return 200 quickly -- Process events asynchronously. Acknowledge receipt immediately and handle the event in a background job.
Handle duplicates -- Use the event id for deduplication. Store processed event IDs and skip duplicates.
Verify signatures -- Always verify webhook signatures in production to ensure events are from AgentPost.
Monitor failures -- Check delivery logs regularly. Endpoints are auto-disabled after 50 consecutive failures.
Use specific events -- Subscribe only to the events you need rather than all events to reduce noise.

Webhook Setup Guide

Webhook Setup Guide

Step 1: Create a Webhook Receiver

Step 2: Register the Endpoint with AgentPost

Step 3: Handle Events in Your Application

Step 4: Monitor Delivery Logs

Best Practices

On this page