From bb23d481e1967388b5ab845145f869d54e16f0d5 Mon Sep 17 00:00:00 2001 From: Roberto Bianchi Date: Fri, 21 Aug 2026 11:43:25 +0200 Subject: [PATCH] feat(observability): add tool call completion hook Signed-off-by: Roberto Bianchi update Signed-off-by: Roberto Bianchi --- README.md | 28 ++++ src/handlers.ts | 78 +++++++++- src/index.ts | 1 + src/types.ts | 40 +++++ test-d/index.test-d.ts | 26 ++++ test/tool-call-complete.test.ts | 266 ++++++++++++++++++++++++++++++++ 6 files changed, 432 insertions(+), 7 deletions(-) create mode 100644 test/tool-call-complete.test.ts diff --git a/README.md b/README.md index c908d1a..c422504 100644 --- a/README.md +++ b/README.md @@ -1558,6 +1558,33 @@ Visibility and execution policies may intentionally differ — a tool hidden fro A hook that throws denies access (fail closed) and logs a warning. Only an explicit `true` grants access. Without the hook, every registered tool stays visible and callable. +## Observing Tool Calls + +Provide `onToolCallComplete` to observe every tool call, regardless of transport, in one place: + +```typescript +await app.register(mcpPlugin, { + onToolCallComplete: async (event) => { + logger.info({ + tool: event.toolName, + source: event.source, + durationMs: event.durationMs, + ok: event.outcome.ok + }, 'tool call completed') + } +}) +``` + +The hook fires exactly once per tool call, after it settles: for JSON-RPC `tools/call`, for `app.mcpCallTool()`, and for a task-augmented call once execution finishes (not when the task is merely accepted). It also fires for access denial, unknown tools, invalid arguments, and task-required outcomes. It does not fire for `tools/list`, initialization/ping, or malformed requests where no tool name was resolved. + +`event.source` is `'json-rpc'`, `'in-process'`, or `'task'`. `event.outcome` is the same `McpCallToolOutcome` returned by `mcpCallTool()`. The hook is awaited before the response is sent, so keep it fast; a throwing or rejecting hook is logged and otherwise ignored — it never changes the tool response. + +Synchronous sources (`'json-rpc'` and `'in-process'`) expose the active `request`/`reply`. Task events never do: a task can finish after the original HTTP response has already completed, so `request`/`reply` are absent on `source: 'task'` events. Every event, regardless of source, carries `requestId` for correlation. + +Duration semantics also differ by source: `'json-rpc'` and `'in-process'` durations include authorization, validation, sanitization, and execution. Task durations measure only actual execution — validation, sanitization, and the tool call — excluding the time the task spent queued. + +`event.arguments` are the raw, unredacted tool arguments. Redact sensitive values yourself before logging or persisting them. + ## Customizing MCP Route Schemas Use `transformRouteSchema` to customize Fastify/OpenAPI schema metadata on MCP transport routes without replacing handlers or route definitions. @@ -1824,6 +1851,7 @@ await app.register(import('@fastify/bearer-auth'), { - `instructions`: Optional server instructions - `enableSSE`: Enable Server-Sent Events support (default: false) - `canAccessTool`: Per-request tool authorization hook consulted by `tools/list` and `tools/call` (optional) +- `onToolCallComplete`: Transport-neutral hook fired once after every tool call settles, across JSON-RPC, `mcpCallTool()`, and tasks (optional) - `authorization`: OAuth 2.1 authorization configuration (optional) - `enabled`: Enable OAuth 2.1 authorization (default: false) - `authorizationServers`: Authorization server URIs diff --git a/src/handlers.ts b/src/handlers.ts index 4ffa6d5..d2098fc 100644 --- a/src/handlers.ts +++ b/src/handlers.ts @@ -31,7 +31,7 @@ import { } from './schema.ts' import type { RequestId } from './schema.ts' -import type { MCPTool, MCPResource, MCPPrompt, MCPPluginOptions, ResourceHandlers, McpCallToolOutcome, ToolAccessOperation } from './types.ts' +import type { MCPTool, MCPResource, MCPPrompt, MCPPluginOptions, ResourceHandlers, McpCallToolOutcome, ToolAccessOperation, MCPToolCallCompleteEvent } from './types.ts' import type { SessionStore } from './stores/session-store.ts' import type { TaskStore, TaskRecord, TaskWaiters } from './stores/task-store.ts' import { isTerminal, toWireTask } from './stores/task-store.ts' @@ -300,6 +300,7 @@ async function handleToolsCall ( const params = paramsValidation.data const toolName = params.name + const startedAt = performance.now() // A denied tool answers exactly like an unknown one, so a caller cannot // distinguish "does not exist" from "exists but not for you" by the protocol @@ -310,6 +311,7 @@ async function handleToolsCall ( // the model cannot correct itself out of missing access. const resolved = await resolveRegisteredTool(toolName, dependencies) if (!resolved.ok) { + await emitToolCallComplete('json-rpc', toolName, params.arguments || {}, resolved, startedAt, dependencies) return toolCallOutcomeToJsonRpc(request.id, toolName, resolved) } @@ -322,18 +324,64 @@ async function handleToolsCall ( : undefined const augmentation = resolveTaskAugmentation(resolved.tool, taskParams !== undefined) if ('error' in augmentation) { + await emitToolCallComplete('json-rpc', toolName, params.arguments || {}, { ok: false, reason: 'task-required' }, startedAt, dependencies) return createError(request.id, METHOD_NOT_FOUND, augmentation.error) } if (augmentation.mode === 'task') { return await runToolCallAsTask( request, taskParams?.ttl, - () => executeToolCall(request, resolved.tool, params, sessionId, dependencies), + // Timed from when the task actually starts executing, not from when it + // was queued, so `durationMs` reflects work done rather than wait time. + () => executeToolCall(request, resolved.tool, params, sessionId, dependencies, { source: 'task', startedAt: performance.now() }), dependencies ) } - return await executeToolCall(request, resolved.tool, params, sessionId, dependencies) + return await executeToolCall(request, resolved.tool, params, sessionId, dependencies, { source: 'json-rpc', startedAt }) +} + +/** An observability failure must never change the tool response. */ +async function emitToolCallComplete ( + source: MCPToolCallCompleteEvent['source'], + toolName: string, + args: Record, + outcome: McpCallToolOutcome, + startedAt: number, + dependencies: ToolCallDependencies +): Promise { + const hook = dependencies.opts.onToolCallComplete + if (!hook) { + return + } + + const common = { + toolName, + arguments: args, + authContext: dependencies.authContext, + sessionId: dependencies.sessionId, + requestId: dependencies.request.id, + durationMs: performance.now() - startedAt, + outcome + } + + // Tasks may complete after the originating HTTP response has already been + // sent, so they must never carry the (possibly finished) request/reply. + const event: MCPToolCallCompleteEvent = source === 'task' + ? { ...common, source } + : { ...common, source, request: dependencies.request, reply: dependencies.reply } + + try { + await hook(event) + } catch (error) { + if (source === 'task') { + // request.log may belong to an already-finished request; app.log plus + // the correlation id is the safe choice for task-time failures. + dependencies.app.log.error({ err: error, tool: toolName, requestId: common.requestId }, 'onToolCallComplete hook failed') + } else { + dependencies.request.log.error({ err: error, tool: toolName }, 'onToolCallComplete hook failed') + } + } } async function resolveRegisteredTool ( @@ -392,17 +440,29 @@ export async function callRegisteredTool ( args: Record, dependencies: ToolCallDependencies ): Promise { + const startedAt = performance.now() + const resolved = await resolveRegisteredTool(name, dependencies) if (!resolved.ok) { + await emitToolCallComplete('in-process', name, args, resolved, startedAt, dependencies) return resolved } const augmentation = resolveTaskAugmentation(resolved.tool, false) if ('error' in augmentation) { - return { ok: false, reason: 'task-required' } + const outcome: McpCallToolOutcome = { ok: false, reason: 'task-required' } + await emitToolCallComplete('in-process', name, args, outcome, startedAt, dependencies) + return outcome } - return await executeRegisteredTool(resolved.tool, name, args, dependencies) + const outcome = await executeRegisteredTool(resolved.tool, name, args, dependencies) + await emitToolCallComplete('in-process', name, args, outcome, startedAt, dependencies) + return outcome +} + +interface ToolCallObservationContext { + source: MCPToolCallCompleteEvent['source'] + startedAt: number } async function executeToolCall ( @@ -410,11 +470,15 @@ async function executeToolCall ( tool: MCPTool, params: { name: string, arguments?: Record }, sessionId: string | undefined, - dependencies: HandlerDependencies + dependencies: HandlerDependencies, + observation: ToolCallObservationContext ): Promise { const toolName = params.name + const args = params.arguments || {} - const outcome = await executeRegisteredTool(tool, toolName, params.arguments || {}, { ...dependencies, sessionId }) + const callDependencies = { ...dependencies, sessionId } + const outcome = await executeRegisteredTool(tool, toolName, args, callDependencies) + await emitToolCallComplete(observation.source, toolName, args, outcome, observation.startedAt, callDependencies) return toolCallOutcomeToJsonRpc(request.id, toolName, outcome) } diff --git a/src/index.ts b/src/index.ts index 79c7fc0..a077ca2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -280,6 +280,7 @@ export type { ToolAccessOperation, McpCallToolContext, McpCallToolOutcome, + MCPToolCallCompleteEvent, MCPTool, MCPResource, MCPPrompt, diff --git a/src/types.ts b/src/types.ts index 6bb1a15..130a9fb 100644 --- a/src/types.ts +++ b/src/types.ts @@ -224,6 +224,34 @@ export type McpCallToolOutcome = | { ok: false, reason: 'invalid-arguments', detail: string } | { ok: false, reason: 'task-required' } +interface MCPToolCallCompleteEventBase { + toolName: string + arguments: Record + authContext?: AuthorizationContext + sessionId?: string + /** Correlates this event back to the originating request, for every source */ + requestId: string + durationMs: number + outcome: McpCallToolOutcome +} + +/** + * `source: 'task'` fires after the original HTTP response may already have + * completed, so it must not carry the (possibly finished) `request`/`reply` + * objects that the synchronous sources expose. + */ +export type MCPToolCallCompleteEvent = + | (MCPToolCallCompleteEventBase & { + source: 'json-rpc' | 'in-process' + request: FastifyRequest + reply: FastifyReply + }) + | (MCPToolCallCompleteEventBase & { + source: 'task' + request?: undefined + reply?: undefined + }) + export interface MCPPluginOptions { serverInfo?: Implementation capabilities?: ServerCapabilities @@ -275,6 +303,18 @@ export interface MCPPluginOptions { toolName: string, context: ToolAccessContext ) => boolean | Promise + /** + * Fires once per tool call, after it settles, regardless of transport + * (JSON-RPC, `app.mcpCallTool()`, or a completed task). Not called for + * `tools/list`, initialization/ping, or malformed requests where no tool + * name was resolved. Awaited before the response is sent, so keep it fast; + * a throwing hook is logged and otherwise ignored, it never changes the + * tool response. `event.arguments` are the raw, unredacted tool + * arguments, so redact sensitive values yourself before logging them. + */ + onToolCallComplete?: ( + event: MCPToolCallCompleteEvent + ) => void | Promise /** * Customize Fastify/OpenAPI schema metadata for MCP transport routes. * This callback runs once per registered route during startup. diff --git a/test-d/index.test-d.ts b/test-d/index.test-d.ts index 5036ede..c14b99e 100644 --- a/test-d/index.test-d.ts +++ b/test-d/index.test-d.ts @@ -32,6 +32,7 @@ import type { MCPRouteSchemaTransformer, McpCallToolContext, McpCallToolOutcome, + MCPToolCallCompleteEvent, } from '../dist/index.js' // ─── ToolHandler ───────────────────────────────────────────────────── @@ -395,6 +396,31 @@ expectAssignable({ ok: false, reason: 'invalid-arguments', d expectAssignable({ ok: false, reason: 'task-required' }) expectNotAssignable({ ok: false, reason: 'nope' }) +// ─── onToolCallComplete event ─────────────────────────────────────── + +// Every source carries a requestId +expectType(({} as MCPToolCallCompleteEvent).requestId) + +// Synchronous sources expose the live request/reply +declare const jsonRpcEvent: MCPToolCallCompleteEvent & { source: 'json-rpc' | 'in-process' } +expectType(jsonRpcEvent.request) +expectType(jsonRpcEvent.reply) + +// Task events cannot carry request/reply +declare const taskEvent: MCPToolCallCompleteEvent & { source: 'task' } +expectType(taskEvent.request) +expectType(taskEvent.reply) +expectNotAssignable({ + source: 'task', + toolName: 'echo', + arguments: {}, + requestId: 'req-1', + durationMs: 1, + outcome: { ok: true, result: { content: [] } }, + request: {} as FastifyRequest, + reply: {} as FastifyReply, +}) + // Sync and async hooks are both accepted expectAssignable({ canAccessTool: () => true }) expectAssignable({ canAccessTool: async () => false }) diff --git a/test/tool-call-complete.test.ts b/test/tool-call-complete.test.ts new file mode 100644 index 0000000..910a946 --- /dev/null +++ b/test/tool-call-complete.test.ts @@ -0,0 +1,266 @@ +import { test, describe } from 'node:test' +import type { TestContext } from 'node:test' +import Fastify from 'fastify' +import { Type } from '@sinclair/typebox' +import mcpPlugin from '../src/index.ts' +import type { MCPToolCallCompleteEvent } from '../src/index.ts' +import { JSONRPC_VERSION, LATEST_PROTOCOL_VERSION } from '../src/schema.ts' + +async function call (app: any, method: string, params: unknown, id = 1) { + const response = await app.inject({ + method: 'POST', + url: '/mcp', + headers: { 'mcp-protocol-version': LATEST_PROTOCOL_VERSION }, + payload: { jsonrpc: JSONRPC_VERSION, id, method, params } + }) + return response.json() +} + +/** Poll tasks/get until the task leaves the `working` state */ +async function waitForTaskEvent (events: () => MCPToolCallCompleteEvent[]): Promise { + for (let i = 0; i < 100; i++) { + if (events().some(e => e.source === 'task')) { + return + } + await new Promise(resolve => setTimeout(resolve, 10)) + } + throw new Error('task never emitted a completion event') +} + +async function buildApp (t: TestContext, opts: Record = {}) { + const events: MCPToolCallCompleteEvent[] = [] + + const app = Fastify({ logger: false }) + t.after(() => app.close()) + + await app.register(mcpPlugin, { + enableTasks: true, + canAccessTool: (toolName) => toolName !== 'denied', + onToolCallComplete: (event) => { events.push(event) }, + ...opts + }) + + app.mcpAddTool({ + name: 'echo', + description: 'Echo a message', + inputSchema: Type.Object({ message: Type.String() }) + }, async (args) => ({ content: [{ type: 'text', text: args.message }] })) + + app.mcpAddTool({ + name: 'boom', + description: 'Throws from handler', + inputSchema: Type.Object({}) + }, async () => { throw new Error('kaboom') }) + + app.mcpAddTool({ + name: 'denied', + description: 'Denied by canAccessTool', + inputSchema: Type.Object({}) + }, async () => ({ content: [{ type: 'text', text: 'nope' }] })) + + app.mcpAddTool({ + name: 'slow', + description: 'Optional task support', + inputSchema: Type.Object({}), + execution: { taskSupport: 'optional' } + } as any, async () => ({ content: [{ type: 'text', text: 'task ok' }] })) + + app.mcpAddTool({ + name: 'task-only', + description: 'Must be invoked as a task', + inputSchema: Type.Object({}), + execution: { taskSupport: 'required' } + } as any, async () => ({ content: [{ type: 'text', text: 'done' }] })) + + app.post('/direct-tool-call', async (request, reply) => { + const body = request.body as { name: string, args?: Record } + return await app.mcpCallTool(body.name, body.args ?? {}, { request, reply }) + }) + + await app.ready() + + return { app, events: () => events } +} + +describe('onToolCallComplete', () => { + test('a successful JSON-RPC call emits one event', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await call(app, 'tools/call', { name: 'echo', arguments: { message: 'hi' } }) + + t.assert.strictEqual(events().length, 1) + t.assert.strictEqual(events()[0].source, 'json-rpc') + t.assert.strictEqual(events()[0].toolName, 'echo') + t.assert.deepStrictEqual(events()[0].arguments, { message: 'hi' }) + }) + + test('an in-process call emits one event with source: in-process', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await app.inject({ method: 'POST', url: '/direct-tool-call', payload: { name: 'echo', args: { message: 'hi' } } }) + + t.assert.strictEqual(events().length, 1) + t.assert.strictEqual(events()[0].source, 'in-process') + }) + + test('invalid arguments expose the structured outcome', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await call(app, 'tools/call', { name: 'echo', arguments: {} }) + + t.assert.strictEqual(events().length, 1) + const outcome = events()[0].outcome + t.assert.strictEqual(outcome.ok, false) + t.assert.strictEqual(!outcome.ok && outcome.reason, 'invalid-arguments') + }) + + test('denied and unknown tools expose their normalized not-found outcome', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await call(app, 'tools/call', { name: 'denied', arguments: {} }) + await call(app, 'tools/call', { name: 'does-not-exist', arguments: {} }) + + t.assert.strictEqual(events().length, 2) + for (const event of events()) { + t.assert.deepStrictEqual(event.outcome, { ok: false, reason: 'not-found' }) + } + }) + + test('a handler error exposes the resulting CallToolResult', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await call(app, 'tools/call', { name: 'boom', arguments: {} }) + + t.assert.strictEqual(events().length, 1) + const outcome = events()[0].outcome + t.assert.strictEqual(outcome.ok, true) + t.assert.strictEqual(outcome.ok && outcome.result.isError, true) + }) + + test('a task-required tool called directly emits a task-required outcome', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await call(app, 'tools/call', { name: 'task-only', arguments: {} }) + + t.assert.strictEqual(events().length, 1) + t.assert.deepStrictEqual(events()[0].outcome, { ok: false, reason: 'task-required' }) + }) + + test('the original request, auth context, arguments, and session are provided', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await app.inject({ + method: 'POST', + url: '/mcp', + headers: { 'mcp-protocol-version': LATEST_PROTOCOL_VERSION }, + payload: { jsonrpc: JSONRPC_VERSION, id: 1, method: 'tools/call', params: { name: 'echo', arguments: { message: 'hi' } } } + }) + + const event = events()[0] + t.assert.ok(event.request) + t.assert.ok(event.reply) + t.assert.deepStrictEqual(event.arguments, { message: 'hi' }) + }) + + test('duration is a non-negative number', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await call(app, 'tools/call', { name: 'echo', arguments: { message: 'hi' } }) + + t.assert.strictEqual(typeof events()[0].durationMs, 'number') + t.assert.ok(events()[0].durationMs >= 0) + }) + + test('every event carries a requestId for correlation', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await call(app, 'tools/call', { name: 'echo', arguments: { message: 'hi' } }) + + t.assert.strictEqual(typeof events()[0].requestId, 'string') + t.assert.ok(events()[0].requestId.length > 0) + }) + + test('task events do not expose completed request lifecycle objects', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await call(app, 'tools/call', { name: 'slow', arguments: {}, task: {} }) + + await waitForTaskEvent(events) + + const event = events().find(candidate => candidate.source === 'task') + + t.assert.ok(event) + t.assert.strictEqual('request' in event!, false) + t.assert.strictEqual('reply' in event!, false) + t.assert.strictEqual(typeof event!.requestId, 'string') + t.assert.ok(event!.requestId.length > 0) + }) + + test('JSON-RPC duration includes authorization', async (t: TestContext) => { + const authorizationDelayMs = 30 + + const { app, events } = await buildApp(t, { + canAccessTool: async () => { + await new Promise(resolve => setTimeout(resolve, authorizationDelayMs)) + return true + } + }) + + await call(app, 'tools/call', { name: 'echo', arguments: { message: 'hello' } }) + + t.assert.strictEqual(events().length, 1) + t.assert.ok(events()[0].durationMs >= authorizationDelayMs - 5) + }) + + test('in-process duration includes authorization', async (t: TestContext) => { + const authorizationDelayMs = 30 + + const { app, events } = await buildApp(t, { + canAccessTool: async () => { + await new Promise(resolve => setTimeout(resolve, authorizationDelayMs)) + return true + } + }) + + await app.inject({ method: 'POST', url: '/direct-tool-call', payload: { name: 'echo', args: { message: 'hi' } } }) + + t.assert.strictEqual(events().length, 1) + t.assert.ok(events()[0].durationMs >= authorizationDelayMs - 5) + }) + + test('a throwing observer does not change the MCP response', async (t: TestContext) => { + const app = Fastify({ logger: false }) + t.after(() => app.close()) + await app.register(mcpPlugin, { onToolCallComplete: () => { throw new Error('observer exploded') } }) + app.mcpAddTool({ + name: 'echo', + description: 'Echo a message', + inputSchema: Type.Object({ message: Type.String() }) + }, async (args) => ({ content: [{ type: 'text', text: args.message }] })) + await app.ready() + + const body = await call(app, 'tools/call', { name: 'echo', arguments: { message: 'hi' } }) + + t.assert.deepStrictEqual(body.result, { content: [{ type: 'text', text: 'hi' }] }) + }) + + test('tools/list does not emit an event', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await call(app, 'tools/list', {}) + + t.assert.strictEqual(events().length, 0) + }) + + test('a task emits one event, with source: task, when execution finishes', async (t: TestContext) => { + const { app, events } = await buildApp(t) + + await call(app, 'tools/call', { name: 'slow', arguments: {}, task: {} }) + + await waitForTaskEvent(events) + + t.assert.strictEqual(events().length, 1) + t.assert.strictEqual(events()[0].source, 'task') + t.assert.strictEqual(events()[0].outcome.ok, true) + }) +})