dispatcher

(no submodules)

Purpose

JsonRpcDispatcher is the symmetric JSON-RPC 2.0 engine sitting under MCPServer and MCPClient. It owns id allocation, the pending-request map, request/response correlation, notification fan-out, abort propagation (in both directions), per-request timeout enforcement, and Zod-validated handler dispatch. Both the server and client wire one dispatcher to a Transport; the only difference between the two sides is which methods each one registers — the dispatcher itself is direction-agnostic.

Features

request(method, params, options?) — async, returns Promise<unknown>; correlates with the matching response or rejects on abort/timeout/transport-close.
notify(method, params?) — fire-and-forget notifications.
registerHandler(method, fn) / unregisterHandler(method) — server-side method handlers.
onNotification(method, fn) — subscribe to incoming notifications.
paramSchemas map — per-method Zod schemas validated before the handler runs; rejections respond with JSON-RPC -32602.
Cancellation: AbortSignal on a caller sends notifications/cancelled and rejects with MCPError(REQUEST_CANCELLED). Incoming notifications/cancelled aborts the matching in-flight handler.
Progress: handlers receive ctx.sendProgress(token, progress, total?) which emits notifications/progress; callers' onProgress hook is invoked when correlated by token.
OTel + telemetry: every request/notification increments the package meter via recordRequest / recordNotification.

Usage: Basic Example

import { JsonRpcDispatcher, InMemoryTransport } from "@suaveplan/mcp";

const [a, b] = InMemoryTransport.pair();
const left = new JsonRpcDispatcher({ side: "server", transport: a, dispatcherId: "L" });
const right = new JsonRpcDispatcher({ side: "client", transport: b, dispatcherId: "R" });

left.registerHandler("ping", async () => ({ ok: true }));
const result = await right.request("ping", undefined);

Usage: Advanced Example

import { JsonRpcDispatcher, MCPError, MCP_CODES } from "@suaveplan/mcp";
import z from "zod";

const schema = z.object({ msg: z.string() });
const d = new JsonRpcDispatcher({
    side: "server",
    transport,
    dispatcherId: "demo",
    paramSchemas: { "echo": schema },
});

d.registerHandler("echo", async (params, ctx) => {
    const ac = new AbortController();
    ctx.signal.addEventListener("abort", () => ac.abort());
    ctx.sendProgress("p", 1, 3);
    if (ctx.signal.aborted) throw new MCPError(MCP_CODES.REQUEST_CANCELLED, "cancelled");
    return { echoed: (params as { msg: string }).msg };
});

API Reference

new JsonRpcDispatcher(options: DispatcherOptions) — constructor takes the transport, dispatcher id, optional clock + id-generator, optional listeners + param-schema map.
request(method, params, options?: { signal?, timeoutMs?, onProgress? }) → Promise<unknown>.
notify(method, params?) → Promise<void>.
setProgressToken(method, token) — used by MCPClient to attach an auto-generated token to the most-recent pending request for forward-progress correlation.
close(reason?) — rejects every pending request with MCPTransportError(TRANSPORT_CLOSED), aborts every in-flight handler, and closes the transport.

Implementation Notes

The pending map keys requests by stringifyId(id) so numeric and string ids never collide. request() schedules the abort+timeout cleanup inside the executor so cancellation paths can synchronously remove the pending entry before resolution. Notifications never carry an id; the type guards in handleMessage discriminate request/response/notification by the id + method + result|error shape. The paramSchemas registry is keyed by the JSON-RPC method name so adding a new validated method is a one-line registration on the server side — the dispatcher itself stays unaware of MCP-specific semantics.

3.9 KiB Raw Blame History