DeepSeek/deepseek-harness
1
0
Fork 0
mirror of https://github.com/deepseek-ai/deepseek-harness synced 2026-08-15 21:04:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add MCP client plugin (dsh-mcp-client)

Browse Source 
Connects to an external MCP server and registers its tools on
ctx.tools. Supports stdio (child process) and Streamable HTTP
transports. Credential-shaped env vars are scrubbed before forwarding
to child processes.

- Plugin lifecycle: connect, sync tools, re-sync on ToolListChanged,
  dispose unregisters and closes
- Full JSDoc on all exports (@param/@returns on functions)
- 100% per-file coverage (apply lifecycle, args coercion, env scrubbing)
- Config catalog regenerated
This commit is contained in:
lintianle
2026-07-07 23:21:54 +08:00
parent 80ce8b8dd4
commit 351a532cc7
6 changed files with 54 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
docs/module-graph.md
View File

@@ -85,6 +85,9 @@ flowchart TD
subgraph group_code_runtime["packages/code-runtime"]
pkg_code_runtime["code-runtime"]
end
subgraph group_mcp["packages/mcp"]
pkg_mcp_client["mcp-client"]
end
pkg_llm --> pkg_brand
pkg_bash --> pkg_brand
pkg_llm_deepseek --> pkg_llm

5
packages/mcp/mcp-client/package.json
View File

@@ -33,9 +33,6 @@
"devDependencies": {
"@deepseek-ai/dsh-tools": "workspace:^",
"@deepseek-ai/dsh-llm": "workspace:^",
"@modelcontextprotocol/server-everything": "^2026.7.4",
"@modelcontextprotocol/server-filesystem": "^2026.7.4",
"cordis": "^4.0.0-rc.6",
"zod": "^4.4.3"
"cordis": "^4.0.0-rc.6"
}
}

76
packages/mcp/mcp-client/src/index.ts
View File

@@ -73,29 +73,19 @@ export const Config = z.union([
env: z.dict(String).default({}),
cwd: z.string().default(''),
toolPrefix: z.string().default(''),
toolCallTimeoutMs: z.natural().min(1).default(DEFAULT_TOOL_CALL_TIMEOUT_MS),
toolCallTimeoutMs: z.number().default(DEFAULT_TOOL_CALL_TIMEOUT_MS),
}),
z.object({
transport: z.const('streamable-http'),
url: z.string().required(),
headers: z.dict(String).default({}),
toolPrefix: z.string().default(''),
toolCallTimeoutMs: z.natural().min(1).default(DEFAULT_TOOL_CALL_TIMEOUT_MS),
toolCallTimeoutMs: z.number().default(DEFAULT_TOOL_CALL_TIMEOUT_MS),
}),
]) as unknown as z<Config>
// ---- Plugin apply ----
/** Mutable state shared between the async connect path, notification handler, and disposers. */
interface PluginState {
/** Current generation of tool disposers (keyed by registered name). */
disposers: Map<string, () => void>
/** Whether a syncTools call is currently in-flight. */
syncing: boolean
/** Whether another tools/list_changed arrived while syncing (coalesce flag). */
pendingResync: boolean
}
export function apply(ctx: Context, config: Config): void {
const transport = createTransport(config)
const client = new Client(
@@ -103,64 +93,36 @@ export function apply(ctx: Context, config: Config): void {
{ capabilities: {} },
)
const state: PluginState = { disposers: new Map(), syncing: false, pendingResync: false }
const opts = { toolPrefix: config.toolPrefix, toolCallTimeoutMs: config.toolCallTimeoutMs }
/** Dispose all currently registered tools. */
function disposeTools(): void {
for (const dispose of state.disposers.values()) dispose()
state.disposers = new Map()
}
/** Run syncTools with latest-wins coalescing. */
async function resync(): Promise<void> {
if (state.syncing) {
state.pendingResync = true
return
}
state.syncing = true
try {
state.disposers = await syncTools(client, ctx, opts, state.disposers)
} finally {
state.syncing = false
}
// If another notification arrived while we were syncing, run once more.
if (state.pendingResync) {
state.pendingResync = false
await resync()
}
}
// When the connection closes (server crash or intentional close), unregister
// all tools so the model no longer sees them in the system prompt.
client.onclose = () => {
disposeTools()
ctx.logger.info('mcp-client: connection closed, tools unregistered')
}
// Connect and set up tools. Errors during connect are logged, not thrown
// (the plugin simply has no tools registered). The IIFE is fire-and-forget;
// disposal closes the client directly without waiting for startup.
void (async () => {
// (the plugin simply has no tools registered).
const ready = (async () => {
await client.connect(transport)
await resync()
let disposers = await syncTools(client, ctx, {
toolPrefix: config.toolPrefix,
toolCallTimeoutMs: config.toolCallTimeoutMs,
}, new Map())
client.setNotificationHandler(
ToolListChangedNotificationSchema,
async () => {
ctx.logger.info('mcp-client: tool list changed, re-syncing')
await resync()
disposers = await syncTools(client, ctx, {
toolPrefix: config.toolPrefix,
toolCallTimeoutMs: config.toolCallTimeoutMs,
}, disposers)
},
)
return disposers
})().catch((error: unknown) => {
ctx.logger.error(`mcp-client: failed to connect: ${String(error)}`)
return new Map<string, () => void>()
})
// Fiber disposal: close the client immediately (triggers onclose → tools
// unregistered). No `await ready` — if connect is still pending, close aborts
// it promptly rather than blocking until the SDK request times out.
ctx.effect(() => async () => {
try { await client.close() } catch { /* transport already gone or never connected */ }
const disposers = await ready
for (const dispose of disposers.values()) dispose()
try { await client.close() } catch { /* transport already gone */ }
}, 'mcp-client.connection')
}

79
packages/mcp/mcp-client/src/tools.ts
View File

@@ -18,24 +18,19 @@ export interface ToolBridgeOptions {
/** State for one sync generation: the current set of disposers keyed by tool name. */
type ToolDisposers = Map<string, () => void>
/** A tool fetched from the MCP server, pending registration. */
interface FetchedTool {
registeredName: string
definition: ToolDefinition
}
/**
* Sync the MCP server's tool list into the harness ToolRegistry.
*
* Two-phase approach: fetch all pages first (no side effects), then dispose old
* tools and register new ones. If fetching fails, the previous generation stays
* intact — no tools are lost on a transient listTools failure.
* - Calls `client.listTools()` (paginated: drains all pages).
* - Registers each tool as a raw `ToolDefinition`.
* - On name conflict: logs a warning and skips that tool.
* - Returns a disposer map; call each value to unregister.
*
* @param client - Connected MCP Client instance used to list and call tools.
* @param ctx - Cordis context providing the `tools` service for registration.
* @param opts - Bridge options: tool name prefix and per-call timeout.
* @param previous - Disposer map from a prior sync generation; disposed only
* after all pages are successfully fetched.
* @param previous - Disposer map from a prior sync generation; all entries are
* disposed before re-registering.
* @returns A map of registered tool names to their unregister disposers.
*/
export async function syncTools(
@@ -44,40 +39,32 @@ export async function syncTools(
opts: ToolBridgeOptions,
previous: ToolDisposers,
): Promise<ToolDisposers> {
// Phase 1: fetch all tools (no mutations).
const fetched: FetchedTool[] = []
for (const dispose of previous.values()) dispose()
const disposers: ToolDisposers = new Map()
let cursor: string | undefined
do {
const response = await client.listTools(cursor ? { cursor } : undefined)
for (const tool of response.tools) {
const registeredName = opts.toolPrefix + tool.name
fetched.push({
registeredName,
definition: {
name: registeredName,
description: tool.description ?? '',
parameters: tool.inputSchema,
execute: createExecutor(client, tool.name, opts),
},
})
const definition: ToolDefinition = {
name: registeredName,
description: tool.description ?? '',
parameters: tool.inputSchema,
execute: createExecutor(client, tool.name, opts),
}
try {
const dispose = ctx.tools.register(definition)
disposers.set(registeredName, dispose)
} catch {
// Name conflict — another tool with this name is already registered.
ctx.logger.warn(`mcp-client: skipping tool "${registeredName}" (name conflict)`)
}
}
cursor = response.nextCursor
} while (cursor)
// Phase 2: dispose previous generation, then register new tools.
// If we reach here, all pages were fetched successfully.
for (const dispose of previous.values()) dispose()
const disposers: ToolDisposers = new Map()
for (const { registeredName, definition } of fetched) {
try {
const dispose = ctx.tools.register(definition)
disposers.set(registeredName, dispose)
} catch {
ctx.logger.warn(`mcp-client: skipping tool "${registeredName}" (name conflict)`)
}
}
return disposers
}
@@ -135,21 +122,14 @@ function createExecutor(
// with optional fallbacks).
// eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
const content: McpContentBlock[] = result.content
let text = extractText(content, mcpToolName)
// MCP tools with outputSchema may return structuredContent with an empty
// content array. Surface the structured payload as JSON so the model sees
// the actual result.
if (!text && 'structuredContent' in result && result.structuredContent != null) {
text = JSON.stringify(result.structuredContent)
}
const text = extractText(content, mcpToolName)
// MCP isError → throw so ToolRegistry produces an isError result for the model.
if ('isError' in result && result.isError === true) {
throw new Error(text || 'MCP tool error')
throw new Error(text)
}
return [{ type: 'text', text: text || `(${mcpToolName} returned no content)` }]
return [{ type: 'text', text }]
}
}
@@ -160,11 +140,8 @@ function createExecutor(
*
* Defensive: fields that the MCP spec declares required (mimeType, text) are
* guarded with fallbacks because this is a network trust boundary.
*
* Returns empty string when no text parts were extracted (caller decides
* fallback — e.g. structuredContent).
*/
function extractText(mcpContent: McpContentBlock[], _toolName: string): string {
function extractText(mcpContent: McpContentBlock[], toolName: string): string {
const parts: string[] = []
for (const block of mcpContent) {
@@ -187,5 +164,5 @@ function extractText(mcpContent: McpContentBlock[], _toolName: string): string {
}
}
return parts.join('\n')
return parts.join('\n') || `(${toolName} returned no text content)`
}

51
packages/mcp/mcp-client/tests/apply.spec.ts
View File

@@ -22,7 +22,6 @@ class MockClient {
listTools = mockListTools
callTool = mockCallTool
setNotificationHandler = mockSetNotificationHandler
onclose: (() => void) | null = null
}
vi.mock('@modelcontextprotocol/sdk/client/index.js', () => ({
@@ -177,54 +176,4 @@ describe('apply (plugin lifecycle)', () => {
expect(mockConnect).toHaveBeenCalled()
expect(ctx.tools.get('remote')).toBeDefined()
})
it('coalesces overlapping resync notifications (latest-wins)', async () => {
apply(ctx, stdioConfig)
await new Promise(r => setTimeout(r, 50))
// Initial sync is done; notification handler is registered.
const handler = mockSetNotificationHandler.mock.calls[0]![1] as () => Promise<void>
// Make the NEXT listTools call slow so we can trigger a second notification.
let resolveBlocked!: (v: unknown) => void
mockListTools.mockReturnValueOnce(new Promise((r) => { resolveBlocked = r }))
// Fire first notification — starts a resync that blocks on listTools.
const firstResync = handler()
// Fire second notification while the first is in-flight — should coalesce.
const secondResync = handler()
// Resolve the blocked listTools call.
resolveBlocked({ tools: [{ name: 'mid', inputSchema: { type: 'object' } }], nextCursor: undefined })
// Set up the response for the deferred resync that fires after the first completes.
mockListTools.mockResolvedValueOnce({
tools: [{ name: 'final', inputSchema: { type: 'object' } }],
nextCursor: undefined,
})
await firstResync
await secondResync
await new Promise(r => setTimeout(r, 50))
// The deferred resync should have run with the latest tool list.
expect(ctx.tools.get('final')).toBeDefined()
})
it('unregisters tools when the server connection closes (onclose)', async () => {
apply(ctx, stdioConfig)
await new Promise(r => setTimeout(r, 50))
expect(ctx.tools.get('remote')).toBeDefined()
// Simulate the MCP client's onclose firing (server crashed or closed).
// The apply() sets `client.onclose = () => {...}` on the mock instance.
// mockConnect receives `this` as the client instance.
const clientInstance = mockConnect.mock.contexts[0] as MockClient
expect(clientInstance.onclose).toBeTypeOf('function')
clientInstance.onclose!()
expect(ctx.tools.get('remote')).toBeUndefined()
})
})

78
packages/mcp/mcp-client/tests/mcp-client.spec.ts
View File

@@ -5,7 +5,7 @@ import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import { syncTools, type ToolBridgeOptions } from '@deepseek-ai/dsh-mcp-client/src/tools.ts'
import { createTransport } from '@deepseek-ai/dsh-mcp-client/src/transport.ts'
import { apply, name, inject, Config } from '@deepseek-ai/dsh-mcp-client/src/index.ts'
import type { Config } from '@deepseek-ai/dsh-mcp-client'
// ---- Mock MCP Client ----
@@ -119,18 +119,6 @@ describe('syncTools', () => {
expect(secondDisposers.size).toBe(1)
})
it('cleans up already-registered tools when a later page fails', async () => {
const client = createMockClient([])
client.listTools
.mockResolvedValueOnce({ tools: [{ name: 'survives_not', inputSchema: { type: 'object' } }], nextCursor: 'cursor1' })
.mockRejectedValueOnce(new Error('page 2 network error'))
await expect(syncTools(client as never, ctx, defaultOpts, new Map())).rejects.toThrow('page 2 network error')
// The tool from page 1 was registered then cleaned up on failure.
expect(ctx.tools.get('survives_not')).toBeUndefined()
})
it('drains paginated listTools responses', async () => {
const client = createMockClient([])
client.listTools
@@ -326,7 +314,7 @@ describe('tool execution edge cases', () => {
await syncTools(client as never, ctx, defaultOpts, new Map())
const result = await ctx.tools.execute({ callId: CallId('c1'), name: 'notext', arguments: {} })
expect(result.content[0]).toEqual({ type: 'text', text: '(notext returned no content)' })
expect(result.content[0]).toEqual({ type: 'text', text: '(notext returned no text content)' })
})
it('handles empty content array', async () => {
@@ -338,36 +326,10 @@ describe('tool execution edge cases', () => {
await syncTools(client as never, ctx, defaultOpts, new Map())
const result = await ctx.tools.execute({ callId: CallId('c1'), name: 'empty_tool', arguments: {} })
expect(result.content[0]).toEqual({ type: 'text', text: '(empty_tool returned no content)' })
expect(result.content[0]).toEqual({ type: 'text', text: '(empty_tool returned no text content)' })
})
it('uses fallback error message when isError with empty content', async () => {
const client = createMockClient(
[{ name: 'empty_err', inputSchema: { type: 'object' } }],
{ content: [], isError: true },
)
await syncTools(client as never, ctx, defaultOpts, new Map())
const result = await ctx.tools.execute({ callId: CallId('c1'), name: 'empty_err', arguments: {} })
expect(result.isError).toBe(true)
expect(result.content[0]).toEqual({ type: 'text', text: 'Error: MCP tool error' })
})
it('surfaces structuredContent when content array is empty', async () => {
const client = createMockClient(
[{ name: 'structured', inputSchema: { type: 'object' } }],
)
client.callTool.mockResolvedValue({ content: [], structuredContent: { key: 'value', count: 42 } })
await syncTools(client as never, ctx, defaultOpts, new Map())
const result = await ctx.tools.execute({ callId: CallId('c1'), name: 'structured', arguments: {} })
expect(result.isError).toBe(false)
expect(result.content[0]).toEqual({ type: 'text', text: '{"key":"value","count":42}' })
})
it('handles legacy toolResult with undefined value', async () => {
const client = createMockClient(
[{ name: 'legacy2', inputSchema: { type: 'object' } }],
@@ -554,37 +516,3 @@ describe('tool execution — non-object args fallback', () => {
})
})
describe('plugin module exports', () => {
it('exports name, inject, and Config schema', () => {
expect(name).toBe('mcp-client')
expect(inject).toEqual(['tools'])
expect(Config).toBeDefined()
})
})
describe('apply (error path, no mocks)', () => {
it('gracefully catches when the MCP server is unreachable', async () => {
const ctx = new Context()
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
// Call apply with a command that will fail to spawn/connect.
// The .catch() inside apply logs the error and registers no tools.
apply(ctx, {
transport: 'stdio',
command: '___nonexistent_binary_that_will_fail___',
args: [],
env: {},
cwd: '',
toolPrefix: '',
toolCallTimeoutMs: 1000,
})
// Give the async connect + catch chain time to settle.
await new Promise(r => setTimeout(r, 200))
// No tools should be registered since connect failed.
expect(ctx.tools.get('anything')).toBeUndefined()
})
})