Compare commits
5 Commits
fix/chat-s
...
fix/wire-s
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
eb0e97e76e | ||
|
|
065ce02a60 | ||
| db38de7e09 | |||
|
|
ac5dee906e | ||
| d4c33baf03 |
@@ -70,7 +70,7 @@ export function createChatCommand(deps: ChatCommandDeps): Command {
|
||||
}
|
||||
|
||||
/** What the chat is bound to: a named Agent or a Project. */
|
||||
interface ChatSubject {
|
||||
export interface ChatSubject {
|
||||
kind: 'agent' | 'project';
|
||||
name: string;
|
||||
/** URL segment, e.g. `agents/reviewer` or `projects/sre` (name url-encoded). */
|
||||
@@ -97,14 +97,17 @@ function resolveSubject(agent: string | undefined, opts: ChatOpts): ChatSubject
|
||||
* `personality` overlay (the project schema rejects unknown fields) and adds
|
||||
* `allowSecrets` when requested.
|
||||
*/
|
||||
function chatBody(subject: ChatSubject, message: string, threadId: string | undefined, overrides: Overrides, stream?: boolean): Record<string, unknown> {
|
||||
export function chatBody(subject: ChatSubject, message: string, threadId: string | undefined, overrides: Overrides, stream?: boolean): Record<string, unknown> {
|
||||
const o: Record<string, unknown> = { ...overrides };
|
||||
if (subject.kind === 'project') {
|
||||
delete o.personality;
|
||||
if (subject.allowSecrets) o.allowSecrets = true;
|
||||
}
|
||||
const body: Record<string, unknown> = { message, ...o };
|
||||
if (threadId !== undefined) body.threadId = threadId;
|
||||
// Guard the empty string, not just undefined: a turn that dies before its
|
||||
// `final` frame yields no thread id, and sending `threadId: ""` trips mcpd's
|
||||
// min(1) validation — bricking every later message in the REPL with a 400.
|
||||
if (threadId !== undefined && threadId !== '') body.threadId = threadId;
|
||||
if (stream === true) body.stream = true;
|
||||
return body;
|
||||
}
|
||||
@@ -205,7 +208,11 @@ async function runOneShot(
|
||||
const bar = installStatusBar();
|
||||
try {
|
||||
const finalThread = await streamOnce(deps, subject, message, threadId, overrides, bar);
|
||||
if (finalThread !== undefined) {
|
||||
process.stderr.write(`\n(thread: ${finalThread})\n`);
|
||||
} else {
|
||||
process.stderr.write('\n');
|
||||
}
|
||||
} finally {
|
||||
bar?.teardown();
|
||||
}
|
||||
@@ -262,7 +269,9 @@ async function runRepl(
|
||||
const answered = formatAnswered(res.llm, res.model, res.failedOver);
|
||||
if (answered !== '') process.stderr.write(`${styleStats(`(${answered})`)}\n`);
|
||||
} else {
|
||||
threadId = await streamOnce(deps, subject, line, threadId, overrides, bar);
|
||||
// A failed turn resolves undefined — keep the previous thread (or
|
||||
// none) instead of overwriting it, so the next message still works.
|
||||
threadId = await streamOnce(deps, subject, line, threadId, overrides, bar) ?? threadId;
|
||||
process.stdout.write('\n');
|
||||
}
|
||||
} catch (err) {
|
||||
@@ -502,15 +511,21 @@ async function chatRequestNonStream(
|
||||
});
|
||||
}
|
||||
|
||||
/** Stream a single chat call. Returns the resolved threadId. */
|
||||
async function streamOnce(
|
||||
/**
|
||||
* Stream a single chat call. Returns the resolved threadId, or undefined when
|
||||
* the turn never produced a `final` frame (upstream error, early disconnect).
|
||||
* Returning undefined — instead of the old '' — lets callers keep their
|
||||
* previous thread state rather than poisoning the next request with an empty
|
||||
* id that mcpd's validation rejects.
|
||||
*/
|
||||
export async function streamOnce(
|
||||
deps: ChatCommandDeps,
|
||||
subject: ChatSubject,
|
||||
message: string,
|
||||
threadId: string | undefined,
|
||||
overrides: Overrides,
|
||||
bar: StatusBar | null = null,
|
||||
): Promise<string> {
|
||||
): Promise<string | undefined> {
|
||||
const url = new URL(`${deps.baseUrl}/api/v1/${subject.path}/chat`);
|
||||
const body = JSON.stringify(chatBody(subject, message, threadId, overrides, true));
|
||||
|
||||
@@ -531,7 +546,7 @@ async function streamOnce(
|
||||
}
|
||||
}
|
||||
|
||||
return new Promise<string>((resolve, reject) => {
|
||||
return new Promise<string | undefined>((resolve, reject) => {
|
||||
const driver = url.protocol === 'https:' ? https : http;
|
||||
const req = driver.request({
|
||||
hostname: url.hostname,
|
||||
@@ -552,7 +567,7 @@ async function streamOnce(
|
||||
return;
|
||||
}
|
||||
let buf = '';
|
||||
let resolvedThread = threadId ?? '';
|
||||
let resolvedThread: string | undefined = threadId;
|
||||
let answered = '';
|
||||
res.setEncoding('utf-8');
|
||||
res.on('data', (chunk: string) => {
|
||||
|
||||
116
src/cli/tests/commands/chat-thread-brick.test.ts
Normal file
116
src/cli/tests/commands/chat-thread-brick.test.ts
Normal file
@@ -0,0 +1,116 @@
|
||||
/**
|
||||
* Regression: a failed first turn must not brick the REPL.
|
||||
*
|
||||
* Observed live: the first message died upstream (anthropic 429) before the
|
||||
* stream's `final` frame, streamOnce resolved '' as the thread id, the REPL
|
||||
* stored it, and every later message sent `threadId: ""` — which mcpd's
|
||||
* `z.string().min(1)` rejects with HTTP 400. The session was permanently
|
||||
* stuck: no turn could succeed again, so no `final` frame could ever repair
|
||||
* the thread id.
|
||||
*
|
||||
* The fix has two independent layers, pinned separately below:
|
||||
* 1. streamOnce resolves `undefined` (not '') when no `final` frame arrived,
|
||||
* and the REPL keeps its previous thread state on undefined;
|
||||
* 2. chatBody never serializes an empty threadId, even if one leaks in.
|
||||
*/
|
||||
import http from 'node:http';
|
||||
import type { AddressInfo } from 'node:net';
|
||||
|
||||
import { describe, it, expect, afterEach } from 'vitest';
|
||||
|
||||
import { chatBody, streamOnce } from '../../src/commands/chat.js';
|
||||
import type { ChatCommandDeps, ChatSubject } from '../../src/commands/chat.js';
|
||||
import type { ApiClient } from '../../src/api-client.js';
|
||||
|
||||
const subject: ChatSubject = {
|
||||
kind: 'agent',
|
||||
name: 'reviewer',
|
||||
path: 'agents/reviewer',
|
||||
allowSecrets: false,
|
||||
};
|
||||
|
||||
// streamOnce only touches baseUrl + token; the ApiClient is for the
|
||||
// non-streaming path and never dereferenced here.
|
||||
function depsFor(baseUrl: string): ChatCommandDeps {
|
||||
return { client: null as unknown as ApiClient, baseUrl, log: () => {} };
|
||||
}
|
||||
|
||||
let server: http.Server | null = null;
|
||||
|
||||
afterEach(async () => {
|
||||
if (server !== null) {
|
||||
await new Promise<void>((r) => server!.close(() => r()));
|
||||
server = null;
|
||||
}
|
||||
});
|
||||
|
||||
/** Serve one SSE response body for any POST, return the base URL. */
|
||||
async function serveSse(frames: string[]): Promise<string> {
|
||||
server = http.createServer((_req, res) => {
|
||||
res.writeHead(200, { 'Content-Type': 'text/event-stream' });
|
||||
for (const f of frames) res.write(`data: ${f}\n\n`);
|
||||
res.end();
|
||||
});
|
||||
await new Promise<void>((r) => server!.listen(0, '127.0.0.1', r));
|
||||
const { port } = server.address() as AddressInfo;
|
||||
return `http://127.0.0.1:${String(port)}`;
|
||||
}
|
||||
|
||||
describe('chatBody — threadId serialization', () => {
|
||||
it('omits threadId when undefined', () => {
|
||||
expect(chatBody(subject, 'hi', undefined, {})).not.toHaveProperty('threadId');
|
||||
});
|
||||
|
||||
it('omits threadId when empty — the exact payload that 400s against mcpd', () => {
|
||||
expect(chatBody(subject, 'hi', '', {})).not.toHaveProperty('threadId');
|
||||
});
|
||||
|
||||
it('includes a real threadId', () => {
|
||||
expect(chatBody(subject, 'hi', 'cthread123', {})).toHaveProperty('threadId', 'cthread123');
|
||||
});
|
||||
});
|
||||
|
||||
describe('streamOnce — thread id after a failed turn', () => {
|
||||
it('resolves undefined when the stream errors before any final frame', async () => {
|
||||
const base = await serveSse([
|
||||
'{"type":"error","message":"anthropic stream: HTTP 429"}',
|
||||
'[DONE]',
|
||||
]);
|
||||
const resolved = await streamOnce(depsFor(base), subject, 'hi', undefined, {});
|
||||
expect(resolved).toBeUndefined();
|
||||
});
|
||||
|
||||
it('keeps the caller-supplied thread when the turn fails mid-conversation', async () => {
|
||||
const base = await serveSse([
|
||||
'{"type":"error","message":"upstream died"}',
|
||||
'[DONE]',
|
||||
]);
|
||||
const resolved = await streamOnce(depsFor(base), subject, 'hi', 'cexisting1', {});
|
||||
expect(resolved).toBe('cexisting1');
|
||||
});
|
||||
|
||||
it('resolves the threadId announced by the final frame', async () => {
|
||||
const base = await serveSse([
|
||||
'{"type":"text","delta":"pong"}',
|
||||
'{"type":"final","threadId":"cfresh42"}',
|
||||
'[DONE]',
|
||||
]);
|
||||
const resolved = await streamOnce(depsFor(base), subject, 'hi', undefined, {});
|
||||
expect(resolved).toBe('cfresh42');
|
||||
});
|
||||
|
||||
it('REPL chain: failed turn 1 leaves turn 2 sendable (the brick)', async () => {
|
||||
const base = await serveSse([
|
||||
'{"type":"error","message":"anthropic stream: HTTP 429"}',
|
||||
'[DONE]',
|
||||
]);
|
||||
// Mirrors runRepl's assignment: threadId = streamOnce(...) ?? threadId
|
||||
let threadId: string | undefined = undefined;
|
||||
threadId = (await streamOnce(depsFor(base), subject, 'hi', threadId, {})) ?? threadId;
|
||||
|
||||
// Turn 2's body must be valid for mcpd: no threadId key at all.
|
||||
const body = chatBody(subject, 'hi again', threadId, {}, true);
|
||||
expect(body).not.toHaveProperty('threadId');
|
||||
expect(body).toHaveProperty('message', 'hi again');
|
||||
});
|
||||
});
|
||||
@@ -13,6 +13,7 @@ import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/
|
||||
import type { JSONRPCMessage } from '@modelcontextprotocol/sdk/types.js';
|
||||
import type { McpRouter } from '../router.js';
|
||||
import type { JsonRpcRequest } from '../types.js';
|
||||
import { WireNameCodec, routeWithWireNames } from '../util/wire-names.js';
|
||||
|
||||
interface SessionEntry {
|
||||
transport: StreamableHTTPServerTransport;
|
||||
@@ -20,6 +21,9 @@ interface SessionEntry {
|
||||
|
||||
export function registerMcpEndpoint(app: FastifyInstance, router: McpRouter): void {
|
||||
const sessions = new Map<string, SessionEntry>();
|
||||
// One codec for the shared router: serve OpenAI-safe tool names on the wire,
|
||||
// map tools/call names back to the internal `server/tool` form.
|
||||
const wireCodec = new WireNameCodec();
|
||||
|
||||
// POST /mcp — JSON-RPC requests (initialize, tools/call, etc.)
|
||||
app.post('/mcp', async (request, reply) => {
|
||||
@@ -52,7 +56,11 @@ export function registerMcpEndpoint(app: FastifyInstance, router: McpRouter): vo
|
||||
transport.onmessage = async (message: JSONRPCMessage) => {
|
||||
// The transport sends us JSON-RPC messages; route them through McpRouter
|
||||
if ('method' in message && 'id' in message) {
|
||||
const response = await router.route(message as unknown as JsonRpcRequest);
|
||||
const response = await routeWithWireNames(
|
||||
wireCodec,
|
||||
(req) => router.route(req),
|
||||
message as unknown as JsonRpcRequest,
|
||||
);
|
||||
await transport.send(response as unknown as JSONRPCMessage);
|
||||
}
|
||||
// Notifications (no id) are ignored — router doesn't handle inbound notifications
|
||||
|
||||
@@ -27,6 +27,7 @@ import { createFavouriteIndexPlugin } from '../proxymodel/plugins/favourite-inde
|
||||
import { composePlugins } from '../proxymodel/plugins/compose.js';
|
||||
import type { ProxyModelPlugin } from '../proxymodel/plugin.js';
|
||||
import { AuditCollector } from '../audit/collector.js';
|
||||
import { WireNameCodec, routeWithWireNames } from '../util/wire-names.js';
|
||||
|
||||
interface ProjectCacheEntry {
|
||||
router: McpRouter;
|
||||
@@ -45,6 +46,10 @@ export function registerProjectMcpEndpoint(app: FastifyInstance, mcpdClient: Mcp
|
||||
let resolvedUserName: string | null | undefined; // undefined = not yet resolved
|
||||
const projectCache = new Map<string, ProjectCacheEntry>();
|
||||
const sessions = new Map<string, SessionEntry>();
|
||||
// Wire-name codecs are keyed per project and OUTLIVE the router cache TTL:
|
||||
// a client may call a tool it listed minutes ago through a refreshed router,
|
||||
// and the mapping must still resolve.
|
||||
const wireCodecs = new Map<string, WireNameCodec>();
|
||||
|
||||
/** Resolve the mcplocal owner's userName once from /auth/me using mcplocal's own credentials. */
|
||||
async function ensureUserName(): Promise<string | null> {
|
||||
@@ -331,7 +336,18 @@ export function registerProjectMcpEndpoint(app: FastifyInstance, mcpdClient: Mcp
|
||||
});
|
||||
|
||||
const ctx = transport.sessionId ? { sessionId: transport.sessionId, correlationId } : { correlationId };
|
||||
const response = await router.route(message as unknown as JsonRpcRequest, ctx);
|
||||
// Wire-name translation happens HERE, at the client boundary, so the
|
||||
// router, plugins and audit all keep canonical `server/tool` names.
|
||||
let codec = wireCodecs.get(projectName);
|
||||
if (!codec) {
|
||||
codec = new WireNameCodec();
|
||||
wireCodecs.set(projectName, codec);
|
||||
}
|
||||
const response = await routeWithWireNames(
|
||||
codec,
|
||||
(req) => router.route(req, ctx),
|
||||
message as unknown as JsonRpcRequest,
|
||||
);
|
||||
|
||||
// Forward queued notifications BEFORE the response — the response send
|
||||
// closes the POST SSE stream, so notifications must go first.
|
||||
|
||||
126
src/mcplocal/src/util/wire-names.ts
Normal file
126
src/mcplocal/src/util/wire-names.ts
Normal file
@@ -0,0 +1,126 @@
|
||||
/**
|
||||
* Wire-safe tool-name codec for client-facing MCP endpoints.
|
||||
*
|
||||
* Internally the proxy namespaces tools as `server/tool`, and presentation
|
||||
* plugins add `favourite/<tool>`, `all/<server>/<tool>` and
|
||||
* `agent-<name>/chat`. A `/` is not a valid character in OpenAI-style
|
||||
* function names (`^[a-zA-Z0-9_.-]{1,64}$`), so hosts that forward MCP tool
|
||||
* names verbatim as LLM function names (LibreChat) depend on the model
|
||||
* faithfully echoing an illegal name. deepseek-v4-flash intermittently drops
|
||||
* the `server/` prefix; the host's registry lookup then fails and it reports
|
||||
* the tool's "MCP server is temporarily unavailable" while nothing is down
|
||||
* (the librechat fetch_content incident, 2026-08-25). Claude Code and the pi
|
||||
* extension dodge this only because they sanitize names client-side.
|
||||
*
|
||||
* The codec translates ONLY at the HTTP boundary:
|
||||
* - tools/list responses are rewritten to wire-safe names (`/` → `_`),
|
||||
* - tools/call requests are mapped back to the presented (internal) name
|
||||
* via an exact-match reverse map.
|
||||
*
|
||||
* Everything inside the proxy — routing maps, plugins, favourites config,
|
||||
* audit events — keeps the canonical names. Inbound names with no map entry
|
||||
* (legacy clients echoing slash names, virtual tools called before any
|
||||
* tools/list) pass through unchanged, so existing clients keep working.
|
||||
*
|
||||
* The maps live per project router (not per session) so a client that
|
||||
* reconnects mid-conversation still resolves names listed on its previous
|
||||
* session, as long as the process is alive. After a restart the first
|
||||
* tools/list (which every MCP client performs on initialize) repopulates them.
|
||||
*/
|
||||
import type { JsonRpcRequest, JsonRpcResponse } from '../types.js';
|
||||
|
||||
/** Replace every character that is invalid in an OpenAI-style function name. */
|
||||
export function sanitizeWireName(name: string): string {
|
||||
return name.replace(/[^A-Za-z0-9_.-]/g, '_');
|
||||
}
|
||||
|
||||
export class WireNameCodec {
|
||||
/** wire name → presented (internal) name */
|
||||
private toPresented = new Map<string, string>();
|
||||
/** presented (internal) name → wire name */
|
||||
private toWire = new Map<string, string>();
|
||||
|
||||
/**
|
||||
* Wire name for a presented tool name. Stable for the codec's lifetime.
|
||||
*
|
||||
* Collisions (two presented names sanitizing to the same string, or a
|
||||
* sanitized name shadowing a tool that already uses that exact name) get a
|
||||
* numeric suffix — first registration wins the plain name. This keeps the
|
||||
* reverse map unambiguous; order within one tools/list pass is stable, so
|
||||
* suffixes are deterministic in practice.
|
||||
*/
|
||||
encodeName(presented: string): string {
|
||||
const existing = this.toWire.get(presented);
|
||||
if (existing !== undefined) return existing;
|
||||
|
||||
const base = sanitizeWireName(presented);
|
||||
let wire = base;
|
||||
for (let i = 2; this.toPresented.has(wire) && this.toPresented.get(wire) !== presented; i++) {
|
||||
wire = `${base}_${String(i)}`;
|
||||
}
|
||||
if (wire !== base) {
|
||||
console.warn(`[wire-names] collision: '${presented}' presented as '${wire}' (base '${base}' taken by '${this.toPresented.get(base) ?? '?'}')`);
|
||||
}
|
||||
this.toWire.set(presented, wire);
|
||||
this.toPresented.set(wire, presented);
|
||||
return wire;
|
||||
}
|
||||
|
||||
/** The presented name a wire name maps to, or the input unchanged if unknown. */
|
||||
decodeName(wire: string): string {
|
||||
return this.toPresented.get(wire) ?? wire;
|
||||
}
|
||||
|
||||
/**
|
||||
* Rewrite a tools/list response's tool names to wire-safe names,
|
||||
* registering each mapping. Non-list responses and errors pass through.
|
||||
*/
|
||||
encodeToolsList(response: JsonRpcResponse): JsonRpcResponse {
|
||||
if (response.error !== undefined) return response;
|
||||
if (response.result === null || typeof response.result !== 'object') return response;
|
||||
const result = response.result as Record<string, unknown>;
|
||||
const tools: unknown = result['tools'];
|
||||
if (!Array.isArray(tools)) return response;
|
||||
|
||||
let changed = false;
|
||||
const encoded = (tools as unknown[]).map((tool) => {
|
||||
if (tool === null || typeof tool !== 'object' || typeof (tool as { name?: unknown }).name !== 'string') return tool;
|
||||
const presented = (tool as { name: string }).name;
|
||||
const wire = this.encodeName(presented);
|
||||
if (wire === presented) return tool;
|
||||
changed = true;
|
||||
return { ...(tool as Record<string, unknown>), name: wire };
|
||||
});
|
||||
|
||||
if (!changed) return response;
|
||||
return { ...response, result: { ...result, tools: encoded } };
|
||||
}
|
||||
|
||||
/**
|
||||
* Map a tools/call request's wire name back to the presented name.
|
||||
* Requests for unknown names (or without a name) pass through unchanged.
|
||||
*/
|
||||
decodeToolCall(request: JsonRpcRequest): JsonRpcRequest {
|
||||
const params = request.params;
|
||||
const name = params?.['name'];
|
||||
if (typeof name !== 'string') return request;
|
||||
const presented = this.toPresented.get(name);
|
||||
if (presented === undefined || presented === name) return request;
|
||||
return { ...request, params: { ...params, name: presented } };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Route one client request through `route` with wire-name translation:
|
||||
* decode the tool name on the way in (tools/call), encode tool names on the
|
||||
* way out (tools/list). Every other method is untouched.
|
||||
*/
|
||||
export async function routeWithWireNames(
|
||||
codec: WireNameCodec,
|
||||
route: (request: JsonRpcRequest) => Promise<JsonRpcResponse>,
|
||||
request: JsonRpcRequest,
|
||||
): Promise<JsonRpcResponse> {
|
||||
const inbound = request.method === 'tools/call' ? codec.decodeToolCall(request) : request;
|
||||
const response = await route(inbound);
|
||||
return request.method === 'tools/list' ? codec.encodeToolsList(response) : response;
|
||||
}
|
||||
100
src/mcplocal/tests/mcp-endpoint-wire-names.test.ts
Normal file
100
src/mcplocal/tests/mcp-endpoint-wire-names.test.ts
Normal file
@@ -0,0 +1,100 @@
|
||||
import { describe, it, expect, vi } from 'vitest';
|
||||
import Fastify from 'fastify';
|
||||
import { registerMcpEndpoint } from '../src/http/mcp-endpoint.js';
|
||||
import type { McpRouter } from '../src/router.js';
|
||||
import type { JsonRpcRequest, JsonRpcResponse } from '../src/types.js';
|
||||
|
||||
/**
|
||||
* End-to-end over the Streamable HTTP transport: the /mcp endpoint must serve
|
||||
* OpenAI-safe tool names on tools/list and map them back to the router's
|
||||
* canonical `server/tool` names on tools/call (the librechat fetch_content
|
||||
* incident, 2026-08-25).
|
||||
*/
|
||||
|
||||
function parseSse(body: string): JsonRpcResponse {
|
||||
const dataLine = body.split('\n').find((l) => l.startsWith('data: '));
|
||||
if (!dataLine) throw new Error(`no SSE data line in: ${body}`);
|
||||
return JSON.parse(dataLine.slice('data: '.length)) as JsonRpcResponse;
|
||||
}
|
||||
|
||||
describe('registerMcpEndpoint wire names', () => {
|
||||
it('lists wire-safe names and routes calls back to canonical names', async () => {
|
||||
const routed: JsonRpcRequest[] = [];
|
||||
const fakeRouter = {
|
||||
route: vi.fn(async (req: JsonRpcRequest): Promise<JsonRpcResponse> => {
|
||||
routed.push(req);
|
||||
switch (req.method) {
|
||||
case 'initialize':
|
||||
return {
|
||||
jsonrpc: '2.0',
|
||||
id: req.id,
|
||||
result: {
|
||||
protocolVersion: '2024-11-05',
|
||||
serverInfo: { name: 'test', version: '0' },
|
||||
capabilities: { tools: {} },
|
||||
},
|
||||
};
|
||||
case 'tools/list':
|
||||
return {
|
||||
jsonrpc: '2.0',
|
||||
id: req.id,
|
||||
result: { tools: [{ name: 'websearch/fetch_content', inputSchema: { type: 'object' } }] },
|
||||
};
|
||||
case 'tools/call':
|
||||
return {
|
||||
jsonrpc: '2.0',
|
||||
id: req.id,
|
||||
result: { content: [{ type: 'text', text: 'ok' }] },
|
||||
};
|
||||
default:
|
||||
return { jsonrpc: '2.0', id: req.id, result: {} };
|
||||
}
|
||||
}),
|
||||
} as unknown as McpRouter;
|
||||
|
||||
const app = Fastify();
|
||||
registerMcpEndpoint(app, fakeRouter);
|
||||
await app.ready();
|
||||
try {
|
||||
const headers = {
|
||||
'content-type': 'application/json',
|
||||
accept: 'application/json, text/event-stream',
|
||||
};
|
||||
|
||||
const init = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/mcp',
|
||||
headers,
|
||||
payload: { jsonrpc: '2.0', id: 1, method: 'initialize', params: { protocolVersion: '2024-11-05', capabilities: {}, clientInfo: { name: 't', version: '0' } } },
|
||||
});
|
||||
expect(init.statusCode).toBe(200);
|
||||
const sessionId = init.headers['mcp-session-id'] as string;
|
||||
expect(sessionId).toBeTruthy();
|
||||
const sessionHeaders = { ...headers, 'mcp-session-id': sessionId };
|
||||
|
||||
const list = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/mcp',
|
||||
headers: sessionHeaders,
|
||||
payload: { jsonrpc: '2.0', id: 2, method: 'tools/list' },
|
||||
});
|
||||
const listResponse = parseSse(list.body);
|
||||
const tools = (listResponse.result as { tools: Array<{ name: string }> }).tools;
|
||||
expect(tools.map((t) => t.name)).toEqual(['websearch_fetch_content']);
|
||||
|
||||
const call = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/mcp',
|
||||
headers: sessionHeaders,
|
||||
payload: { jsonrpc: '2.0', id: 3, method: 'tools/call', params: { name: 'websearch_fetch_content', arguments: { url: 'https://x' } } },
|
||||
});
|
||||
const callResponse = parseSse(call.body);
|
||||
expect(callResponse.error).toBeUndefined();
|
||||
|
||||
const routedCall = routed.find((r) => r.method === 'tools/call');
|
||||
expect(routedCall?.params?.['name']).toBe('websearch/fetch_content');
|
||||
} finally {
|
||||
await app.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
145
src/mcplocal/tests/project-mcp-endpoint-wire-names.test.ts
Normal file
145
src/mcplocal/tests/project-mcp-endpoint-wire-names.test.ts
Normal file
@@ -0,0 +1,145 @@
|
||||
import { describe, it, expect, vi } from 'vitest';
|
||||
import Fastify from 'fastify';
|
||||
import { registerProjectMcpEndpoint } from '../src/http/project-mcp-endpoint.js';
|
||||
import type { McpRouter } from '../src/router.js';
|
||||
import type { JsonRpcRequest, JsonRpcResponse } from '../src/types.js';
|
||||
|
||||
/**
|
||||
* End-to-end wire-name test on the endpoint the librechat incident actually
|
||||
* hit: /projects/:name/mcp. A fake `websearch` upstream serves a tool named
|
||||
* `fetch_content`; the client must see `websearch_fetch_content` in
|
||||
* tools/list, and calling that wire name must reach the upstream as plain
|
||||
* `fetch_content` (router canonical `websearch/fetch_content`, prefix
|
||||
* stripped on dispatch).
|
||||
*/
|
||||
|
||||
const upstreamRequests: JsonRpcRequest[] = [];
|
||||
|
||||
vi.mock('../src/discovery.js', () => ({
|
||||
refreshProjectUpstreams: vi.fn(async (router: McpRouter) => {
|
||||
router.addUpstream({
|
||||
name: 'websearch',
|
||||
send: async (req: JsonRpcRequest): Promise<JsonRpcResponse> => {
|
||||
upstreamRequests.push(req);
|
||||
if (req.method === 'tools/list') {
|
||||
return {
|
||||
jsonrpc: '2.0',
|
||||
id: req.id,
|
||||
result: { tools: [{ name: 'fetch_content', description: 'fetch a page', inputSchema: { type: 'object' } }] },
|
||||
};
|
||||
}
|
||||
if (req.method === 'tools/call') {
|
||||
return {
|
||||
jsonrpc: '2.0',
|
||||
id: req.id,
|
||||
result: { content: [{ type: 'text', text: 'page body' }] },
|
||||
};
|
||||
}
|
||||
return { jsonrpc: '2.0', id: req.id, result: {} };
|
||||
},
|
||||
close: async () => {},
|
||||
isAlive: () => true,
|
||||
});
|
||||
return ['websearch'];
|
||||
}),
|
||||
// gated: false → no gate plugin, the full catalog is served at initialize
|
||||
// (the chat-web configuration).
|
||||
fetchProjectLlmConfig: vi.fn(async () => ({ gated: false, llmProvider: 'none' })),
|
||||
}));
|
||||
|
||||
vi.mock('../src/http/config.js', async () => {
|
||||
const actual = await vi.importActual<typeof import('../src/http/config.js')>('../src/http/config.js');
|
||||
return { ...actual, loadProjectLlmOverride: vi.fn(() => undefined) };
|
||||
});
|
||||
|
||||
function mockMcpdClient() {
|
||||
const client: Record<string, unknown> = {
|
||||
baseUrl: 'http://test:3100',
|
||||
token: 'test-token',
|
||||
get: vi.fn(async () => []),
|
||||
post: vi.fn(async () => ({})),
|
||||
put: vi.fn(),
|
||||
delete: vi.fn(),
|
||||
forward: vi.fn(async () => ({ status: 200, body: [] })),
|
||||
withHeaders: vi.fn(),
|
||||
withToken: vi.fn(),
|
||||
withTimeout: vi.fn(),
|
||||
};
|
||||
(client.withHeaders as ReturnType<typeof vi.fn>).mockReturnValue(client);
|
||||
(client.withToken as ReturnType<typeof vi.fn>).mockReturnValue(client);
|
||||
(client.withTimeout as ReturnType<typeof vi.fn>).mockReturnValue(client);
|
||||
return client;
|
||||
}
|
||||
|
||||
function parseSse(body: string): JsonRpcResponse {
|
||||
const dataLine = body.split('\n').find((l) => l.startsWith('data: '));
|
||||
if (!dataLine) throw new Error(`no SSE data line in: ${body}`);
|
||||
return JSON.parse(dataLine.slice('data: '.length)) as JsonRpcResponse;
|
||||
}
|
||||
|
||||
describe('registerProjectMcpEndpoint wire names', () => {
|
||||
it('serves wire-safe names on the project endpoint and dispatches calls upstream', async () => {
|
||||
upstreamRequests.length = 0;
|
||||
const app = Fastify();
|
||||
registerProjectMcpEndpoint(app, mockMcpdClient() as never);
|
||||
await app.ready();
|
||||
try {
|
||||
const headers = {
|
||||
'content-type': 'application/json',
|
||||
accept: 'application/json, text/event-stream',
|
||||
};
|
||||
|
||||
const init = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/projects/chat-web/mcp',
|
||||
headers,
|
||||
payload: { jsonrpc: '2.0', id: 1, method: 'initialize', params: { protocolVersion: '2024-11-05', capabilities: {}, clientInfo: { name: 't', version: '0' } } },
|
||||
});
|
||||
expect(init.statusCode).toBe(200);
|
||||
const sessionId = init.headers['mcp-session-id'] as string;
|
||||
expect(sessionId).toBeTruthy();
|
||||
const sessionHeaders = { ...headers, 'mcp-session-id': sessionId };
|
||||
|
||||
const list = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/projects/chat-web/mcp',
|
||||
headers: sessionHeaders,
|
||||
payload: { jsonrpc: '2.0', id: 2, method: 'tools/list' },
|
||||
});
|
||||
const listResponse = parseSse(list.body);
|
||||
const tools = (listResponse.result as { tools: Array<{ name: string }> }).tools;
|
||||
const names = tools.map((t) => t.name);
|
||||
expect(names).toContain('websearch_fetch_content');
|
||||
// Every served name must be a valid OpenAI-style function name — the
|
||||
// invariant the librechat incident violated.
|
||||
for (const name of names) {
|
||||
expect(name).toMatch(/^[A-Za-z0-9_.-]+$/);
|
||||
}
|
||||
|
||||
const call = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/projects/chat-web/mcp',
|
||||
headers: sessionHeaders,
|
||||
payload: { jsonrpc: '2.0', id: 3, method: 'tools/call', params: { name: 'websearch_fetch_content', arguments: { url: 'https://x' } } },
|
||||
});
|
||||
const callResponse = parseSse(call.body);
|
||||
expect(callResponse.error).toBeUndefined();
|
||||
expect((callResponse.result as { content: Array<{ text: string }> }).content[0]?.text).toBe('page body');
|
||||
|
||||
// The upstream saw the bare tool name — namespace stripped, not mangled.
|
||||
const upstreamCall = upstreamRequests.find((r) => r.method === 'tools/call');
|
||||
expect(upstreamCall?.params?.['name']).toBe('fetch_content');
|
||||
|
||||
// Legacy clients echoing the canonical slash name keep working.
|
||||
const legacy = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/projects/chat-web/mcp',
|
||||
headers: sessionHeaders,
|
||||
payload: { jsonrpc: '2.0', id: 4, method: 'tools/call', params: { name: 'websearch/fetch_content', arguments: { url: 'https://x' } } },
|
||||
});
|
||||
expect(parseSse(legacy.body).error).toBeUndefined();
|
||||
} finally {
|
||||
await app.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
168
src/mcplocal/tests/wire-names.test.ts
Normal file
168
src/mcplocal/tests/wire-names.test.ts
Normal file
@@ -0,0 +1,168 @@
|
||||
import { describe, it, expect, vi } from 'vitest';
|
||||
import { sanitizeWireName, WireNameCodec, routeWithWireNames } from '../src/util/wire-names.js';
|
||||
import type { JsonRpcRequest, JsonRpcResponse } from '../src/types.js';
|
||||
|
||||
describe('sanitizeWireName', () => {
|
||||
it('replaces slashes with underscores', () => {
|
||||
expect(sanitizeWireName('websearch/fetch_content')).toBe('websearch_fetch_content');
|
||||
expect(sanitizeWireName('all/websearch/fetch_content')).toBe('all_websearch_fetch_content');
|
||||
});
|
||||
|
||||
it('keeps names that are already OpenAI-safe', () => {
|
||||
expect(sanitizeWireName('begin_session')).toBe('begin_session');
|
||||
expect(sanitizeWireName('my-grafana.tool')).toBe('my-grafana.tool');
|
||||
});
|
||||
|
||||
it('replaces every character outside [A-Za-z0-9_.-]', () => {
|
||||
expect(sanitizeWireName('a b:c/d')).toBe('a_b_c_d');
|
||||
});
|
||||
});
|
||||
|
||||
describe('WireNameCodec', () => {
|
||||
it('round-trips a namespaced tool name', () => {
|
||||
const codec = new WireNameCodec();
|
||||
const wire = codec.encodeName('websearch/fetch_content');
|
||||
expect(wire).toBe('websearch_fetch_content');
|
||||
expect(codec.decodeName(wire)).toBe('websearch/fetch_content');
|
||||
});
|
||||
|
||||
it('is stable across repeated encodes', () => {
|
||||
const codec = new WireNameCodec();
|
||||
expect(codec.encodeName('searxng/web_url_read')).toBe('searxng_web_url_read');
|
||||
expect(codec.encodeName('searxng/web_url_read')).toBe('searxng_web_url_read');
|
||||
});
|
||||
|
||||
it('passes unknown inbound names through unchanged', () => {
|
||||
const codec = new WireNameCodec();
|
||||
// Legacy client echoing the slash form, or a virtual tool never listed.
|
||||
expect(codec.decodeName('websearch/fetch_content')).toBe('websearch/fetch_content');
|
||||
expect(codec.decodeName('begin_session')).toBe('begin_session');
|
||||
});
|
||||
|
||||
it('suffixes on collision, first registration wins the plain name', () => {
|
||||
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
|
||||
try {
|
||||
const codec = new WireNameCodec();
|
||||
expect(codec.encodeName('foo_bar/baz')).toBe('foo_bar_baz');
|
||||
expect(codec.encodeName('foo/bar_baz')).toBe('foo_bar_baz_2');
|
||||
// Both decode back to their own presented names.
|
||||
expect(codec.decodeName('foo_bar_baz')).toBe('foo_bar/baz');
|
||||
expect(codec.decodeName('foo_bar_baz_2')).toBe('foo/bar_baz');
|
||||
// And stay stable.
|
||||
expect(codec.encodeName('foo/bar_baz')).toBe('foo_bar_baz_2');
|
||||
expect(warn).toHaveBeenCalledOnce();
|
||||
} finally {
|
||||
warn.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it('encodes tools/list responses and leaves other fields intact', () => {
|
||||
const codec = new WireNameCodec();
|
||||
const response: JsonRpcResponse = {
|
||||
jsonrpc: '2.0',
|
||||
id: 1,
|
||||
result: {
|
||||
tools: [
|
||||
{ name: 'websearch/fetch_content', description: 'fetch', inputSchema: { type: 'object' } },
|
||||
{ name: 'begin_session', description: 'gate' },
|
||||
],
|
||||
},
|
||||
};
|
||||
const encoded = codec.encodeToolsList(response);
|
||||
const tools = (encoded.result as { tools: Array<{ name: string; description?: string }> }).tools;
|
||||
expect(tools.map((t) => t.name)).toEqual(['websearch_fetch_content', 'begin_session']);
|
||||
expect(tools[0]?.description).toBe('fetch');
|
||||
// Original response object is not mutated.
|
||||
const originalTools = (response.result as { tools: Array<{ name: string }> }).tools;
|
||||
expect(originalTools[0]?.name).toBe('websearch/fetch_content');
|
||||
});
|
||||
|
||||
it('returns error and non-list responses unchanged', () => {
|
||||
const codec = new WireNameCodec();
|
||||
const err: JsonRpcResponse = { jsonrpc: '2.0', id: 1, error: { code: -32603, message: 'boom' } };
|
||||
expect(codec.encodeToolsList(err)).toBe(err);
|
||||
const other: JsonRpcResponse = { jsonrpc: '2.0', id: 1, result: { content: [] } };
|
||||
expect(codec.encodeToolsList(other)).toBe(other);
|
||||
});
|
||||
|
||||
it('decodes tools/call requests for known wire names only', () => {
|
||||
const codec = new WireNameCodec();
|
||||
codec.encodeName('websearch/fetch_content');
|
||||
|
||||
const known: JsonRpcRequest = {
|
||||
jsonrpc: '2.0',
|
||||
id: 2,
|
||||
method: 'tools/call',
|
||||
params: { name: 'websearch_fetch_content', arguments: { url: 'https://x' } },
|
||||
};
|
||||
const decoded = codec.decodeToolCall(known);
|
||||
expect(decoded.params?.['name']).toBe('websearch/fetch_content');
|
||||
expect(decoded.params?.['arguments']).toEqual({ url: 'https://x' });
|
||||
// Original request object is not mutated.
|
||||
expect(known.params?.['name']).toBe('websearch_fetch_content');
|
||||
|
||||
const unknown: JsonRpcRequest = {
|
||||
jsonrpc: '2.0',
|
||||
id: 3,
|
||||
method: 'tools/call',
|
||||
params: { name: 'not_listed', arguments: {} },
|
||||
};
|
||||
expect(codec.decodeToolCall(unknown)).toBe(unknown);
|
||||
});
|
||||
});
|
||||
|
||||
describe('routeWithWireNames', () => {
|
||||
const listResponse: JsonRpcResponse = {
|
||||
jsonrpc: '2.0',
|
||||
id: 1,
|
||||
result: { tools: [{ name: 'websearch/fetch_content' }, { name: 'searxng/web_url_read' }] },
|
||||
};
|
||||
|
||||
it('serves wire-safe names on tools/list and maps tools/call back', async () => {
|
||||
const codec = new WireNameCodec();
|
||||
const seen: JsonRpcRequest[] = [];
|
||||
const route = async (req: JsonRpcRequest): Promise<JsonRpcResponse> => {
|
||||
seen.push(req);
|
||||
if (req.method === 'tools/list') return listResponse;
|
||||
return { jsonrpc: '2.0', id: req.id, result: { content: [{ type: 'text', text: 'ok' }] } };
|
||||
};
|
||||
|
||||
const listed = await routeWithWireNames(codec, route, { jsonrpc: '2.0', id: 1, method: 'tools/list' });
|
||||
const names = (listed.result as { tools: Array<{ name: string }> }).tools.map((t) => t.name);
|
||||
expect(names).toEqual(['websearch_fetch_content', 'searxng_web_url_read']);
|
||||
|
||||
// The exact scenario from the librechat incident: the model echoes the
|
||||
// wire name; the router must receive the canonical name.
|
||||
await routeWithWireNames(codec, route, {
|
||||
jsonrpc: '2.0',
|
||||
id: 2,
|
||||
method: 'tools/call',
|
||||
params: { name: 'websearch_fetch_content', arguments: { url: 'https://x' } },
|
||||
});
|
||||
expect(seen[1]?.params?.['name']).toBe('websearch/fetch_content');
|
||||
});
|
||||
|
||||
it('keeps legacy slash-name calls working', async () => {
|
||||
const codec = new WireNameCodec();
|
||||
const seen: JsonRpcRequest[] = [];
|
||||
const route = async (req: JsonRpcRequest): Promise<JsonRpcResponse> => {
|
||||
seen.push(req);
|
||||
return { jsonrpc: '2.0', id: req.id, result: {} };
|
||||
};
|
||||
await routeWithWireNames(codec, route, {
|
||||
jsonrpc: '2.0',
|
||||
id: 1,
|
||||
method: 'tools/call',
|
||||
params: { name: 'websearch/fetch_content', arguments: {} },
|
||||
});
|
||||
expect(seen[0]?.params?.['name']).toBe('websearch/fetch_content');
|
||||
});
|
||||
|
||||
it('does not touch other methods', async () => {
|
||||
const codec = new WireNameCodec();
|
||||
const init: JsonRpcRequest = { jsonrpc: '2.0', id: 1, method: 'initialize', params: {} };
|
||||
const response: JsonRpcResponse = { jsonrpc: '2.0', id: 1, result: { protocolVersion: '2024-11-05' } };
|
||||
const out = await routeWithWireNames(codec, async () => response, init);
|
||||
expect(out).toBe(response);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user