| name | langfuse-webhooks-events |
| description | Configure Langfuse webhooks and event callbacks for real-time notifications.
Use when setting up trace notifications, configuring evaluation callbacks,
or integrating Langfuse events with external systems.
Trigger with phrases like "langfuse webhooks", "langfuse events",
"langfuse notifications", "langfuse callbacks", "langfuse alerts".
|
| allowed-tools | Read, Write, Edit |
| version | 1.0.0 |
| license | MIT |
| author | Jeremy Longshore <jeremy@intentsolutions.io> |
Langfuse Webhooks & Events
Overview
Configure webhooks and event callbacks to receive real-time notifications from Langfuse.
Prerequisites
- Langfuse account with webhook access
- HTTPS endpoint to receive webhooks
- Understanding of event-driven architecture
Instructions
Step 1: Create Webhook Endpoint
import { NextRequest, NextResponse } from "next/server";
import crypto from "crypto";
const WEBHOOK_SECRET = process.env.LANGFUSE_WEBHOOK_SECRET!;
interface LangfuseWebhookPayload {
event: string;
timestamp: string;
data: {
traceId?: string;
observationId?: string;
scoreId?: string;
projectId: string;
[key: string]: any;
};
}
function verifySignature(payload: string, signature: string): boolean {
const expectedSignature = crypto
.createHmac("sha256", WEBHOOK_SECRET)
.update(payload)
.digest("hex");
return crypto.timingSafeEqual(
Buffer.from(signature),
Buffer.from(expectedSignature)
);
}
export async function POST(request: NextRequest) {
const payload = await request.text();
const signature = request.headers.get("x-langfuse-signature");
if (!signature || !verifySignature(payload, signature)) {
console.error("Invalid webhook signature");
return NextResponse.json({ error: "Invalid signature" }, { status: 401 });
}
const event: LangfuseWebhookPayload = JSON.parse(payload);
console.log(`Received Langfuse event: ${event.event}`);
switch (event.event) {
case "trace.created":
await handleTraceCreated(event.data);
break;
case "trace.updated":
await handleTraceUpdated(event.data);
break;
case "score.created":
await handleScoreCreated(event.data);
break;
case "generation.created":
await handleGenerationCreated(event.data);
break;
default:
console.log(`Unhandled event type: ${event.event}`);
}
return NextResponse.json({ received: true });
}
async function handleTraceCreated(data: any) {
console.log(`New trace created: ${data.traceId}`);
}
async function handleTraceUpdated(data: any) {
if (data.level === "ERROR") {
await sendAlertNotification({
title: "Langfuse Error Trace",
traceId: data.traceId,
message: data.statusMessage,
});
}
}
async function handleScoreCreated(data: any) {
if (data.value < 0.5) {
await sendAlertNotification({
title: "Low Score Alert",
traceId: data.traceId,
score: data.name,
value: data.value,
});
}
}
async function handleGenerationCreated(data: any) {
if (data.usage) {
await trackTokenUsage({
model: data.model,
promptTokens: data.usage.promptTokens,
completionTokens: data.usage.completionTokens,
});
}
}
Step 2: Configure Webhook in Langfuse Dashboard
1. Go to Langfuse Dashboard -> Settings -> Webhooks
2. Click "Add Webhook"
3. Configure:
- URL: https://your-domain.com/api/webhooks/langfuse
- Events: Select events to subscribe to
- Secret: Generate and save webhook secret
4. Test webhook with "Send Test Event"
Step 3: Implement Event Processing Queue
import { Queue, Worker } from "bullmq";
import Redis from "ioredis";
const connection = new Redis(process.env.REDIS_URL!);
export const langfuseQueue = new Queue("langfuse-events", { connection });
export async function queueWebhookEvent(event: LangfuseWebhookPayload) {
await langfuseQueue.add(event.event, event, {
removeOnComplete: 1000,
removeOnFail: 5000,
attempts: 3,
backoff: {
type: "exponential",
delay: 1000,
},
});
}
const worker = new Worker(
"langfuse-events",
async (job) => {
const event = job.data ;
(job.) {
:
(event);
;
:
(event);
;
}
},
{ connection }
);
worker.(, {
.(, error);
});
Step 4: Real-Time Event Streaming (Alternative)
import { Langfuse } from "langfuse";
class LangfuseEventStream {
private langfuse: Langfuse;
private lastChecked: Date;
private pollInterval: number;
constructor(pollIntervalMs: number = 5000) {
this.langfuse = new Langfuse();
this.lastChecked = new Date();
this.pollInterval = pollIntervalMs;
}
async start(handlers: {
onTrace?: (trace: any) => void;
onScore?: (score: any) => void;
}) {
setInterval(async () => {
await this.pollForUpdates(handlers);
}, this.pollInterval);
}
private async pollForUpdates(: {
onTrace?: (trace: ) => ;
onScore?: (score: ) => ;
}) {
{
traces = ..({
: .,
});
( trace traces.) {
handlers.?.(trace);
}
. = ();
} (error) {
.(, error);
}
}
}
stream = ();
stream.({
: {
(trace. === ) {
();
}
},
});
Step 5: Integration with External Services
async function sendSlackNotification(event: LangfuseWebhookPayload) {
if (event.data.level !== "ERROR") return;
await fetch(process.env.SLACK_WEBHOOK_URL!, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
blocks: [
{
type: "header",
text: {
type: "plain_text",
text: "Langfuse Error Alert",
},
},
{
type: "section",
fields: [
{
type: "mrkdwn",
text: `*Trace ID:*\n${event.data.traceId}`,
},
{
type: "mrkdwn",
text: `*Error:*\n${event.data.statusMessage}`,
},
],
},
{
type: "actions",
elements: [
{
type: ,
: { : , : },
: ,
},
],
},
],
}),
});
}
() {
(, {
: ,
: { : },
: .({
: process..,
: ,
: {
: ,
: ,
: ,
: event.,
},
}),
});
}
Output
- Secure webhook endpoint with signature verification
- Event processing queue for reliability
- Real-time polling alternative
- External service integrations (Slack, PagerDuty)
Webhook Event Types
| Event | Description | Use Case |
|---|
trace.created | New trace started | Real-time monitoring |
trace.updated | Trace modified | Error detection |
generation.created | LLM call logged | Token tracking |
score.created | Score added | Quality alerts |
score.updated | Score modified | Evaluation updates |
Error Handling
| Issue | Cause | Solution |
|---|
| Invalid signature | Wrong secret | Verify webhook secret |
| Missed events | Handler failure | Use queue with retries |
| Duplicate events | No idempotency | Track processed event IDs |
| Timeout | Slow handler | Queue events, respond fast |
Examples
Idempotent Event Processing
const processedEvents = new Set<string>();
async function handleWebhook(event: LangfuseWebhookPayload) {
const eventId = `${event.event}-${event.timestamp}-${event.data.traceId}`;
if (processedEvents.has(eventId)) {
console.log(`Skipping duplicate event: ${eventId}`);
return;
}
processedEvents.add(eventId);
if (processedEvents.size > 10000) {
processedEvents.clear();
}
}
Resources
Next Steps
For performance optimization, see langfuse-performance-tuning.