AliveMCP

Guide · LLM Observability & Evaluation

MCP Server Langfuse — trace IDs, generation spans, flush timing, prompt versioning

Three Langfuse behaviours catch MCP server authors off-guard: traces are only correlated if you thread the traceId from the agent session into every tool call — without this, each LLM call inside a tool creates an orphaned trace, making it impossible to see which agent session produced which generation; langfuse.flush() must be awaited before the MCP tool handler returns — Langfuse batches events and POSTs them asynchronously, and in short-lived handlers the process exits before the HTTP upload completes, silently dropping all trace data; and langfuse.getPrompt() is asynchronous and adds 50–200 ms on cache miss — always supply a fallback prompt string so a Langfuse API outage doesn't propagate as a tool error to the calling agent.

TL;DR

Create one Langfuse instance at module level. In each tool handler, call langfuse.trace({ id: sessionId, name: 'tool.call' }) using the agent session ID as id, then trace.generation({ ... }) around the LLM call, and await langfuse.flushAsync() before returning. Fetch prompts with await langfuse.getPrompt('my-prompt', { fallback: FALLBACK_PROMPT, cacheTtlSeconds: 300 }). Never call new Langfuse() inside a tool handler — that creates a new SDK client with no batching state on every call, multiplying API overhead by the number of concurrent requests.

Client setup and trace correlation

The Langfuse Node.js SDK reads LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, and LANGFUSE_BASEURL from environment variables. Create the client once at module scope so the internal batch queue and HTTP connection pool are shared across all tool handler invocations.

import { Langfuse } from 'langfuse';
import OpenAI from 'openai';
import { z } from 'zod';
import { randomUUID } from 'crypto';

// Module-level singletons — one shared batch queue + HTTP pool
const langfuse = new Langfuse({
  // Env vars read automatically: LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_BASEURL
  // Explicit overrides only needed in test/dev:
  // publicKey: process.env.LANGFUSE_PUBLIC_KEY,
  // secretKey: process.env.LANGFUSE_SECRET_KEY,
  // baseUrl:   process.env.LANGFUSE_BASEURL ?? 'https://cloud.langfuse.com',
  flushAt: 10,   // flush when 10 events accumulate
  flushInterval: 5_000,  // or every 5 seconds
});

const openai = new OpenAI();

// MCP tool — summarise text via LLM, traced in Langfuse
server.tool(
  'summarize',
  {
    text:        z.string().min(1).max(20_000),
    style:       z.enum(['bullet', 'paragraph']).optional().default('bullet'),
    // Agent passes its session ID here so we can correlate all calls in one trace
    session_id:  z.string().uuid().optional(),
  },
  async ({ text, style, session_id }) => {
    const traceId = session_id ?? randomUUID();

    // trace() is idempotent on traceId — calling it again with the same ID
    // updates the existing trace rather than creating a new orphaned one
    const trace = langfuse.trace({
      id:       traceId,
      name:     'mcp.tool.summarize',
      userId:   traceId,  // use session as userId for cost attribution
      metadata: { style, textLength: text.length },
    });

    const span = trace.span({ name: 'prepare.prompt' });
    const prompt = `Summarize the following text in ${style} format:\n\n${text}`;
    span.end();

    // generation() wraps the actual LLM call with token and cost tracking
    const generation = trace.generation({
      name:    'openai.chat',
      model:   'gpt-4o-mini',
      input:   [{ role: 'user', content: prompt }],
      modelParameters: { max_tokens: 512 },
    });

    let summary: string;
    let usage: { promptTokens: number; completionTokens: number } | null = null;

    try {
      const completion = await openai.chat.completions.create({
        model:      'gpt-4o-mini',
        messages:   [{ role: 'user', content: prompt }],
        max_tokens: 512,
      });
      summary = completion.choices[0]?.message?.content ?? '';
      usage   = {
        promptTokens:     completion.usage?.prompt_tokens     ?? 0,
        completionTokens: completion.usage?.completion_tokens ?? 0,
      };
    } catch (err: any) {
      generation.end({ level: 'ERROR', statusMessage: err.message });
      await langfuse.flushAsync();
      throw err;
    }

    generation.end({
      output: summary,
      usage:  {
        input:  usage.promptTokens,
        output: usage.completionTokens,
        total:  usage.promptTokens + usage.completionTokens,
      },
    });

    // CRITICAL: flush before returning — handler exits after return
    await langfuse.flushAsync();

    return { content: [{ type: 'text', text: summary }] };
  }
);

Pass the agent's session ID as both the traceId and the optional userId. When MCP clients (Claude Desktop, Cursor, Windsurf) send _meta.sessionId in the tool call context, read it from the extra meta field and thread it through. All LLM calls within the same agent conversation then appear as one Langfuse trace with a timeline of spans instead of dozens of disconnected generations.

Generation spans and flush timing

Langfuse generation spans capture the model, input messages, output text, token counts, and cost automatically when you use Langfuse-native model pricing configs. The key invariant for MCP tool handlers is that generation.end() must be called before flushAsync(), and flushAsync() must be awaited before the handler returns. Skipping the await is the single most common source of missing Langfuse data in MCP deployments.

import { Langfuse, type LangfuseGenerationClient } from 'langfuse';
import Anthropic from '@anthropic-ai/sdk';

const langfuse = new Langfuse();
const anthropic = new Anthropic();

// Multi-step tool: search → summarise → classify
// Each step gets its own span so the Langfuse timeline is readable
server.tool(
  'research_topic',
  { query: z.string().min(1).max(500), trace_id: z.string().optional() },
  async ({ query, trace_id }) => {
    const traceId = trace_id ?? randomUUID();
    const trace   = langfuse.trace({ id: traceId, name: 'research_topic' });

    // Step 1: search (placeholder — replace with real search)
    const searchSpan = trace.span({ name: 'web.search', input: { query } });
    const searchResults = `Search results for: ${query} (stub)`;
    searchSpan.end({ output: { resultCount: 3 } });

    // Step 2: summarise results with LLM
    const summaryGen = trace.generation({
      name:   'summarise_results',
      model:  'claude-haiku-4-5-20251001',
      input:  [{ role: 'user', content: `Summarise: ${searchResults}` }],
    });

    const summaryMsg = await anthropic.messages.create({
      model:      'claude-haiku-4-5-20251001',
      max_tokens: 256,
      messages:   [{ role: 'user', content: `Summarise: ${searchResults}` }],
    });
    const summary = summaryMsg.content[0]?.type === 'text' ? summaryMsg.content[0].text : '';

    summaryGen.end({
      output: summary,
      usage:  {
        input:  summaryMsg.usage.input_tokens,
        output: summaryMsg.usage.output_tokens,
        total:  summaryMsg.usage.input_tokens + summaryMsg.usage.output_tokens,
      },
    });

    // Step 3: classify
    const classifyGen = trace.generation({
      name:   'classify_topic',
      model:  'claude-haiku-4-5-20251001',
      input:  [{ role: 'user', content: `Classify into one category: ${summary}` }],
    });

    const classifyMsg = await anthropic.messages.create({
      model:      'claude-haiku-4-5-20251001',
      max_tokens: 32,
      messages:   [{ role: 'user', content: `Classify into one category: ${summary}` }],
    });
    const category = classifyMsg.content[0]?.type === 'text' ? classifyMsg.content[0].text.trim() : 'unknown';

    classifyGen.end({
      output: category,
      usage: {
        input:  classifyMsg.usage.input_tokens,
        output: classifyMsg.usage.output_tokens,
        total:  classifyMsg.usage.input_tokens + classifyMsg.usage.output_tokens,
      },
    });

    trace.update({ output: { summary, category } });

    // Always await — never fire-and-forget in a request handler
    await langfuse.flushAsync();

    return {
      content: [{ type: 'text', text: JSON.stringify({ summary, category }) }],
    };
  }
);

If the MCP server runs in a long-lived process (not serverless), the automatic flush interval (flushInterval: 5000) will eventually drain the queue without explicit flushAsync() calls — but per-request flushing makes traces appear in the Langfuse UI within seconds of the tool call completing, which is important for debugging live agent sessions.

Prompt versioning with fallback and cache TTL

Langfuse's prompt management API lets you version prompts in the UI and pull the active version at runtime. langfuse.getPrompt() caches the result locally for cacheTtlSeconds seconds. The critical pattern for MCP tools is to always provide a fallback compiled prompt string — if Langfuse's API is unreachable, your tool should still work using the last known-good prompt baked into the code.

import { Langfuse } from 'langfuse';
import { z } from 'zod';

const langfuse = new Langfuse();

// Compile-time fallback — matches the Langfuse prompt template
const SUMMARISE_PROMPT_FALLBACK = `Summarise the following content in {{style}} format.
Focus on key insights and be concise.

Content:
{{content}}`;

server.tool(
  'summarize_with_versioned_prompt',
  { content: z.string(), style: z.enum(['bullets', 'paragraph']).default('bullets') },
  async ({ content, style }) => {
    // getPrompt: checks cache first, fetches from API on miss
    // cacheTtlSeconds: how long the local cache entry is valid
    // fallback: used when API is unreachable (avoids tool failure on Langfuse outage)
    const promptObj = await langfuse.getPrompt('mcp-summarise', {
      cacheTtlSeconds: 300,
      fallback: SUMMARISE_PROMPT_FALLBACK,
    });

    // compile() replaces {{variables}} with actual values
    const promptText = promptObj.compile({ style, content });

    const trace      = langfuse.trace({ name: 'summarize_with_versioned_prompt' });
    const generation = trace.generation({
      name:         'gpt4o-mini.completion',
      model:        'gpt-4o-mini',
      prompt:       promptObj,  // links generation to this prompt version in Langfuse UI
      input:        [{ role: 'user', content: promptText }],
    });

    const completion = await openai.chat.completions.create({
      model:    'gpt-4o-mini',
      messages: [{ role: 'user', content: promptText }],
    });
    const result = completion.choices[0]?.message?.content ?? '';

    generation.end({
      output: result,
      usage: {
        input:  completion.usage?.prompt_tokens     ?? 0,
        output: completion.usage?.completion_tokens ?? 0,
        total:  completion.usage?.total_tokens      ?? 0,
      },
    });

    await langfuse.flushAsync();
    return { content: [{ type: 'text', text: result }] };
  }
);

Linking the promptObj to the generation via prompt: promptObj in trace.generation() tells Langfuse which prompt version produced each output. This powers the Langfuse prompt analytics view — click a prompt version to see all generations that used it, their scores, and token cost.

Scoring and user feedback from MCP tool results

Langfuse scores let you attach quality signals to generations after the fact. For MCP tools, useful scores include: the agent's downstream task success (did the agent complete its goal?), explicit user ratings passed back through a follow-up tool call, and automated model-graded evals run async after the generation completes.

// MCP tool — submit a score on a previously recorded trace/generation
// Call this when the user provides feedback or the agent reports task outcome
server.tool(
  'report_tool_outcome',
  {
    trace_id:  z.string().uuid(),
    outcome:   z.enum(['success', 'failure', 'partial']),
    comment:   z.string().max(500).optional(),
  },
  async ({ trace_id, outcome, comment }) => {
    const scoreValue = outcome === 'success' ? 1 : outcome === 'partial' ? 0.5 : 0;

    await langfuse.score({
      traceId: trace_id,
      name:    'task_outcome',
      value:   scoreValue,
      comment: comment ?? `Agent reported outcome: ${outcome}`,
    });

    await langfuse.flushAsync();
    return {
      content: [{ type: 'text', text: JSON.stringify({ scored: true, traceId: trace_id }) }],
    };
  }
);

// Async model-graded eval — run after tool completes, not blocking the response
async function runLLMJudge(traceId: string, input: string, output: string): Promise<void> {
  try {
    const judgeCompletion = await openai.chat.completions.create({
      model:    'gpt-4o-mini',
      messages: [{
        role: 'user',
        content: `Rate this response from 0-1 for helpfulness and accuracy.
Input: ${input}
Output: ${output}
Respond with only a JSON object: {"score": number, "reason": "brief explanation"}`
      }],
      response_format: { type: 'json_object' },
    });

    const judgeResult = JSON.parse(judgeCompletion.choices[0]?.message?.content ?? '{}');

    await langfuse.score({
      traceId: traceId,
      name:    'llm_judge_quality',
      value:   judgeResult.score ?? 0,
      comment: judgeResult.reason ?? '',
    });

    await langfuse.flushAsync();
  } catch {
    // Judge failures don't propagate — scoring is best-effort
  }
}

Run runLLMJudge in the background after the tool handler returns by calling it without await at the end of a tool. The tool response reaches the agent immediately while the quality signal is recorded asynchronously. Over time, Langfuse score distributions show you which tools degrade in quality as models or prompts change.