feat(langfuse): add tool-level spans with input/output and trace-level I/O

Each MCP tool handler now sets:
- LANGFUSE_TRACE_OUTPUT_ATTR on tool spans (maps to top-level trace output)
- LANGFUSE_OBSERVATION_INPUT_ATTR and LANGFUSE_OBSERVATION_OUTPUT_ATTR
  for detailed child observation views

HTTP root span renamed to 'mcp_http_request' for consistency with stdio.
All attribute keys use constants from langfuse-transport.ts.
This commit is contained in:
2026-04-27 18:56:50 +05:30
parent 54bc20cb7f
commit d71726205d
3 changed files with 162 additions and 12 deletions
+42
View File
@@ -1,8 +1,50 @@
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { LangfuseSpanProcessor } from "@langfuse/otel"; import { LangfuseSpanProcessor } from "@langfuse/otel";
/** Trace name used for HTTP transport in Langfuse. */
export const TRACE_NAME_HTTP = "ask262-http";
/** Trace name used for stdio transport in Langfuse. */
export const TRACE_NAME_STDIO = "ask262-stdio";
/** Langfuse attribute key for setting the trace name on a span. */
export const LANGFUSE_TRACE_NAME_ATTR = "langfuse.trace.name";
/** Langfuse attribute key for observation input. */
export const LANGFUSE_OBSERVATION_INPUT_ATTR = "langfuse.observation.input";
/** Langfuse attribute key for observation output. */
export const LANGFUSE_OBSERVATION_OUTPUT_ATTR = "langfuse.observation.output";
/** Langfuse attribute key for trace-level input. */
export const LANGFUSE_TRACE_INPUT_ATTR = "langfuse.trace.input";
/** Langfuse attribute key for trace-level output. */
export const LANGFUSE_TRACE_OUTPUT_ATTR = "langfuse.trace.output";
let provider: NodeTracerProvider | null = null; let provider: NodeTracerProvider | null = null;
/**
* Extract MCP tool call information from a JSON-RPC request body.
* Used to populate trace-level input in Langfuse.
*/
export function extractMcpToolInfo(
body: unknown,
): { method?: string; tool?: string; input?: unknown } {
if (typeof body !== "object" || body === null) return {};
const b = body as Record<string, unknown>;
const rpcMethod = b.method as string | undefined;
if (rpcMethod === "tools/call") {
const params = b.params as Record<string, unknown> | undefined;
return {
method: rpcMethod,
tool: params?.name as string | undefined,
input: params?.arguments,
};
}
return { method: rpcMethod };
}
/** /**
* Initialize Langfuse OTel span processor. * Initialize Langfuse OTel span processor.
* Called once at server startup before any spans are created. * Called once at server startup before any spans are created.
+58 -8
View File
@@ -36,6 +36,14 @@ import {
import { DEFAULT_PORT, STORAGE_DIR as STORAGE_DIR_REL } from "./constants.js"; import { DEFAULT_PORT, STORAGE_DIR as STORAGE_DIR_REL } from "./constants.js";
import { createEmbeddings } from "./lib/embeddings-factory.js"; import { createEmbeddings } from "./lib/embeddings-factory.js";
import { LogOperation, logger } from "./lib/logger.js"; import { LogOperation, logger } from "./lib/logger.js";
import { trace } from "@opentelemetry/api";
import {
LANGFUSE_OBSERVATION_INPUT_ATTR,
LANGFUSE_OBSERVATION_OUTPUT_ATTR,
LANGFUSE_TRACE_NAME_ATTR,
LANGFUSE_TRACE_OUTPUT_ATTR,
TRACE_NAME_HTTP,
} from "./lib/langfuse-transport.js";
import { getSessionMetadata, setupTracing, withSpan } from "./lib/tracing.js"; import { getSessionMetadata, setupTracing, withSpan } from "./lib/tracing.js";
// Resolve storage path relative to this script's directory // Resolve storage path relative to this script's directory
@@ -93,12 +101,24 @@ async function createMcpServer() {
return await withSpan( return await withSpan(
"ask262_search_spec_sections", "ask262_search_spec_sections",
{ {
"langfuse.observation.input": JSON.stringify({ query }), [LANGFUSE_OBSERVATION_INPUT_ATTR]: JSON.stringify({ query }),
tool: searchSpecToolName, tool: searchSpecToolName,
query, query,
}, },
async () => { async () => {
const result = await searchSpecTool({ query }); const result = await searchSpecTool({ query });
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_TRACE_OUTPUT_ATTR,
JSON.stringify(result),
);
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_OBSERVATION_OUTPUT_ATTR,
JSON.stringify(result),
);
return { return {
content: [{ type: "text", text: JSON.stringify(result, null, 2) }], content: [{ type: "text", text: JSON.stringify(result, null, 2) }],
structuredContent: result, structuredContent: result,
@@ -126,7 +146,7 @@ async function createMcpServer() {
return await withSpan( return await withSpan(
"ask262_get_section_content", "ask262_get_section_content",
{ {
"langfuse.observation.input": JSON.stringify({ [LANGFUSE_OBSERVATION_INPUT_ATTR]: JSON.stringify({
sectionIds, sectionIds,
recursive, recursive,
}), }),
@@ -135,6 +155,18 @@ async function createMcpServer() {
}, },
async () => { async () => {
const result = await getSectionContentTool({ sectionIds, recursive }); const result = await getSectionContentTool({ sectionIds, recursive });
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_TRACE_OUTPUT_ATTR,
JSON.stringify(result),
);
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_OBSERVATION_OUTPUT_ATTR,
JSON.stringify(result),
);
return { return {
content: [{ type: "text", text: JSON.stringify(result, null, 2) }], content: [{ type: "text", text: JSON.stringify(result, null, 2) }],
structuredContent: result, structuredContent: result,
@@ -162,14 +194,24 @@ async function createMcpServer() {
return await withSpan( return await withSpan(
"ask262_evaluate_in_engine262", "ask262_evaluate_in_engine262",
{ {
"langfuse.observation.input": JSON.stringify({ [LANGFUSE_OBSERVATION_INPUT_ATTR]: JSON.stringify({ code }),
code: code.slice(0, 200),
}),
tool: evaluateToolName, tool: evaluateToolName,
code_length: code.length, code_length: code.length,
}, },
async () => { async () => {
const result = await evaluateTool({ code }); const result = await evaluateTool({ code });
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_TRACE_OUTPUT_ATTR,
JSON.stringify(result),
);
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_OBSERVATION_OUTPUT_ATTR,
JSON.stringify(result),
);
const isError = result.error !== undefined; const isError = result.error !== undefined;
const text = isError const text = isError
? result.error ? result.error
@@ -259,8 +301,17 @@ export async function main() {
// Handle request within trace context (passing trace ID from header if available) // Handle request within trace context (passing trace ID from header if available)
const sessionMetadata = getSessionMetadata("http"); const sessionMetadata = getSessionMetadata("http");
return await withSpan( return await withSpan(
LogOperation.HANDLING_MCP_HTTP_REQUEST, "mcp_http_request",
{ method: c.req.method, client_ip: clientIp }, {
[LANGFUSE_TRACE_NAME_ATTR]: TRACE_NAME_HTTP,
[LANGFUSE_OBSERVATION_INPUT_ATTR]: JSON.stringify({
method: c.req.method,
endpoint: "/mcp",
client_ip: clientIp,
}),
method: c.req.method,
client_ip: clientIp,
},
async () => { async () => {
const op = log.start(LogOperation.HANDLING_MCP_HTTP_REQUEST, { const op = log.start(LogOperation.HANDLING_MCP_HTTP_REQUEST, {
method: c.req.method, method: c.req.method,
@@ -284,7 +335,6 @@ export async function main() {
op.end({ status: "success" }); op.end({ status: "success" });
// Return the Web Standard Response directly
return response; return response;
} catch (err) { } catch (err) {
const error = err instanceof Error ? err : new Error(String(err)); const error = err instanceof Error ? err : new Error(String(err));
+62 -4
View File
@@ -37,6 +37,15 @@ import {
import { STORAGE_DIR as STORAGE_DIR_REL } from "./constants.js"; import { STORAGE_DIR as STORAGE_DIR_REL } from "./constants.js";
import { createEmbeddings } from "./lib/embeddings-factory.js"; import { createEmbeddings } from "./lib/embeddings-factory.js";
import { LogOperation, logger } from "./lib/logger.js"; import { LogOperation, logger } from "./lib/logger.js";
import { trace } from "@opentelemetry/api";
import {
LANGFUSE_OBSERVATION_INPUT_ATTR,
LANGFUSE_OBSERVATION_OUTPUT_ATTR,
LANGFUSE_TRACE_INPUT_ATTR,
LANGFUSE_TRACE_NAME_ATTR,
LANGFUSE_TRACE_OUTPUT_ATTR,
TRACE_NAME_STDIO,
} from "./lib/langfuse-transport.js";
import { import {
createProcessScopedTrace, createProcessScopedTrace,
getSessionMetadata, getSessionMetadata,
@@ -149,12 +158,29 @@ export async function main() {
sessionTraceId, sessionTraceId,
"ask262_search_spec_sections", "ask262_search_spec_sections",
{ {
"langfuse.observation.input": JSON.stringify({ query }), LANGFUSE_TRACE_NAME_ATTR: TRACE_NAME_STDIO,
LANGFUSE_TRACE_INPUT_ATTR: JSON.stringify({
tool: searchSpecToolName,
input: { query },
}),
LANGFUSE_OBSERVATION_INPUT_ATTR: JSON.stringify({ query }),
tool: searchSpecToolName, tool: searchSpecToolName,
query, query,
}, },
async () => { async () => {
const result = await searchSpecTool({ query }); const result = await searchSpecTool({ query });
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_OBSERVATION_OUTPUT_ATTR,
JSON.stringify(result),
);
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_TRACE_OUTPUT_ATTR,
JSON.stringify(result),
);
return { return {
content: [{ type: "text", text: JSON.stringify(result, null, 2) }], content: [{ type: "text", text: JSON.stringify(result, null, 2) }],
structuredContent: result, structuredContent: result,
@@ -187,7 +213,12 @@ export async function main() {
sessionTraceId, sessionTraceId,
"ask262_get_section_content", "ask262_get_section_content",
{ {
"langfuse.observation.input": JSON.stringify({ LANGFUSE_TRACE_NAME_ATTR: TRACE_NAME_STDIO,
LANGFUSE_TRACE_INPUT_ATTR: JSON.stringify({
tool: sectionContentToolName,
input: { sectionIds, recursive },
}),
LANGFUSE_OBSERVATION_INPUT_ATTR: JSON.stringify({
sectionIds, sectionIds,
recursive, recursive,
}), }),
@@ -196,6 +227,18 @@ export async function main() {
}, },
async () => { async () => {
const result = await getSectionContentTool({ sectionIds, recursive }); const result = await getSectionContentTool({ sectionIds, recursive });
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_OBSERVATION_OUTPUT_ATTR,
JSON.stringify(result),
);
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_TRACE_OUTPUT_ATTR,
JSON.stringify(result),
);
return { return {
content: [{ type: "text", text: JSON.stringify(result, null, 2) }], content: [{ type: "text", text: JSON.stringify(result, null, 2) }],
structuredContent: result, structuredContent: result,
@@ -225,14 +268,29 @@ export async function main() {
sessionTraceId, sessionTraceId,
"ask262_evaluate_in_engine262", "ask262_evaluate_in_engine262",
{ {
"langfuse.observation.input": JSON.stringify({ LANGFUSE_TRACE_NAME_ATTR: TRACE_NAME_STDIO,
code: code.slice(0, 200), LANGFUSE_TRACE_INPUT_ATTR: JSON.stringify({
tool: evaluateToolName,
input: { code },
}), }),
LANGFUSE_OBSERVATION_INPUT_ATTR: JSON.stringify({ code }),
tool: evaluateToolName, tool: evaluateToolName,
code_length: code.length, code_length: code.length,
}, },
async () => { async () => {
const result = await evaluateTool({ code }); const result = await evaluateTool({ code });
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_OBSERVATION_OUTPUT_ATTR,
JSON.stringify(result),
);
trace
.getActiveSpan()
?.setAttribute(
LANGFUSE_TRACE_OUTPUT_ATTR,
JSON.stringify(result),
);
const isError = result.error !== undefined; const isError = result.error !== undefined;
const text = isError ? result.error : JSON.stringify(result, null, 2); const text = isError ? result.error : JSON.stringify(result, null, 2);
return { return {