> ## Documentation Index
> Fetch the complete documentation index at: https://docs.meter.sh/llms.txt
> Use this file to discover all available pages before exploring further.

# Webhooks Guide

> Receive real-time notifications when content changes

# Webhooks Guide

Set up webhook endpoints to receive immediate notifications when Meter detects content changes.

## Overview

Webhooks allow Meter to push change notifications to your application in real-time, eliminating the need for polling.

**Use webhooks when:**

* Changes need immediate action
* Building event-driven systems
* Triggering downstream workflows

**Use pull-based instead when:**

* Batch processing changes
* Webhooks aren't feasible (firewall, no public endpoint)
* Prefer manual control over timing

## How it works

```mermaid theme={null}
sequenceDiagram
    participant M as Meter
    participant Y as Your Webhook
    participant D as Your Database

    M->>M: Run scheduled job
    M->>M: Detect changes
    M->>Y: POST /webhooks/meter
    Y->>D: Update data
    Y->>M: 200 OK
```

## Webhook types

Meter supports four webhook formats:

| Type             | Description                              | When to use                          |
| ---------------- | ---------------------------------------- | ------------------------------------ |
| `standard`       | Full JSON payload with all results       | Most integrations                    |
| `slack`          | Formatted Slack incoming webhook message | Slack channels via incoming webhooks |
| `slack_workflow` | Slack Workflow Builder trigger payload   | Slack Workflow Builder automations   |
| `discord`        | Formatted Discord webhook embed          | Discord channels                     |

The webhook type is **auto-detected from the URL**:

| URL pattern                          | Detected type    |
| ------------------------------------ | ---------------- |
| Contains `hooks.slack.com/services/` | `slack`          |
| Contains `hooks.slack.com/triggers/` | `slack_workflow` |
| Contains `discord.com/api/webhooks/` | `discord`        |
| Everything else                      | `standard`       |

You can also set it explicitly:

```python theme={null}
schedule = client.create_schedule(
    strategy_id=strategy_id,
    url="https://example.com/products",
    interval_seconds=3600,
    webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
    webhook_type="slack"  # auto-detected, but can be explicit
)
```

## Webhook payload

Meter sends a POST request for both successful and failed jobs.

**Success payload:**

```json theme={null}
{
  "job_id": "660e8400-e29b-41d4-a716-446655440000",
  "schedule_id": "880e8400-e29b-41d4-a716-446655440000",
  "status": "completed",
  "url": "https://example.com/products",
  "results": [
    {"title": "Product A", "price": "$19.99"},
    {"title": "Product B", "price": "$29.99"}
  ],
  "item_count": 2,
  "has_changes": true,
  "content_hash": "7f3d9a2b4c1e...",
  "completed_at": "2025-01-15T10:30:12Z",
  "delivery_reason": "first_run",
  "metadata": {"project": "my-project", "env": "prod"}
}
```

<Tip>
  `delivery_reason` distinguishes the initial backfill (`first_run`) from
  change-driven deliveries (`content_changed`). Use it to skip notifying
  end-users on the very first run for a new schedule.
</Tip>

**Failure payload:**

```json theme={null}
{
  "job_id": "660e8400-e29b-41d4-a716-446655440000",
  "schedule_id": "880e8400-e29b-41d4-a716-446655440000",
  "status": "failed",
  "url": "https://example.com/products",
  "error": "Page not accessible: 404 Not Found",
  "completed_at": "2025-01-15T10:30:12Z",
  "metadata": {"project": "my-project", "env": "prod"}
}
```

The `metadata` field contains your custom JSON from `webhook_metadata` — it's included in every payload so you can identify which project or environment the webhook belongs to.

See [Webhook payload formats](/api-reference/rest/webhooks#webhook-payload-formats) for full field documentation.

## Webhook metadata

Attach custom JSON data to every webhook payload. This is useful for routing, tagging, or identifying which schedule triggered the webhook:

```python theme={null}
schedule = client.create_schedule(
    strategy_id=strategy_id,
    url="https://example.com/products",
    interval_seconds=3600,
    webhook_url="https://your-app.com/webhooks/meter",
    webhook_metadata={
        "project": "price-monitor",
        "env": "production",
        "team": "data-eng"
    }
)
```

Your webhook handler can then use this metadata for routing:

```python theme={null}
@app.post("/webhooks/meter")
async def handle_webhook(request: Request):
    payload = await request.json()
    metadata = payload.get("metadata", {})

    if metadata.get("project") == "price-monitor":
        await update_price_database(payload)
    elif metadata.get("project") == "news-tracker":
        await update_news_feed(payload)
```

## Webhook secrets

Webhook secrets let you verify that incoming requests are from Meter, not a third party.

### How it works

1. When you create a schedule with a `webhook_url`, Meter auto-generates a secret with a `whsec_` prefix
2. Every webhook request includes the secret in the `X-Webhook-Secret` header
3. Your endpoint verifies the header matches your stored secret

### Storing the secret

The secret is returned **once** when the schedule is created. Store it securely:

```python theme={null}
schedule = client.create_schedule(
    strategy_id=strategy_id,
    url="https://example.com/products",
    interval_seconds=3600,
    webhook_url="https://your-app.com/webhooks/meter"
)

# Save this — it won't be shown again in other API responses
webhook_secret = schedule.get("webhook_secret")
print(f"Store this secret: {webhook_secret}")
```

You can also provide your own secret:

```python theme={null}
schedule = client.create_schedule(
    strategy_id=strategy_id,
    url="https://example.com/products",
    interval_seconds=3600,
    webhook_url="https://your-app.com/webhooks/meter",
    webhook_secret="whsec_my_custom_secret_here"
)
```

### Verifying requests

Check the `X-Webhook-Secret` header in your webhook handler:

<Tabs>
  <Tab title="FastAPI (Python)">
    ```python theme={null}
    from fastapi import FastAPI, Request, HTTPException

    app = FastAPI()
    WEBHOOK_SECRET = "whsec_..."  # From schedule creation

    @app.post("/webhooks/meter")
    async def handle_webhook(request: Request):
        # Verify the secret
        secret = request.headers.get("X-Webhook-Secret")
        if secret != WEBHOOK_SECRET:
            raise HTTPException(status_code=401, detail="Invalid webhook secret")

        payload = await request.json()
        # Process payload...
        return {"status": "ok"}
    ```
  </Tab>

  <Tab title="Express (Node.js)">
    ```javascript theme={null}
    const express = require('express');
    const app = express();
    app.use(express.json());

    const WEBHOOK_SECRET = 'whsec_...'; // From schedule creation

    app.post('/webhooks/meter', (req, res) => {
      // Verify the secret
      const secret = req.headers['x-webhook-secret'];
      if (secret !== WEBHOOK_SECRET) {
        return res.status(401).json({ error: 'Invalid webhook secret' });
      }

      const payload = req.body;
      // Process payload...
      res.status(200).json({ status: 'ok' });
    });
    ```
  </Tab>

  <Tab title="Flask (Python)">
    ```python theme={null}
    from flask import Flask, request, jsonify

    app = Flask(__name__)
    WEBHOOK_SECRET = "whsec_..."  # From schedule creation

    @app.route('/webhooks/meter', methods=['POST'])
    def handle_webhook():
        # Verify the secret
        secret = request.headers.get('X-Webhook-Secret')
        if secret != WEBHOOK_SECRET:
            return jsonify({"error": "Invalid secret"}), 401

        payload = request.json
        # Process payload...
        return jsonify({"status": "ok"}), 200
    ```
  </Tab>
</Tabs>

### Regenerating secrets

If a secret is compromised, regenerate it:

```python theme={null}
result = client.regenerate_webhook_secret(schedule_id)
new_secret = result["webhook_secret"]
# Update your webhook handler with the new secret
```

<Warning>
  The old secret is immediately invalidated. Update your webhook handler before the next delivery.
</Warning>

## Retry behavior

Meter automatically retries failed webhook deliveries with exponential backoff:

| Retry       | Delay      | Total elapsed      |
| ----------- | ---------- | ------------------ |
| 1st         | 15 minutes | 15 minutes         |
| 2nd         | 30 minutes | 45 minutes         |
| 3rd         | 1 hour     | 1 hour 45 minutes  |
| 4th         | 2 hours    | 3 hours 45 minutes |
| 5th (final) | 4 hours    | 7 hours 45 minutes |

**Retry rules:**

* **2xx response**: Delivery successful, no retry
* **4xx response** (client error): Delivery stops immediately — no retries. Fix your endpoint and the next scheduled job will deliver normally
* **5xx response** (server error): Retries with backoff
* **Timeout** (>30 seconds): Retries with backoff
* **Connection failure**: Retries with backoff

<Tip>
  Return `200 OK` as quickly as possible. Process the payload asynchronously in a background task to avoid timeouts.
</Tip>

## Implementation

### Step 1: Create a webhook endpoint

<Tabs>
  <Tab title="FastAPI (Python)">
    ```python theme={null}
    from fastapi import FastAPI, Request, HTTPException, BackgroundTasks

    app = FastAPI()
    WEBHOOK_SECRET = "whsec_..."

    @app.post("/webhooks/meter")
    async def handle_meter_webhook(
        request: Request,
        background_tasks: BackgroundTasks
    ):
        # Verify secret
        secret = request.headers.get("X-Webhook-Secret")
        if secret != WEBHOOK_SECRET:
            raise HTTPException(status_code=401, detail="Invalid secret")

        payload = await request.json()

        if payload["status"] == "failed":
            background_tasks.add_task(handle_failure, payload)
            return {"status": "ok"}

        if payload.get("has_changes"):
            background_tasks.add_task(process_changes, payload)

        return {"status": "ok"}

    async def process_changes(payload):
        results = payload["results"]
        metadata = payload.get("metadata", {})
        # Update your database, trigger pipelines, etc.

    async def handle_failure(payload):
        # Log, alert, or retry logic
        print(f"Job failed: {payload['error']}")
    ```
  </Tab>

  <Tab title="Express (Node.js)">
    ```javascript theme={null}
    const express = require('express');
    const app = express();
    app.use(express.json());

    const WEBHOOK_SECRET = 'whsec_...';

    app.post('/webhooks/meter', async (req, res) => {
      // Verify secret
      const secret = req.headers['x-webhook-secret'];
      if (secret !== WEBHOOK_SECRET) {
        return res.status(401).json({ error: 'Invalid secret' });
      }

      const payload = req.body;

      if (payload.status === 'failed') {
        console.error(`Job failed: ${payload.error}`);
        return res.status(200).json({ status: 'ok' });
      }

      if (payload.has_changes) {
        // Process asynchronously
        setImmediate(() => processChanges(payload));
      }

      res.status(200).json({ status: 'ok' });
    });

    async function processChanges(payload) {
      const results = payload.results;
      const metadata = payload.metadata || {};
      // Update database, etc.
    }

    app.listen(3000);
    ```
  </Tab>

  <Tab title="Flask (Python)">
    ```python theme={null}
    from flask import Flask, request, jsonify
    import threading

    app = Flask(__name__)
    WEBHOOK_SECRET = "whsec_..."

    @app.route('/webhooks/meter', methods=['POST'])
    def handle_meter_webhook():
        # Verify secret
        secret = request.headers.get('X-Webhook-Secret')
        if secret != WEBHOOK_SECRET:
            return jsonify({"error": "Invalid secret"}), 401

        payload = request.json

        if payload["status"] == "failed":
            print(f"Job failed: {payload['error']}")
            return jsonify({"status": "ok"}), 200

        if payload.get("has_changes"):
            # Process in background thread
            threading.Thread(
                target=process_changes, args=(payload,)
            ).start()

        return jsonify({"status": "ok"}), 200

    def process_changes(payload):
        results = payload["results"]
        metadata = payload.get("metadata", {})
        # Update database, etc.
    ```
  </Tab>
</Tabs>

### Step 2: Make endpoint publicly accessible

Options:

* Deploy to cloud (AWS Lambda, Google Cloud Functions, etc.)
* Use ngrok for local development: `ngrok http 3000`
* Use a VPS with public IP

### Step 3: Create schedule with webhook URL

```python theme={null}
from meter_sdk import MeterClient

client = MeterClient(api_key="sk_live_...")

schedule = client.create_schedule(
    strategy_id="your-strategy-id",
    url="https://example.com/products",
    interval_seconds=3600,
    webhook_url="https://your-app.com/webhooks/meter",
    webhook_metadata={"project": "price-monitor"}
)

# Store the auto-generated webhook secret
print(f"Webhook secret: {schedule.get('webhook_secret')}")
```

## Slack webhooks

Send notifications directly to Slack channels using incoming webhooks:

```python theme={null}
schedule = client.create_schedule(
    strategy_id=strategy_id,
    url="https://example.com/products",
    interval_seconds=3600,
    webhook_url="https://hooks.slack.com/services/T.../B.../xxx"
    # webhook_type auto-detected as "slack"
)
```

Slack payloads are automatically formatted as readable messages with item counts and a preview of results.

## Slack Workflow webhooks

Trigger Slack Workflow Builder automations with scrape results. Use this when you've built a workflow in Slack's Workflow Builder and want Meter to trigger it.

```python theme={null}
schedule = client.create_schedule(
    strategy_id=strategy_id,
    url="https://example.com/products",
    interval_seconds=3600,
    webhook_url="https://hooks.slack.com/triggers/T.../B.../xxx"
    # webhook_type auto-detected as "slack_workflow"
)
```

<Note>
  Slack Workflow webhooks do not require a webhook secret. The URL itself serves as the authentication mechanism.
</Note>

## Discord webhooks

Send formatted notifications to Discord channels:

```python theme={null}
schedule = client.create_schedule(
    strategy_id=strategy_id,
    url="https://example.com/products",
    interval_seconds=3600,
    webhook_url="https://discord.com/api/webhooks/123/abc..."
    # webhook_type auto-detected as "discord"
)
```

Discord payloads are formatted as embeds with item counts and a preview of results.

<Note>
  Discord webhooks do not require a webhook secret. Discord authenticates via the webhook URL.
</Note>

## Best practices

### 1. Respond quickly

Return `200 OK` within 30 seconds. Process payloads asynchronously:

```python theme={null}
@app.post("/webhooks/meter")
async def handle_webhook(request: Request, background_tasks: BackgroundTasks):
    payload = await request.json()
    background_tasks.add_task(process_changes, payload)
    return {"status": "ok"}  # Respond immediately
```

### 2. Handle duplicates

Make processing idempotent using the `job_id`:

```python theme={null}
processed_jobs = set()

async def process_changes(payload):
    job_id = payload['job_id']
    if job_id in processed_jobs:
        return  # Skip duplicate
    processed_jobs.add(job_id)
    # Process...
```

### 3. Handle failures gracefully

```python theme={null}
@app.post("/webhooks/meter")
async def handle_webhook(request: Request):
    try:
        payload = await request.json()
        await process_changes(payload)
        return {"status": "ok"}
    except Exception as e:
        logger.error(f"Webhook processing failed: {e}")
        # Return 200 to prevent retries (payload is stored for manual review)
        await save_failed_webhook(payload, str(e))
        return {"status": "error", "message": str(e)}
```

## Testing webhooks

### Local testing with ngrok

```bash theme={null}
# Start your webhook server
python app.py

# In another terminal, expose with ngrok
ngrok http 3000

# Use the ngrok URL in Meter
# https://abc123.ngrok.io/webhooks/meter
```

### Testing with webhook.site

1. Go to [https://webhook.site](https://webhook.site)
2. Copy the unique URL
3. Use it in your Meter schedule
4. Trigger a job and view the payload on webhook.site

### Test endpoint

Use the Meter test endpoint to verify delivery:

```bash theme={null}
curl -X POST https://api.meter.sh/api/webhooks/test \
  -H "Authorization: Bearer sk_live_..." \
  -H "Content-Type: application/json" \
  -d '{"webhook_url": "https://your-app.com/webhooks/meter"}'
```

### Manual testing

```bash theme={null}
curl -X POST http://localhost:3000/webhooks/meter \
  -H "Content-Type: application/json" \
  -H "X-Webhook-Secret: whsec_test" \
  -d '{
    "job_id": "test-123",
    "status": "completed",
    "url": "https://example.com",
    "has_changes": true,
    "results": [{"test": "data"}],
    "item_count": 1,
    "metadata": {"project": "test"}
  }'
```

## Troubleshooting

<AccordionGroup>
  <Accordion title="Not receiving webhooks">
    **Solutions:**

    * Verify URL is publicly accessible
    * Check endpoint returns 200 OK
    * Test with webhook.site
    * Check server logs for errors
    * Verify the schedule has a `webhook_url` set
  </Accordion>

  <Accordion title="Timeout errors">
    **Cause:** Endpoint takes >30 seconds to respond

    **Solution:** Return 200 OK immediately, process asynchronously:

    ```python theme={null}
    @app.post("/webhooks/meter")
    async def handle_webhook(request: Request, background_tasks: BackgroundTasks):
        payload = await request.json()
        background_tasks.add_task(process_changes, payload)
        return {"status": "ok"}  # Immediate response
    ```
  </Accordion>

  <Accordion title="Duplicate webhooks">
    **Cause:** Retries after timeout or connection issues

    **Solution:** Make processing idempotent using `job_id`:

    ```python theme={null}
    processed_jobs = set()

    async def process_changes(payload):
        job_id = payload['job_id']
        if job_id in processed_jobs:
            return
        processed_jobs.add(job_id)
        # Process...
    ```
  </Accordion>

  <Accordion title="4xx errors stopping delivery">
    **Cause:** Your endpoint returns a 4xx status code

    **Solution:** Meter treats 4xx as a permanent failure and does not retry. Check your endpoint for:

    * Invalid webhook secret (401)
    * Incorrect URL path (404)
    * Request validation errors (422)

    Fix the issue and delivery will resume on the next scheduled job.
  </Accordion>
</AccordionGroup>

## Next steps

* Try [Pull-Based Monitoring](/guides/pull-based-monitoring) as an alternative
* See [RAG Integration](/guides/rag-integration) for vector database updates
* Check [Schedule API Reference](/api-reference/python/schedules) for webhook configuration
* See [Webhook payload formats](/api-reference/rest/webhooks) for full field documentation

## Need help?

Email me at [mckinnon@meter.sh](mailto:mckinnon@meter.sh)