Webhooks

Receive AI-generated results from Convoy via HTTP callbacks. This is the foundational pattern that all integrations build upon.

Testing tip — Build and verify this integration end-to-end with convoy-mock first. It’s a free, synthetic model that returns a callback in ~60 seconds and is never billed. Once your wiring works, swap "model": "convoy-mock" for a production model like claude-3-haiku.

How Convoy Webhooks Work

When you submit a prompt to Convoy, you include a callback_url. Once batch processing completes, Convoy sends an HTTP POST request to that URL with the results.


┌──────────┐         POST /cargo/load          ┌─────────┐
│  Your    │ ─────────────────────────────────▶ │  Convoy │
│  System  │   { params, callback_url }         │  API    │
└──────────┘                                    └────┬────┘
                                                     │
                                              (batch processing)
                                                     │
┌──────────┐         POST to callback_url      ┌────▼────┐
│  Your    │ ◀───────────────────────────────── │  Convoy │
│  Webhook │   { cargo_id, success, response }  │  Worker │
└──────────┘                                    └─────────┘

Convoy batches requests for cost efficiency. Results typically arrive within minutes to hours, depending on queue volume and batch size (up to 24-hour SLA). Design your webhook receiver to handle asynchronous delivery.

Callback Payload

When processing completes, Convoy POSTs this JSON payload to your callback_url:

Success Payload


{
  "cargo_id": "crg_abc123def456",
  "success": true,
  "response": {
    "id": "msg_01XFDUDYJgAACzvnptvVoYEL",
    "type": "message",
    "role": "assistant",
    "content": [
      {
        "type": "text",
        "text": "Here is the generated content..."
      }
    ],
    "model": "claude-sonnet-4-5",
    "stop_reason": "end_turn",
    "usage": {
      "input_tokens": 150,
      "output_tokens": 500
    }
  },
  "error": null
}

Error Payload


{
  "cargo_id": "crg_abc123def456",
  "success": false,
  "response": null,
  "error": "Model rate limit exceeded. Request will be retried."
}

Field Reference

Field	Type	Description
`cargo_id`	string	Unique identifier for the request (starts with `crg_`)
`success`	boolean	Whether processing completed successfully
`response`	object \| null	The AI model’s response (null on failure)
`response.content[].text`	string	The generated text content
`response.model`	string	The model that processed the request
`response.stop_reason`	string	Why the model stopped (`end_turn`, `max_tokens`, `stop_sequence`)
`response.usage.input_tokens`	integer	Tokens consumed by the prompt
`response.usage.output_tokens`	integer	Tokens generated in the response
`error`	string \| null	Error description (null on success)

Building a Webhook Receiver

Requirements

Your webhook endpoint must:

Accept POST requests with a JSON body
Return a 2xx status code within 30 seconds to acknowledge receipt
Be publicly accessible via HTTPS (for production use)

If your endpoint doesn’t respond with a 2xx within 30 seconds, Convoy will retry the delivery. Design your handler to be idempotent — use the cargo_id to deduplicate.

Python (Flask)


from flask import Flask, request, jsonify
 
app = Flask(__name__)
 
@app.route("/webhook/convoy", methods=["POST"])
def convoy_callback():
    payload = request.get_json()
 
    cargo_id = payload["cargo_id"]
    success = payload["success"]
 
    if success:
        text = payload["response"]["content"][0]["text"]
        model = payload["response"]["model"]
        tokens = payload["response"]["usage"]["output_tokens"]
        print(f"[{cargo_id}] Received {tokens} tokens from {model}")
        # Process the result — save to DB, send notification, etc.
    else:
        error = payload["error"]
        print(f"[{cargo_id}] Failed: {error}")
        # Handle the error — alert, retry logic, etc.
 
    # Always return 200 to acknowledge receipt
    return jsonify({"received": True}), 200
 
if __name__ == "__main__":
    app.run(port=5000)

Python (FastAPI)


from fastapi import FastAPI, Request
from pydantic import BaseModel
from typing import Optional
 
app = FastAPI()
 
class ConvoyCallback(BaseModel):
    cargo_id: str
    success: bool
    response: Optional[dict] = None
    error: Optional[str] = None
 
@app.post("/webhook/convoy")
async def convoy_callback(payload: ConvoyCallback):
    if payload.success:
        text = payload.response["content"][0]["text"]
        # Process the result
        print(f"[{payload.cargo_id}] Result: {text[:100]}...")
    else:
        # Handle the error
        print(f"[{payload.cargo_id}] Error: {payload.error}")
 
    return {"received": True}

Node.js (Express)


const express = require('express');
const app = express();
 
app.use(express.json());
 
app.post('/webhook/convoy', (req, res) => {
  const { cargo_id, success, response, error } = req.body;
 
  if (success) {
    const text = response.content[0].text;
    const model = response.model;
    console.log(`[${cargo_id}] Received from ${model}: ${text.slice(0, 100)}...`);
    // Process the result
  } else {
    console.error(`[${cargo_id}] Failed: ${error}`);
    // Handle the error
  }
 
  // Always respond 200 to acknowledge
  res.status(200).json({ received: true });
});
 
app.listen(3000, () => console.log('Webhook receiver running on :3000'));

Node.js (Next.js API Route)


// app/api/webhook/convoy/route.ts
import { NextRequest, NextResponse } from 'next/server';
 
export async function POST(request: NextRequest) {
  const payload = await request.json();
  const { cargo_id, success, response, error } = payload;
 
  if (success) {
    const text = response.content[0].text;
    // Save to database, trigger notification, etc.
    console.log(`[${cargo_id}] Result received`);
  } else {
    console.error(`[${cargo_id}] Error: ${error}`);
  }
 
  return NextResponse.json({ received: true }, { status: 200 });
}

Retry Behavior

Convoy retries failed callback deliveries with exponential backoff:

Attempt	Delay	Total Time Elapsed
1st retry	30 seconds	30s
2nd retry	2 minutes	2m 30s
3rd retry	10 minutes	12m 30s
4th retry	30 minutes	42m 30s
5th retry (final)	1 hour	1h 42m 30s

A delivery is considered failed if:

Your endpoint returns a non-2xx status code
The connection times out (30 second limit)
DNS resolution fails
The connection is refused

After all retries are exhausted, the callback status changes to callback_failed. You can still retrieve the result using the tracking endpoint: GET /cargo/{cargo_id}/tracking

Idempotency

Your webhook handler should be idempotent — processing the same callback twice should not cause duplicate side effects.

Use the cargo_id as a deduplication key:


# Example: Track processed callbacks in a set or database
processed_callbacks = set()
 
@app.route("/webhook/convoy", methods=["POST"])
def convoy_callback():
    payload = request.get_json()
    cargo_id = payload["cargo_id"]
 
    # Skip if already processed
    if cargo_id in processed_callbacks:
        return jsonify({"received": True, "duplicate": True}), 200
 
    processed_callbacks.add(cargo_id)
    # ... process the result ...
    return jsonify({"received": True}), 200

For production, use a database or Redis to track processed IDs rather than an in-memory set.

Local Development

During development, your webhook endpoint isn’t publicly accessible. Use a tunneling tool to expose it:

Using ngrok


# Start your local server
python app.py  # Running on localhost:5000
 
# In another terminal, expose it
ngrok http 5000

ngrok gives you a public URL like https://abc123.ngrok.io — use this as your callback_url:


curl -X POST https://api.cnvy.ai/cargo/load \
  -H "Content-Type: application/json" \
  -H "X-API-Key: convoy_sk_your_key_here" \
  -d '{
    "params": {
      "model": "claude-3-haiku",
      "max_tokens": 512,
      "messages": [{"role": "user", "content": "Hello, world!"}]
    },
    "callback_url": "https://abc123.ngrok.io/webhook/convoy"
  }'

Using the Tracking Endpoint (No Webhook Needed)

If you don’t want to set up a webhook receiver during development, you can poll for results instead:


# Submit without a callback_url
curl -X POST https://api.cnvy.ai/cargo/load \
  -H "Content-Type: application/json" \
  -H "X-API-Key: convoy_sk_your_key_here" \
  -d '{
    "params": {
      "model": "claude-3-haiku",
      "max_tokens": 512,
      "messages": [{"role": "user", "content": "Hello, world!"}]
    }
  }'
 
# Poll for the result
curl -H "X-API-Key: convoy_sk_your_key_here" \
  https://api.cnvy.ai/cargo/crg_abc123def456/tracking

Security Best Practices

1. Validate the Source

While Convoy doesn’t currently sign webhook payloads, you can add security by:

Using a unique, hard-to-guess URL path (e.g., /webhook/convoy/a8f3b2c1d4e5)
Verifying the cargo_id exists in your system before processing
Checking that the callback corresponds to a request you actually made

2. Use HTTPS

Always use HTTPS for your callback URL in production. Convoy will not deliver callbacks to plain HTTP endpoints in production environments.

3. Rate Limit Your Endpoint

Protect your webhook endpoint from abuse:


from flask_limiter import Limiter
 
limiter = Limiter(app, default_limits=["100 per minute"])
 
@app.route("/webhook/convoy", methods=["POST"])
@limiter.limit("100 per minute")
def convoy_callback():
    # ...

4. Process Asynchronously

For heavy processing, acknowledge the webhook immediately and process in the background:


from threading import Thread
 
@app.route("/webhook/convoy", methods=["POST"])
def convoy_callback():
    payload = request.get_json()
 
    # Acknowledge immediately
    Thread(target=process_result, args=(payload,)).start()
    return jsonify({"received": True}), 200
 
def process_result(payload):
    # Heavy processing happens here
    # Save to DB, generate reports, send emails, etc.
    pass

Troubleshooting

Callback never arrives

Check the request was accepted — The /cargo/load response should include a cargo_id
Verify your endpoint is reachable — Test with curl -X POST your-url
Check the tracking status — If it shows callback_failed, your endpoint may be unreachable
Review your server logs — Look for incoming POST requests
Ensure HTTPS — Production callbacks require HTTPS endpoints

Receiving duplicate callbacks

This is expected behavior during retries. Implement idempotency using the cargo_id as described above.

Payload is empty or malformed

Ensure your server parses the Content-Type: application/json body correctly
Check that your framework’s JSON body parser is enabled (e.g., express.json() in Express)

Timeout errors

Your endpoint must respond within 30 seconds
If processing takes longer, acknowledge immediately and process asynchronously
Consider using a message queue (SQS, Redis) for heavy workloads