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 -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 -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:
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:
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.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:
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
Verify signatures
Always verify webhook signatures using HMAC to ensure authenticity and prevent tampering.
Return quickly
Process webhooks asynchronously. Respond with 200 status immediately and handle processing in background jobs.
Implement idempotency
Use event IDs to track processed webhooks and prevent duplicate processing.
Handle retries
Implement proper retry logic with exponential backoff for failed webhook deliveries.
Log events
Maintain detailed logs of webhook events for debugging and auditing purposes.
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.