Webhooks Local Testing

When developing webhook integrations, you need to test how your application handles CoreValue events before deploying to production. Webhooks local testing uses tunneling tools like ngrok to expose your local development server to the internet, allowing CoreValue to send real webhook events to your local machine for debugging and integration testing.

Why use Webhooks Local Testing

Debug integration issues: Test webhook handlers with real events before deploying to production
Iterate quickly: Make changes to your webhook handler and test immediately without deployment cycles
Validate event handling: Ensure your application correctly processes different webhook event types and payloads

Quick Start

Set up a local server to receive webhook events:

from fastapi import FastAPI, Request
import json

app = FastAPI()

@app.post("/webhook")
async def webhook_handler(request: Request):
    # Parse the webhook payload
    payload = await request.json()
    
    # Log the event for debugging
    print(f"Received event: {payload['event_type']}")
    print(f"Request ID: {payload['request_id']}")
    
    # Your webhook logic here
    # e.g., store in database, trigger notifications, etc.
    
    return {"status": "success"}

# Run with: uvicorn main:app --reload --port 8000

Install and configure ngrok to expose your local server:

# Install ngrok (macOS)
brew install ngrok

# Or download from https://ngrok.com/download

# Start tunnel to your local server
ngrok http 8000

You'll see output like:

Forwarding  https://abc123.ngrok-free.app -> http://localhost:8000

Copy the HTTPS URL for the next step.

Add your ngrok URL to CoreValue's webhook settings:

Go to your CoreValue dashboard → Settings → Webhooks
Click "Add Webhook"
Enter your ngrok URL with the webhook path: https://abc123.ngrok-free.app/webhook
Select the events you want to receive
Save the webhook configuration

Use Cases

Test webhook integration during local development:

from fastapi import FastAPI, Request, HTTPException
import json
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

@app.post("/webhook/corevalue")
async def corevalue_webhook(request: Request):
    try:
        payload = await request.json()
        
        # Log full payload for debugging
        logger.info(f"Webhook payload: {json.dumps(payload, indent=2)}")
        
        # Process the webhook payload
        request_id = payload["request_id"]
        model = payload.get("model")
        cost = payload.get("cost", 0)
        user_id = payload.get("user_id")
        
        logger.info(f"Webhook for request {request_id}")
        logger.info(f"Model: {model}, Cost: {cost}")
        
        if user_id:
            logger.info(f"User: {user_id}")
            
        # Check if this is a high-cost request
        if cost > 1.0:
            logger.warning(f"High cost request detected: {cost}")
        
        return {"status": "processed"}
        
    except Exception as e:
        logger.error(f"Webhook error: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))

# Test with property filter
# Add header: Cova-Property-Environment: development

const express = require('express');
const app = express();

app.use(express.json());

// Webhook endpoint
app.post('/webhook/corevalue', (req, res) => {
  const payload = req.body;
  
  console.log('Webhook received:', JSON.stringify(payload, null, 2));
  
  // Process the webhook payload
  const { request_id, model, cost = 0, user_id } = payload;
  
  console.log(`Webhook for request ${request_id}`);
  console.log(`Model: ${model}, Cost: $${cost}`);
  
  if (user_id) {
    console.log(`User: ${user_id}`);
  }
  
  // Check if this is a high-cost request
  if (cost > 1.0) {
    console.warn(`High cost request detected: $${cost}`);
  }
  
  res.json({ status: 'processed' });
});

app.listen(8000, () => {
  console.log('Webhook server running on http://localhost:8000');
});

Build a complete webhook processing system:

import asyncio
from fastapi import FastAPI, Request, BackgroundTasks
import aioredis
import json

app = FastAPI()
redis = None

@app.on_event("startup")
async def startup():
    global redis
    redis = await aioredis.create_redis_pool('redis://localhost')

@app.post("/webhook")
async def webhook_handler(
    request: Request, 
    background_tasks: BackgroundTasks
):
    # Quick acknowledgment
    payload = await request.json()
    
    # Queue for async processing
    await redis.lpush(
        "webhook_queue", 
        json.dumps(payload)
    )
    
    # Process async
    background_tasks.add_task(
        process_webhook_event, 
        payload
    )
    
    return {"status": "queued"}

async def process_webhook_event(payload):
    """Process webhook events asynchronously"""
    # Process completed request
    cost = payload.get("cost", 0)
    
    # Check for anomalies
    if cost > 1.0:  # High cost threshold
        await send_alert(
            f"High cost request: {cost}",
            payload
        )
    
    # Update metrics
    await update_usage_metrics(payload)
    
    # Check rate limits
    user_id = payload.get("user_id")
    if user_id:
        await check_user_limits(user_id, cost)

async def send_alert(message, data):
    # Send to Slack, email, etc.
    pass

async def update_usage_metrics(payload):
    # Update dashboard, analytics, etc.
    pass

async def check_user_limits(user_id, cost):
    # Implement usage limiting logic
    pass