Handling webhooks

Learn how to receive and process livestream webhooks from Pump.fun

Pump.fun provides webhook endpoints for livestream events. This guide shows you how to handle webhook callbacks for LiveKit and other streaming services.

LiveKit webhooks

LiveKit webhooks notify your application about livestream events such as room creation, participant joining/leaving, and recording status.

Webhook endpoint

The LiveKit webhook endpoint is:

POST https://frontend-api-v3.pump.fun/livestreams/livekit-webhook

Webhook authentication

Webhooks require proper authentication via the Authorization header:

cURL

curl -X POST "https://frontend-api-v3.pump.fun/livestreams/livekit-webhook" \
  -H "Authorization: Bearer <your_token>" \
  -H "Content-Type: application/json" \
  -d '{"event": "room_started", "room": "room123"}'

Webhooks require valid JWT authentication. Ensure your webhook handler includes proper authorization headers when forwarding events.

Call webhooks

The general call webhook endpoint handles various streaming events:

POST https://frontend-api-v3.pump.fun/livestreams/call-webhook

Signature verification

This endpoint uses signature-based authentication via the x-signature header:

cURL

curl -X POST "https://frontend-api-v3.pump.fun/livestreams/call-webhook" \
  -H "Authorization: Bearer <your_token>" \
  -H "x-signature: <computed_signature>" \
  -H "Content-Type: application/json" \
  -d '{"event_type": "call.started", "call_id": "abc123"}'

Always verify webhook signatures to ensure requests are authentic and haven’t been tampered with.

Setting up a webhook receiver

Create a webhook receiver server to handle incoming webhook events:

Python

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

app = Flask(__name__)

WEBHOOK_SECRET = "your_webhook_secret"

def verify_signature(payload, signature):
    """
    Verify webhook signature
    """
    expected_signature = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected_signature)

@app.route('/webhooks/livekit', methods=['POST'])
def handle_livekit_webhook():
    """
    Handle LiveKit webhook events
    """
    try:
        # Get the signature from headers
        signature = request.headers.get('x-signature', '')
        payload = request.get_data(as_text=True)
        
        # Verify signature
        if not verify_signature(payload, signature):
            return jsonify({"error": "Invalid signature"}), 401
        
        # Parse webhook data
        data = request.json
        event_type = data.get('event')
        
        # Handle different event types
        if event_type == 'room_started':
            handle_room_started(data)
        elif event_type == 'participant_joined':
            handle_participant_joined(data)
        elif event_type == 'participant_left':
            handle_participant_left(data)
        elif event_type == 'recording_started':
            handle_recording_started(data)
        else:
            print(f"Unknown event type: {event_type}")
        
        return jsonify({"status": "success"}), 200
        
    except Exception as e:
        print(f"Error handling webhook: {e}")
        return jsonify({"error": str(e)}), 500

def handle_room_started(data):
    print(f"Room started: {data.get('room')}")
    # Implement your logic here

def handle_participant_joined(data):
    print(f"Participant joined: {data.get('participant')}")
    # Implement your logic here

def handle_participant_left(data):
    print(f"Participant left: {data.get('participant')}")
    # Implement your logic here

def handle_recording_started(data):
    print(f"Recording started for room: {data.get('room')}")
    # Implement your logic here

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

Webhook event types

Common webhook event types you might receive:

LiveKit events

• room_started - A new room has been created
• room_finished - A room has ended
• participant_joined - A participant joined the room
• participant_left - A participant left the room
• recording_started - Recording started for a room
• recording_finished - Recording finished for a room
• track_published - A track was published
• track_unpublished - A track was unpublished

Call events

• call.started - A call has started
• call.ended - A call has ended
• call.participant_joined - A participant joined
• call.participant_left - A participant left
• call.recording_ready - Recording is ready for download

Forwarding webhooks

If you need to forward webhook events to Pump.fun from your streaming service:

Python

import requests
import hmac
import hashlib

class WebhookForwarder:
    def __init__(self, pump_fun_token, webhook_secret):
        self.token = pump_fun_token
        self.secret = webhook_secret
        self.base_url = "https://frontend-api-v3.pump.fun"
    
    def forward_livekit_event(self, event_data):
        """
        Forward LiveKit event to Pump.fun
        """
        url = f"{self.base_url}/livestreams/livekit-webhook"
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        }
        
        response = requests.post(url, headers=headers, json=event_data)
        return response.json()
    
    def forward_call_event(self, event_data):
        """
        Forward call event to Pump.fun with signature
        """
        url = f"{self.base_url}/livestreams/call-webhook"
        
        # Convert data to string for signature
        import json
        payload = json.dumps(event_data)
        
        # Compute signature
        signature = hmac.new(
            self.secret.encode(),
            payload.encode(),
            hashlib.sha256
        ).hexdigest()
        
        headers = {
            "Authorization": f"Bearer {self.token}",
            "x-signature": signature,
            "Content-Type": "application/json"
        }
        
        response = requests.post(url, headers=headers, data=payload)
        return response.json()

# Usage
forwarder = WebhookForwarder(
    pump_fun_token="your_token",
    webhook_secret="your_secret"
)

# Forward a LiveKit event
livekit_event = {
    "event": "room_started",
    "room": "room123",
    "created_at": 1234567890
}
forwarder.forward_livekit_event(livekit_event)

# Forward a call event
call_event = {
    "event_type": "call.started",
    "call_id": "abc123",
    "timestamp": 1234567890
}
forwarder.forward_call_event(call_event)

Best practices

1. Verify signatures

   Always verify webhook signatures using HMAC to ensure authenticity and prevent tampering.

2. Return quickly

   Process webhooks asynchronously. Respond with 200 status immediately and handle processing in background jobs.

3. Implement idempotency

   Use event IDs to track processed webhooks and prevent duplicate processing.

4. Handle retries

   Implement proper retry logic with exponential backoff for failed webhook deliveries.

5. Log events

   Maintain detailed logs of webhook events for debugging and auditing purposes.

6. Secure your endpoint

   Use HTTPS and validate all incoming requests to prevent unauthorized access.

Webhook endpoints should respond within 5 seconds. For long-running operations, use a message queue or background job processor.

Testing webhooks locally

Use tools like ngrok to expose your local webhook receiver:

# Install ngrok
brew install ngrok  # macOS
# or download from https://ngrok.com

# Expose local port
ngrok http 8080

# Use the generated URL as your webhook endpoint
# Example: https://abc123.ngrok.io/webhooks/livekit

This allows you to test webhook integration during development without deploying to a public server.