| import assert from 'node:assert/strict' |
| import { setTimeout as delay } from 'node:timers/promises' |
| import test from 'node:test' |
| import { Context } from '@deepseek-ai/cordis' |
| import SystemPrompt from '@deepseek-ai/dsh-system-prompt' |
| import ToolRuntime, { defineTool } from '@deepseek-ai/dsh-tools' |
| import { Client } from '@modelcontextprotocol/sdk/client/index.js' |
| import { InMemoryTransport } from '@modelcontextprotocol/sdk/inMemory.js' |
| import { ToolListChangedNotificationSchema } from '@modelcontextprotocol/sdk/types.js' |
| import plugin, { createMcpServer } from '../src/index.mjs' |
|
|
| async function runtime(t) { |
| const ctx = new Context() |
| await ctx.plugin(SystemPrompt, {}) |
| await ctx.plugin(ToolRuntime, { mode: 'native' }) |
| t.after(() => ctx.fiber.dispose()) |
| return ctx |
| } |
|
|
| function register(ctx, name = 'echo', options = {}) { |
| return ctx.tools.register(defineTool({ |
| name, |
| description: '测试文本工具', |
| parameters: { text: { type: 'string', required: true } }, |
| output: { |
| schema: { type: 'string' }, |
| render: (_args, value) => [{ type: 'text', text: value }], |
| }, |
| async execute({ text }) { return text }, |
| ...options, |
| })) |
| } |
|
|
| async function connect(ctx, t) { |
| const exported = createMcpServer(ctx) |
| const client = new Client({ name: 'message-mcp-test', version: '0.1.0' }) |
| const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair() |
| await exported.server.connect(serverTransport) |
| await client.connect(clientTransport) |
| t.after(async () => { |
| await exported.close() |
| await client.close() |
| }) |
| return { ...exported, client } |
| } |
|
|
| test('从真实工具注册表列出 schema 并通知新增与移除', async (t) => { |
| const ctx = await runtime(t) |
| register(ctx) |
| const { client } = await connect(ctx, t) |
| assert.deepEqual((await client.listTools()).tools, [{ |
| name: 'echo', |
| description: '测试文本工具', |
| inputSchema: { |
| type: 'object', |
| properties: { text: { type: 'string' } }, |
| required: ['text'], |
| }, |
| }]) |
| const changed = Promise.withResolvers() |
| client.setNotificationHandler(ToolListChangedNotificationSchema, () => changed.resolve()) |
| const unregister = register(ctx, 'second') |
| await changed.promise |
| assert.equal((await client.listTools()).tools.length, 2) |
| unregister() |
| assert.deepEqual((await client.listTools()).tools.map((tool) => tool.name), ['echo']) |
| }) |
|
|
| test('调用通过真实前置、执行包装、后置与最终通知策略且不泄漏原始值', async (t) => { |
| const ctx = await runtime(t) |
| const order = [] |
| register(ctx, 'echo', { |
| async execute() { order.push('body'); return '原始私有值' }, |
| }) |
| ctx.on('tools/pre-execute', async (_exec, next) => { order.push('pre'); return next() }) |
| ctx.tools.guard(() => { order.push('guard') }) |
| ctx.on('tools/execute', async (_exec, next) => { order.push('around'); return next() }) |
| ctx.on('tools/post-execute', async () => { |
| order.push('post') |
| return { kind: 'accept', content: [{ type: 'text', text: '已脱敏' }] } |
| }) |
| ctx.on('tools/result', () => { order.push('result') }) |
| const { client } = await connect(ctx, t) |
| const result = await client.callTool({ name: 'echo', arguments: { text: '输入' } }) |
| assert.deepEqual(order, ['pre', 'guard', 'around', 'body', 'post', 'result']) |
| assert.deepEqual(result, { isError: false, content: [{ type: 'text', text: '已脱敏' }] }) |
| assert.equal(JSON.stringify(result).includes('原始私有值'), false) |
| }) |
|
|
| test('guard 拒绝与无 agent 审批不会执行工具', async (t) => { |
| const ctx = await runtime(t) |
| let executed = 0 |
| register(ctx, 'denied', { async execute() { executed++; return '不应执行' } }) |
| register(ctx, 'ask', { async execute() { executed++; return '不应执行' } }) |
| ctx.tools.guard((exec) => exec.name === 'denied' ? '测试禁止' : undefined) |
| ctx.on('tools/pre-execute', async (exec, next) => exec.name === 'ask' |
| ? { kind: 'ask', reason: '需要人工批准' } |
| : next()) |
| const { client } = await connect(ctx, t) |
| for (const name of ['denied', 'ask']) { |
| const result = await client.callTool({ name, arguments: { text: 'x' } }) |
| assert.equal(result.isError, true) |
| } |
| assert.equal(executed, 0) |
| }) |
|
|
| test('参数错误和未知工具保留工具错误响应，之后仍可正常调用', async (t) => { |
| const ctx = await runtime(t) |
| register(ctx) |
| const { client } = await connect(ctx, t) |
| assert.equal((await client.callTool({ name: 'echo', arguments: { text: 12 } })).isError, true) |
| assert.equal((await client.callTool({ name: 'missing', arguments: {} })).isError, true) |
| assert.equal((await client.callTool({ name: 'echo', arguments: { text: '正常' } })).isError, false) |
| }) |
|
|
| test('非文本结果仅给占位，不序列化内部内容', async (t) => { |
| const ctx = await runtime(t) |
| register(ctx, 'echo', { |
| output: { schema: { type: 'string' }, render: () => [{ type: 'reasoning', text: '内部信息' }] }, |
| }) |
| const { client } = await connect(ctx, t) |
| const result = await client.callTool({ name: 'echo', arguments: { text: 'x' } }) |
| assert.equal(JSON.stringify(result).includes('内部信息'), false) |
| assert.match(result.content[0].text, /非文本/) |
| }) |
|
|
| test('并发 MCP 请求串行执行', async (t) => { |
| const ctx = await runtime(t) |
| let active = 0 |
| let maximum = 0 |
| register(ctx, 'echo', { |
| async execute({ text }) { |
| maximum = Math.max(maximum, ++active) |
| await delay(10) |
| active-- |
| return text |
| }, |
| }) |
| const { client } = await connect(ctx, t) |
| const results = await Promise.all(['a', 'b', 'c'].map((text) => client.callTool({ name: 'echo', arguments: { text } }))) |
| assert.equal(maximum, 1) |
| assert.deepEqual(results.map((result) => result.content[0].text), ['a', 'b', 'c']) |
| }) |
|
|
| test('同一 Cordis 根下不同上下文的 MCP Server 共用独占队列', async (t) => { |
| const ctx = await runtime(t) |
| let active = 0 |
| let maximum = 0 |
| register(ctx, 'echo', { |
| async execute({ text }) { |
| maximum = Math.max(maximum, ++active) |
| await delay(10) |
| active-- |
| return text |
| }, |
| }) |
| const first = await connect(ctx, t) |
| const second = await connect(ctx.extend(), t) |
| await Promise.all([first, second].map(({ client }) => client.callTool({ name: 'echo', arguments: { text: '独占' } }))) |
| assert.equal(maximum, 1) |
| }) |
|
|
| test('客户端取消传入真实工具信号且关闭等待工具清理', async (t) => { |
| const ctx = await runtime(t) |
| const started = Promise.withResolvers() |
| const cleaned = Promise.withResolvers() |
| let finalized = false |
| register(ctx, 'wait', { |
| async execute(_args, exec) { |
| started.resolve() |
| await new Promise((resolve) => exec.signal.addEventListener('abort', resolve, { once: true })) |
| await delay(10) |
| finalized = true |
| cleaned.resolve() |
| return '已清理' |
| }, |
| }) |
| const { client, close } = await connect(ctx, t) |
| const controller = new AbortController() |
| const request = client.callTool({ name: 'wait', arguments: { text: 'x' } }, undefined, { signal: controller.signal }) |
| const rejected = assert.rejects(request) |
| await started.promise |
| controller.abort() |
| await rejected |
| await close() |
| await cleaned.promise |
| assert.equal(finalized, true) |
| }) |
|
|
| test('关闭会取消执行和排队请求，且可重复关闭', async (t) => { |
| const ctx = await runtime(t) |
| const started = Promise.withResolvers() |
| let bodies = 0 |
| let finalized = false |
| register(ctx, 'wait', { |
| async execute(_args, exec) { |
| bodies++ |
| started.resolve() |
| await new Promise((resolve) => exec.signal.addEventListener('abort', resolve, { once: true })) |
| await delay(10) |
| finalized = true |
| return '已清理' |
| }, |
| }) |
| const { client, close } = await connect(ctx, t) |
| const first = client.callTool({ name: 'wait', arguments: { text: 'a' } }) |
| const second = client.callTool({ name: 'wait', arguments: { text: 'b' } }) |
| const requests = Promise.allSettled([first, second]) |
| await started.promise |
| await close() |
| await requests |
| await close() |
| assert.equal(bodies, 1) |
| assert.equal(finalized, true) |
| }) |
|
|
| test('Cordis 默认不接管 stdin，错误传输配置不能启动', async (t) => { |
| const ctx = await runtime(t) |
| const before = process.stdin.listenerCount('data') |
| const fiber = await ctx.plugin(plugin, {}) |
| assert.equal(process.stdin.listenerCount('data'), before) |
| await fiber.dispose() |
| |
| const result = await plugin.Config['~standard'].validate({ transport: 'unsupported' }) |
| assert.ok(result.issues) |
| }) |
|
|
| test('Cordis 卸载清理 stdio 监听且允许再次挂载', async (t) => { |
| const ctx = await runtime(t) |
| const beforeData = process.stdin.listenerCount('data') |
| const beforeEnd = process.stdin.listenerCount('end') |
| for (let index = 0; index < 2; index++) { |
| const fiber = await ctx.plugin(plugin, { transport: 'stdio' }) |
| assert.deepEqual(ctx.get('messageMcpExport'), { transport: 'stdio' }) |
| assert.equal(process.stdin.listenerCount('data'), beforeData + 1) |
| assert.equal(process.stdin.listenerCount('end'), beforeEnd + 1) |
| await fiber.dispose() |
| assert.equal(ctx.get('messageMcpExport'), undefined) |
| assert.equal(process.stdin.listenerCount('data'), beforeData) |
| assert.equal(process.stdin.listenerCount('end'), beforeEnd) |
| } |
| }) |
|
|
| test('执行与schema异常经过最终秘密脱敏边界', async (t) => { |
| const secret = 'fixture-only-error-secret' |
| let failList = false |
| const ctx = { |
| get: key => key === 'messageSecretRedactor' ? text => text.split(secret).join('[REDACTED]') : undefined, |
| on: () => () => {}, |
| tools: { |
| schemas: () => { if (failList) throw new Error(secret); return [{ name: 'failure', description: 'error fixture', parameters: { type: 'object', properties: {} } }] }, |
| execute: async () => { throw new Error(secret) }, |
| }, |
| } |
| const { client } = await connect(ctx, t) |
| await assert.rejects(client.callTool({ name: 'failure', arguments: {} }), error => error.message.includes('[REDACTED]') && !error.message.includes(secret)) |
| failList = true |
| await assert.rejects(client.listTools(), error => error.message.includes('[REDACTED]') && !error.message.includes(secret)) |
| }); |
|
|
| test('短环境值脱敏不会破坏JSON Schema的数值约束', async (t) => { |
| const ctx = { |
| get: () => text => text.split('1').join('[REDACTED]'), on: () => () => {}, |
| tools: { |
| schemas: () => [{ name: 'short', description: 'text 1', parameters: { type: 'object', properties: { number: { type: 'number', minimum: 1 } } } }], |
| execute: async () => ({ isError: false, content: [{ type: 'text', text: '1' }] }), |
| }, |
| } |
| const { client } = await connect(ctx, t) |
| assert.equal((await client.listTools()).tools[0].inputSchema.properties.number.minimum, 1) |
| assert.equal((await client.callTool({ name: 'short' })).content[0].text, '[REDACTED]') |
| }); |
|
|
