diff --git a/.changeset/tasks-extension.md b/.changeset/tasks-extension.md new file mode 100644 index 0000000000..b6b80b45ff --- /dev/null +++ b/.changeset/tasks-extension.md @@ -0,0 +1,13 @@ +--- +'@modelcontextprotocol/core-internal': minor +'@modelcontextprotocol/client': minor +'@modelcontextprotocol/server': minor +--- + +The MCP Tasks extension (`io.modelcontextprotocol/tasks`) as a pair of extensions. + +`@modelcontextprotocol/server/ext/tasks`: `new TasksExtension(store)` in `ServerOptions.extensions` advertises the capability, serves `tasks/get`, `tasks/update` and `tasks/cancel`, gates task handles on the client capability (`-32021`), and offers `tasks.create(ctx)` for a tool handler to answer with a task handle. `TaskStore` (create / get / update / cancel over JSON) is the interface a server implements over its own state and execution; `InMemoryTaskStore` is the in-process reference with a writer `handle` for reporting status, requesting input, and settling. + +`@modelcontextprotocol/client/ext/tasks`: `new TasksClientExtension()` in `ClientOptions.extensions` declares the capability on every request, accepts task handles on `tools/call`, and wraps the extension's methods: `callTool` (a task handle or the plain result), `get`, `update`, `cancel`, and `waitFor`, which polls at the server's suggested interval to a terminal snapshot. + +Wire types and zod schemas for the extension's 2026-07-28 schema live at `@modelcontextprotocol/core-internal/ext/tasks` and are re-exported from both subpaths. diff --git a/docs/.vitepress/nav.ts b/docs/.vitepress/nav.ts index 7239b750d1..b73516d9eb 100644 --- a/docs/.vitepress/nav.ts +++ b/docs/.vitepress/nav.ts @@ -28,6 +28,7 @@ export const guideSidebar: DefaultTheme.SidebarItem[] = [ { text: 'Elicitation', link: '/servers/elicitation' }, { text: 'Sampling (sunset)', link: '/servers/sampling' }, { text: 'Input required', link: '/servers/input-required' }, + { text: 'Tasks (extension)', link: '/servers/tasks' }, { text: 'Notifications', link: '/servers/notifications' }, { text: 'Errors', link: '/servers/errors' } ] @@ -54,6 +55,7 @@ export const guideSidebar: DefaultTheme.SidebarItem[] = [ { text: 'Handle server requests', link: '/clients/server-requests' }, { text: 'Roots (sunset)', link: '/clients/roots' }, { text: 'Subscriptions', link: '/clients/subscriptions' }, + { text: 'Tasks (extension)', link: '/clients/tasks' }, { text: 'OAuth', link: '/clients/oauth' }, { text: 'Machine auth', link: '/clients/machine-auth' }, { text: 'Middleware', link: '/clients/middleware' }, diff --git a/docs/clients/tasks.md b/docs/clients/tasks.md new file mode 100644 index 0000000000..020c0171ca --- /dev/null +++ b/docs/clients/tasks.md @@ -0,0 +1,81 @@ +--- +shape: how-to +--- + +# Tasks (extension) + +The [MCP Tasks extension](https://github.com/modelcontextprotocol/ext-tasks) (`io.modelcontextprotocol/tasks`) lets a server answer a tool call with a **task handle** instead of blocking: you poll `tasks/get`, answer `tasks/update`, and stop with `tasks/cancel`. `@modelcontextprotocol/client/ext/tasks` is the client side of that wire, as a [client extension](../advanced/extensions.md). The server side is [Tasks (extension)](../servers/tasks.md). + +## Install the extension + +One extension instance serves one client; pass it at construction. + +```ts +import { Client } from '@modelcontextprotocol/client'; +import { TasksClientExtension } from '@modelcontextprotocol/client/ext/tasks'; + +const tasks = new TasksClientExtension(); +const client = new Client({ name: 'report-client', version: '1.0.0' }, { versionNegotiation: { mode: 'auto' }, extensions: [tasks] }); +await client.connect(transport); +``` + +The client declares `io.modelcontextprotocol/tasks` in the capabilities envelope of every request — the server refuses task handles to clients that do not — and accepts `resultType: "task"` on `tools/call`, which the plain `client.callTool` cannot describe. + +## Call a tool that may become a task + +`tasks.callTool` returns a discriminated outcome: a task handle to follow, or the ordinary result for tools that answered synchronously. + +```ts +const outcome = await tasks.callTool({ name: 'send_report', arguments: { to: 'ops' } }); +if (outcome.kind === 'result') { + console.log(outcome.result.content); +} else { + console.log(outcome.task.taskId, outcome.task.status, outcome.task.pollIntervalMs); +} +``` + +## Poll to the end + +`waitFor` polls `tasks/get` at the server's suggested `pollIntervalMs` until the task is terminal. Every snapshot passes through `onUpdate`, which is where an `input_required` task gets answered. + +```ts +const done = await tasks.waitFor(outcome.task.taskId, { + signal: controller.signal, + onUpdate: async task => { + if (task.status === 'input_required') { + const answers = await askUser(task.inputRequests); + await tasks.update(task.taskId, answers); + } + } +}); + +switch (done.status) { + case 'completed': + console.log(done.result); + break; + case 'failed': + console.error(done.error); + break; + case 'cancelled': + break; +} +``` + +Aborting the signal stops polling and nothing else. Cancelling the task is a separate, cooperative call: `await tasks.cancel(taskId)` resolves on acknowledgement, and the task may still settle `completed` or `failed`. + +## The raw methods + +| Method | Wire | Returns | +| -------------------------------------- | -------------- | --------------------------------------------------------------------- | +| `tasks.get(taskId)` | `tasks/get` | The `DetailedTask` snapshot: result, error, or input requests inlined | +| `tasks.update(taskId, inputResponses)` | `tasks/update` | Resolves on acknowledgement; partial answers are accepted | +| `tasks.cancel(taskId)` | `tasks/cancel` | Resolves on acknowledgement | + +Each takes the usual `RequestOptions` (`signal`, `timeout`) as a last argument. + +## Recap + +- `new TasksClientExtension()` in `ClientOptions.extensions`; one instance per client. +- `tasks.callTool(params)` yields `{ kind: 'task', task }` or `{ kind: 'result', result }`. +- `tasks.waitFor(taskId, { onUpdate, signal })` polls at the server's interval to a terminal snapshot. +- `get`, `update`, `cancel` are the extension's three methods, one to one. diff --git a/docs/servers/tasks.md b/docs/servers/tasks.md new file mode 100644 index 0000000000..1d6eea2954 --- /dev/null +++ b/docs/servers/tasks.md @@ -0,0 +1,78 @@ +--- +shape: how-to +--- + +# Tasks (extension) + +The [MCP Tasks extension](https://github.com/modelcontextprotocol/ext-tasks) (`io.modelcontextprotocol/tasks`) lets a tool answer with a **task handle** instead of blocking on work that takes minutes: the client polls `tasks/get`, answers `tasks/update`, and stops with `tasks/cancel`. `@modelcontextprotocol/server/ext/tasks` is the server side of that wire, as a [server extension](../advanced/extensions.md). It owns the protocol; you own the task's state and its execution, behind a `TaskStore`. + +## Install the extension + +`TasksExtension` takes the store. `InMemoryTaskStore` is the in-process reference; a persistent store implements the same four methods. + +```ts +import { McpServer } from '@modelcontextprotocol/server'; +import { InMemoryTaskStore, TasksExtension } from '@modelcontextprotocol/server/ext/tasks'; + +const store = new InMemoryTaskStore(); +const tasks = new TasksExtension(store); + +const server = new McpServer({ name: 'report-server', version: '1.0.0' }, { extensions: [tasks] }); +``` + +The server advertises `io.modelcontextprotocol/tasks` under `capabilities.extensions` and serves `tasks/get`, `tasks/update` and `tasks/cancel`. + +## Answer a tool call with a task + +Inside a tool handler, `tasks.create(ctx)` creates the task in the store, bound to the request's principal, and returns the handle the handler answers with. The work itself is yours to start however your server runs things — here, an async function driving the in-memory store's writer handle. + +```ts +import * as z from 'zod/v4'; + +server.registerTool('send_report', { inputSchema: z.object({ to: z.string() }) }, async ({ to }, ctx) => { + const task = await tasks.create(ctx, { ttlMs: 3_600_000 }); + void sendReport(store.handle(task.taskId), to); + return task; +}); + +async function sendReport(handle: TaskHandle, to: string) { + await handle.status('compiling'); + const report = await compile(to); + const answers = await handle.requireInput({ + approve: { method: 'elicitation/create', params: { message: `send ${report.pages} pages to ${to}?`, mode: 'form', requestedSchema: { type: 'object', properties: {} } } } + }); + if (answers.approve?.action !== 'accept' || handle.signal.aborted) return handle.complete({ content: [{ type: 'text', text: 'not sent' }] }); + await deliver(report, to); + await handle.complete({ content: [{ type: 'text', text: `report sent to ${to}` }] }); +} +``` + +On the wire the tool answers a flat `CreateTaskResult` (`resultType: "task"`). `requireInput` moves the task to `input_required` and resolves once `tasks/update` has answered every key. `tasks/cancel` aborts `handle.signal`; a later `complete` or `fail` is ignored. + +## Bring your own store + +`TaskStore` is four methods over JSON: + +| Method | Contract | +| ----------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------- | +| `create(params)` | Durably create; MUST NOT resolve before a following `get` would succeed. `params.context` is yours (tool name, workflow id…). | +| `get(taskId, access?)` | The `DetailedTask` snapshot, or `undefined` for unknown, expired, or foreign-principal tasks. | +| `update(taskId, inputResponses, access?)` | Deliver answers; `false` when the task is not found. Unknown keys are ignored, partial answers accepted. | +| `cancel(taskId, access?)` | Cooperative; resolves on acknowledgement. Idempotent on terminal tasks. | + +How the work behind a task runs — a queue, a workflow engine, a durable-execution runtime — is invisible to the SDK. The store is also where a writer API lives if your execution needs one; `InMemoryTaskStore.handle` is the reference shape. + +## Wire notes + +- The extension is served on the 2026-07-28 revision. Task tools are ordinary tools without `outputSchema`; the result encoder forwards `resultType: "task"` for `tools/call` verbatim. +- A request that does not declare `io.modelcontextprotocol/tasks` in its client capabilities is refused with `-32021` — from `tasks.create`, from the `tasks/*` methods, and by the extension's `tools/call` middleware for any handle minted some other way. +- `tasks/update`'s `inputResponses` shares its name with the multi-round-trip retry field: the protocol layer lifts it out of the params and the extension reads it back from `ctx.mcpReq.inputResponses`. +- The SDK `Client` rejects `resultType: "task"` on `tools/call` (typescript-sdk#2637); the requester half of the extension is `@modelcontextprotocol/ext-tasks`. +- `notifications/tasks` over `subscriptions/listen` is not implemented (typescript-sdk#2569); polling only. + +## Recap + +- `new TasksExtension(store)` in `ServerOptions.extensions` serves the extension. +- `tasks.create(ctx, options?)` in a tool handler returns the task handle to answer with. +- `TaskStore` is create / get / update / cancel; `InMemoryTaskStore` is the reference and adds a writer `handle`. +- Execution is the server's; the SDK only defines the API shape. diff --git a/packages/client/package.json b/packages/client/package.json index c9390d3af2..654a773589 100644 --- a/packages/client/package.json +++ b/packages/client/package.json @@ -50,6 +50,16 @@ "default": "./dist/ext/skills/index.cjs" } }, + "./ext/tasks": { + "import": { + "types": "./dist/ext/tasks/index.d.mts", + "default": "./dist/ext/tasks/index.mjs" + }, + "require": { + "types": "./dist/ext/tasks/index.d.cts", + "default": "./dist/ext/tasks/index.cjs" + } + }, "./validators/ajv": { "import": { "types": "./dist/validators/ajv.d.mts", @@ -128,6 +138,9 @@ ], "stdio": [ "dist/stdio.d.mts" + ], + "ext/tasks": [ + "dist/ext/tasks/index.d.mts" ] } }, diff --git a/packages/client/src/ext/tasks/index.ts b/packages/client/src/ext/tasks/index.ts new file mode 100644 index 0000000000..6a58b1ccac --- /dev/null +++ b/packages/client/src/ext/tasks/index.ts @@ -0,0 +1,62 @@ +/** + * `@modelcontextprotocol/client/ext/tasks` — the client side of the MCP + * Tasks extension (`io.modelcontextprotocol/tasks`). + * + * `TasksClientExtension` advertises the capability, accepts task handles on + * `tools/call`, and wraps `tasks/get`, `tasks/update` and `tasks/cancel`, + * plus `waitFor` to poll a task to a terminal status. + */ + +export type { CallToolOutcome, TerminalTask, WaitForOptions } from './tasksClientExtension'; +export { TasksClientExtension } from './tasksClientExtension'; +export type { + CancelledTask, + CancelTaskParams, + CancelTaskRequest, + CancelTaskResult, + CompletedTask, + CreateTaskResult, + DetailedTask, + FailedTask, + GetTaskParams, + GetTaskRequest, + GetTaskResult, + InputRequest, + InputRequests, + InputRequiredTask, + InputResponse, + InputResponses, + Task, + TasksExtensionCapability, + TaskStatus, + UpdateTaskParams, + UpdateTaskRequest, + UpdateTaskResult, + WorkingTask +} from '@modelcontextprotocol/core-internal/ext/tasks'; +export { + cancelledTaskSchema, + cancelTaskParamsSchema, + cancelTaskRequestSchema, + cancelTaskResultSchema, + completedTaskSchema, + createTaskResultSchema, + detailedTaskSchema, + failedTaskSchema, + getTaskParamsSchema, + getTaskRequestSchema, + getTaskResultSchema, + inputRequestSchema, + inputRequestsSchema, + inputRequiredTaskSchema, + inputResponseSchema, + inputResponsesSchema, + taskSchema, + tasksExtensionCapabilitySchema, + taskStatusSchema, + updateTaskParamsSchema, + updateTaskRequestSchema, + updateTaskResultSchema, + workingTaskSchema +} from '@modelcontextprotocol/core-internal/ext/tasks'; +export { TASK_STATUSES, TASKS_EXTENSION_ID } from '@modelcontextprotocol/core-internal/ext/tasks'; diff --git a/packages/client/src/ext/tasks/tasksClientExtension.ts b/packages/client/src/ext/tasks/tasksClientExtension.ts new file mode 100644 index 0000000000..79896f327f --- /dev/null +++ b/packages/client/src/ext/tasks/tasksClientExtension.ts @@ -0,0 +1,142 @@ +/** + * `TasksClientExtension` — the client side of the MCP Tasks extension + * (`io.modelcontextprotocol/tasks`) as a {@linkcode ClientExtension}. It + * advertises the capability on every request, lets `tools/call` answer with a + * task handle, and wraps the extension's methods: `get`, `update`, `cancel`, + * and a `waitFor` that polls at the server's suggested interval until the + * task is terminal. + * + * ```ts + * const tasks = new TasksClientExtension(); + * const client = new Client(info, { extensions: [tasks] }); + * await client.connect(transport); + * + * const outcome = await tasks.callTool({ name: 'send_report', arguments: { to: 'ops' } }); + * if (outcome.kind === 'task') { + * const done = await tasks.waitFor(outcome.task.taskId); + * if (done.status === 'completed') console.log(done.result); + * } + * ``` + * + * One extension instance serves one client: `setup` binds it. + */ + +import type { CallToolRequestParams, CallToolResult, InputResponses, RequestOptions } from '@modelcontextprotocol/core-internal'; +import { CompatibilityCallToolResultSchema, SdkError, SdkErrorCode } from '@modelcontextprotocol/core-internal'; +import type { CreateTaskResult, DetailedTask } from '@modelcontextprotocol/core-internal/ext/tasks'; +import { createTaskResultSchema, detailedTaskSchema, TASKS_EXTENSION_ID } from '@modelcontextprotocol/core-internal/ext/tasks'; +import * as z from 'zod/v4'; + +import type { Client } from '../../client/client'; +import type { ClientExtension, ClientExtensionContext } from '../../client/extension'; + +/** How a task-capable `tools/call` settled: a task handle to follow, or the ordinary result. */ +export type CallToolOutcome = { kind: 'task'; task: CreateTaskResult } | { kind: 'result'; result: CallToolResult }; + +/** Options for {@link TasksClientExtension.waitFor}. */ +export interface WaitForOptions { + /** Aborting stops polling; the task itself is untouched. */ + signal?: AbortSignal; + /** Overrides the server's `pollIntervalMs` hint, ms. */ + pollIntervalMs?: number; + /** Fallback when the task carries no `pollIntervalMs`, ms. Default 1_000. */ + defaultPollIntervalMs?: number; + /** Called with every snapshot, terminal one included. */ + onUpdate?: (task: DetailedTask) => void; +} + +/** A task in a terminal status. */ +export type TerminalTask = Extract; + +const TERMINAL = new Set(['completed', 'failed', 'cancelled']); + +/** + * The `Client` consumes `resultType` before a caller schema runs for every + * kind it knows, so the neutral `DetailedTask` shape is what `tasks/get` + * answers arrive as; a task handle arrives raw, discriminator included. + */ +const callToolOutcomeSchema = z.union([createTaskResultSchema, CompatibilityCallToolResultSchema]); +const ackSchema = z.looseObject({}); + +export class TasksClientExtension implements ClientExtension { + readonly id = TASKS_EXTENSION_ID; + #client: Client | undefined; + + /** + * Binds this instance to the client and lets `tools/call` answer with a + * task handle. The client calls this for each entry in `extensions`. + * + * Experimental: the extension shape may change in a minor release. + * + * @experimental + */ + setup(ctx: ClientExtensionContext): void { + if (this.#client !== undefined) throw new Error('TasksClientExtension is already set up on a client'); + this.#client = ctx.client; + ctx.acceptResultType('tools/call', 'task'); + } + + get client(): Client { + if (this.#client === undefined) throw new SdkError(SdkErrorCode.NotConnected, 'TasksClientExtension is not set up on a client'); + return this.#client; + } + + /** + * Calls a tool that may answer with a task handle. The ordinary result + * comes back as `{ kind: 'result' }` for tools that answer synchronously. + */ + async callTool(params: CallToolRequestParams, options?: RequestOptions): Promise { + const raw = await this.client.request({ method: 'tools/call', params }, callToolOutcomeSchema, options); + if ((raw as { resultType?: unknown }).resultType === 'task') { + return { kind: 'task', task: createTaskResultSchema.parse(raw) }; + } + return { kind: 'result', result: raw as CallToolResult }; + } + + /** `tasks/get`: the task's current snapshot, results and input requests inlined. */ + async get(taskId: string, options?: RequestOptions): Promise { + const task = await this.client.request({ method: 'tasks/get', params: { taskId } }, detailedTaskSchema, options); + return task as DetailedTask; + } + + /** `tasks/update`: answers outstanding input requests. Partial answers are accepted. */ + async update(taskId: string, inputResponses: InputResponses, options?: RequestOptions): Promise { + await this.client.request({ method: 'tasks/update', params: { taskId, inputResponses } }, ackSchema, options); + } + + /** `tasks/cancel`: cooperative; resolves on acknowledgement, the task may still settle. */ + async cancel(taskId: string, options?: RequestOptions): Promise { + await this.client.request({ method: 'tasks/cancel', params: { taskId } }, ackSchema, options); + } + + /** + * Polls `tasks/get` at the task's suggested interval until the task is + * terminal. `input_required` snapshots are reported through `onUpdate` + * and polling continues; answer them with {@link update}. + */ + async waitFor(taskId: string, options?: WaitForOptions): Promise { + for (;;) { + options?.signal?.throwIfAborted(); + const task = await this.get(taskId, { signal: options?.signal }); + options?.onUpdate?.(task); + if (TERMINAL.has(task.status)) return task as TerminalTask; + const delay = options?.pollIntervalMs ?? task.pollIntervalMs ?? options?.defaultPollIntervalMs ?? 1000; + await sleep(delay, options?.signal); + } + } +} + +function sleep(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) return reject(signal.reason); + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, ms); + const onAbort = () => { + clearTimeout(timer); + reject(signal?.reason); + }; + signal?.addEventListener('abort', onAbort, { once: true }); + }); +} diff --git a/packages/client/test/ext/tasks/tasksClientExtension.test.ts b/packages/client/test/ext/tasks/tasksClientExtension.test.ts new file mode 100644 index 0000000000..750144bb8c --- /dev/null +++ b/packages/client/test/ext/tasks/tasksClientExtension.test.ts @@ -0,0 +1,148 @@ +/** + * `TasksClientExtension` against a scripted 2026-07-28 server: the + * capability rides every request, `callTool` splits task handles from plain + * results, and `waitFor` polls at the task's interval until terminal, + * reporting `input_required` on the way. + */ +import type { JSONRPCMessage } from '@modelcontextprotocol/core-internal'; +import { CLIENT_CAPABILITIES_META_KEY, InMemoryTransport } from '@modelcontextprotocol/core-internal'; +import { describe, expect, it } from 'vitest'; + +import { Client } from '../../../src/client/client'; +import { TASKS_EXTENSION_ID, TasksClientExtension } from '../../../src/ext/tasks/index'; + +const MODERN = '2026-07-28'; +const NOW = '2026-09-17T10:00:00.000Z'; + +type Snapshot = Record; + +/** A scripted server: a task whose `tasks/get` answers walk through `script`, one per poll. */ +async function scriptedServer(script: Snapshot[]) { + const [clientTx, serverTx] = InMemoryTransport.createLinkedPair(); + const written: JSONRPCMessage[] = []; + let polls = 0; + serverTx.onmessage = message => { + written.push(message); + const request = message as { id?: number | string; method?: string; params?: { name?: string; taskId?: string } }; + if (request.id === undefined) return; + const reply = (result: Snapshot) => void serverTx.send({ jsonrpc: '2.0', id: request.id as number, result }); + switch (request.method) { + case 'server/discover': { + reply({ resultType: 'complete', supportedVersions: [MODERN], capabilities: { extensions: { [TASKS_EXTENSION_ID]: {} } } }); + break; + } + case 'tools/call': { + reply( + request.params?.name === 'slow' + ? { + resultType: 'task', + taskId: 't-1', + status: 'working', + createdAt: NOW, + lastUpdatedAt: NOW, + ttlMs: null, + pollIntervalMs: 5 + } + : { resultType: 'complete', content: [{ type: 'text', text: 'fast' }] } + ); + break; + } + case 'tasks/get': { + const snapshot = script[Math.min(polls, script.length - 1)] ?? {}; + polls += 1; + reply({ + resultType: 'complete', + taskId: 't-1', + createdAt: NOW, + lastUpdatedAt: NOW, + ttlMs: null, + pollIntervalMs: 5, + ...snapshot + }); + break; + } + case 'tasks/update': + case 'tasks/cancel': { + reply({ resultType: 'complete' }); + break; + } + default: { + reply({ resultType: 'complete' }); + } + } + }; + await serverTx.start(); + const tasks = new TasksClientExtension(); + const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' }, extensions: [tasks] }); + await client.connect(clientTx); + return { client, tasks, written, pollCount: () => polls }; +} + +const methodsWritten = (written: JSONRPCMessage[]) => written.map(message => (message as { method?: string }).method).filter(Boolean); + +describe('TasksClientExtension', () => { + it('is bound to one client and refuses use before setup', async () => { + const tasks = new TasksClientExtension(); + await expect(tasks.get('t-1')).rejects.toThrow(/not set up/); + new Client({ name: 'a', version: '1' }, { extensions: [tasks] }); + expect(() => new Client({ name: 'b', version: '1' }, { extensions: [tasks] })).toThrow(/already set up/); + }); + + it('declares the extension on every request and splits task handles from plain results', async () => { + const { client, tasks, written } = await scriptedServer([]); + const plain = await tasks.callTool({ name: 'fast', arguments: {} }); + expect(plain).toEqual({ kind: 'result', result: { content: [{ type: 'text', text: 'fast' }] } }); + const slow = await tasks.callTool({ name: 'slow', arguments: {} }); + expect(slow.kind === 'task' && slow.task).toMatchObject({ + resultType: 'task', + taskId: 't-1', + status: 'working', + pollIntervalMs: 5 + }); + const call = written.find(message => (message as { method?: string }).method === 'tools/call') as { + params: { _meta: Record }; + }; + expect(call.params._meta[CLIENT_CAPABILITIES_META_KEY]).toEqual({ extensions: { [TASKS_EXTENSION_ID]: {} } }); + await client.close(); + }); + + it('waitFor polls at the task interval, reports input_required, and resolves on the terminal snapshot', async () => { + const { client, tasks, pollCount } = await scriptedServer([ + { status: 'working', statusMessage: 'starting' }, + { + status: 'input_required', + inputRequests: { + q: { + method: 'elicitation/create', + params: { message: '?', mode: 'form', requestedSchema: { type: 'object', properties: {} } } + } + } + }, + { status: 'completed', result: { content: [{ type: 'text', text: 'done' }] } } + ]); + const seen: string[] = []; + const done = await tasks.waitFor('t-1', { + onUpdate: task => { + seen.push(task.status); + if (task.status === 'input_required') void tasks.update('t-1', { q: { action: 'accept' } }); + } + }); + expect(seen).toEqual(['working', 'input_required', 'completed']); + expect(done).toMatchObject({ status: 'completed', result: { content: [{ type: 'text', text: 'done' }] } }); + expect(pollCount()).toBe(3); + await client.close(); + }); + + it('waitFor stops on abort without touching the task; cancel is an explicit call', async () => { + const { client, tasks, written } = await scriptedServer([{ status: 'working' }]); + const controller = new AbortController(); + const waiting = tasks.waitFor('t-1', { signal: controller.signal, pollIntervalMs: 1000 }); + await new Promise(resolve => setTimeout(resolve, 10)); + controller.abort(new Error('stop polling')); + await expect(waiting).rejects.toThrow('stop polling'); + expect(methodsWritten(written)).not.toContain('tasks/cancel'); + await tasks.cancel('t-1'); + expect(methodsWritten(written)).toContain('tasks/cancel'); + await client.close(); + }); +}); diff --git a/packages/client/tsconfig.json b/packages/client/tsconfig.json index 4fb474f8fb..97218ce9e9 100644 --- a/packages/client/tsconfig.json +++ b/packages/client/tsconfig.json @@ -21,7 +21,8 @@ "./node_modules/@modelcontextprotocol/core-internal/src/validators/cfWorkerProvider.ts" ], "@modelcontextprotocol/test-helpers": ["./node_modules/@modelcontextprotocol/test-helpers/src/index.ts"], - "@modelcontextprotocol/client/_shims": ["./src/shimsNode.ts"] + "@modelcontextprotocol/client/_shims": ["./src/shimsNode.ts"], + "@modelcontextprotocol/core-internal/ext/tasks": ["./node_modules/@modelcontextprotocol/core-internal/src/ext/tasks/index.ts"] } } } diff --git a/packages/client/tsdown.config.ts b/packages/client/tsdown.config.ts index d4a8de998a..368ff2af10 100644 --- a/packages/client/tsdown.config.ts +++ b/packages/client/tsdown.config.ts @@ -5,6 +5,7 @@ export default defineConfig({ entry: [ 'src/index.ts', 'src/stdio.ts', + 'src/ext/tasks/index.ts', 'src/shimsNode.ts', 'src/shimsWorkerd.ts', 'src/shimsBrowser.ts', @@ -31,6 +32,7 @@ export default defineConfig({ '@modelcontextprotocol/core-internal': ['../core-internal/src/index.ts'], '@modelcontextprotocol/core-internal/public': ['../core-internal/src/exports/public/index.ts'], '@modelcontextprotocol/core-internal/ext/skills': ['../core-internal/src/ext/skills/index.ts'], + '@modelcontextprotocol/core-internal/ext/tasks': ['../core-internal/src/ext/tasks/index.ts'], '@modelcontextprotocol/core-internal/validators/ajv': ['../core-internal/src/validators/ajvProvider.ts'], '@modelcontextprotocol/core-internal/validators/cfWorker': ['../core-internal/src/validators/cfWorkerProvider.ts'] } diff --git a/packages/core-internal/package.json b/packages/core-internal/package.json index 23d5d1761c..0c122d00e5 100644 --- a/packages/core-internal/package.json +++ b/packages/core-internal/package.json @@ -37,6 +37,10 @@ "types": "./src/ext/skills/index.ts", "import": "./src/ext/skills/index.ts" }, + "./ext/tasks": { + "types": "./src/ext/tasks/index.ts", + "import": "./src/ext/tasks/index.ts" + }, "./validators/ajv": { "types": "./src/validators/ajvProvider.ts", "import": "./src/validators/ajvProvider.ts" diff --git a/packages/core-internal/src/ext/tasks/index.ts b/packages/core-internal/src/ext/tasks/index.ts new file mode 100644 index 0000000000..f55a8bda64 --- /dev/null +++ b/packages/core-internal/src/ext/tasks/index.ts @@ -0,0 +1,7 @@ +/** + * Wire types and zod schemas of the MCP Tasks extension + * (`io.modelcontextprotocol/tasks`, schema 2026-07-28), shared by the server + * and client halves of the extension. + */ +export * from './schemas'; +export * from './types'; diff --git a/packages/core-internal/src/ext/tasks/schemas.ts b/packages/core-internal/src/ext/tasks/schemas.ts new file mode 100644 index 0000000000..2542a5fb47 --- /dev/null +++ b/packages/core-internal/src/ext/tasks/schemas.ts @@ -0,0 +1,177 @@ +/* + * Hand-written zod v4 schemas for the MCP Tasks extension wire types + * (./types). Upstream commits only generated JSON Schema + * (`schema/draft/schema.json`, vendored at `test/fixtures/ext-tasks.schema.json` + * as the conformance fixture); these runtime schemas are authored against + * modelcontextprotocol/ext-tasks pinned at commit dcc8d2b (SEP-2663 Final). + * https://github.com/modelcontextprotocol/ext-tasks + * + * Deliberate deviations from the generated fixture, which reflects the + * pre-envelope TS source rather than the wire: + * - Objects are loose (unknown keys pass through) where the fixture says + * `additionalProperties: false`: modern responses carry `resultType` and + * `_meta`, and modern request params carry the `_meta` envelope. + * - `resultType` literals are REQUIRED on result schemas (spec MUST; the + * fixture omits the field entirely). + * - `InputRequest`/`InputResponse` get minimal structural checks (the fixture + * degenerates them to `anyOf [{}, {}, {}]`). + * + * Copyright (c) Model Context Protocol contributors + */ + +import * as z from 'zod/v4'; + +import { TASK_STATUSES } from './types'; + +/** `TaskStatus` */ +export const taskStatusSchema = z.enum(TASK_STATUSES); + +const metaSchema = z.record(z.string(), z.unknown()); + +/** + * An embedded (de-JSON-RPC'd) input request: an elicitation, sampling, or + * roots request object (`{method, params}`), validated structurally. + */ +export const inputRequestSchema = z.looseObject({ + method: z.string(), + params: z.record(z.string(), z.unknown()).optional() +}); + +/** An embedded input response: the bare result object for its request. */ +export const inputResponseSchema = z.record(z.string(), z.unknown()); + +/** `InputRequests` — keyed by identifiers unique over the task's lifetime. */ +export const inputRequestsSchema = z.record(z.string(), inputRequestSchema); + +/** `InputResponses` — keys correspond to outstanding inputRequest keys. */ +export const inputResponsesSchema = z.record(z.string(), inputResponseSchema); + +const taskShape = { + taskId: z.string(), + status: taskStatusSchema, + statusMessage: z.string().optional(), + createdAt: z.string(), + lastUpdatedAt: z.string(), + ttlMs: z.int().nullable(), + pollIntervalMs: z.int().optional() +}; + +/** `Task` */ +export const taskSchema = z.looseObject(taskShape); + +/** `WorkingTask` */ +export const workingTaskSchema = z.looseObject({ + ...taskShape, + status: z.literal('working') +}); + +/** `InputRequiredTask` */ +export const inputRequiredTaskSchema = z.looseObject({ + ...taskShape, + status: z.literal('input_required'), + inputRequests: inputRequestsSchema +}); + +/** `CompletedTask` — the original request's result structure inlined. */ +export const completedTaskSchema = z.looseObject({ + ...taskShape, + status: z.literal('completed'), + result: z.record(z.string(), z.unknown()) +}); + +/** `FailedTask` — the JSON-RPC error object inlined. */ +export const failedTaskSchema = z.looseObject({ + ...taskShape, + status: z.literal('failed'), + error: z.record(z.string(), z.unknown()) +}); + +/** `CancelledTask` */ +export const cancelledTaskSchema = z.looseObject({ + ...taskShape, + status: z.literal('cancelled') +}); + +/** `DetailedTask` — discriminated on `status`. */ +export const detailedTaskSchema = z.discriminatedUnion('status', [ + workingTaskSchema, + inputRequiredTaskSchema, + completedTaskSchema, + failedTaskSchema, + cancelledTaskSchema +]); + +/** `CreateTaskResult` — flat `Result & Task` with `resultType: "task"` (MUST). */ +export const createTaskResultSchema = z.looseObject({ + ...taskShape, + resultType: z.literal('task'), + _meta: metaSchema.optional() +}); + +const completeResultShape = { + resultType: z.literal('complete'), + _meta: metaSchema.optional() +}; + +/** `GetTaskResult` — a `DetailedTask` variant with `resultType: "complete"` (MUST). */ +export const getTaskResultSchema = z.discriminatedUnion('status', [ + workingTaskSchema.extend(completeResultShape), + inputRequiredTaskSchema.extend(completeResultShape), + completedTaskSchema.extend(completeResultShape), + failedTaskSchema.extend(completeResultShape), + cancelledTaskSchema.extend(completeResultShape) +]); + +/** `UpdateTaskResult` — empty ack with `resultType: "complete"` (MUST). */ +export const updateTaskResultSchema = z.looseObject(completeResultShape); + +/** `CancelTaskResult` — empty ack with `resultType: "complete"` (MUST). */ +export const cancelTaskResultSchema = z.looseObject(completeResultShape); + +const requestIdSchema = z.union([z.string(), z.int()]); + +/** `tasks/get` params (`_meta` carries the modern per-request envelope). */ +export const getTaskParamsSchema = z.looseObject({ + taskId: z.string(), + _meta: metaSchema.optional() +}); + +/** `GetTaskRequest` */ +export const getTaskRequestSchema = z.looseObject({ + jsonrpc: z.literal('2.0'), + id: requestIdSchema, + method: z.literal('tasks/get'), + params: getTaskParamsSchema +}); + +/** `tasks/update` params (`_meta` carries the modern per-request envelope). */ +export const updateTaskParamsSchema = z.looseObject({ + taskId: z.string(), + inputResponses: inputResponsesSchema, + _meta: metaSchema.optional() +}); + +/** `UpdateTaskRequest` */ +export const updateTaskRequestSchema = z.looseObject({ + jsonrpc: z.literal('2.0'), + id: requestIdSchema, + method: z.literal('tasks/update'), + params: updateTaskParamsSchema +}); + +/** `tasks/cancel` params (`_meta` carries the modern per-request envelope). */ +export const cancelTaskParamsSchema = z.looseObject({ + taskId: z.string(), + _meta: metaSchema.optional() +}); + +/** `CancelTaskRequest` */ +export const cancelTaskRequestSchema = z.looseObject({ + jsonrpc: z.literal('2.0'), + id: requestIdSchema, + method: z.literal('tasks/cancel'), + params: cancelTaskParamsSchema +}); + +/** `TasksExtensionCapability` — an empty object declares support. */ +export const tasksExtensionCapabilitySchema = z.strictObject({}); diff --git a/packages/core-internal/src/ext/tasks/types.ts b/packages/core-internal/src/ext/tasks/types.ts new file mode 100644 index 0000000000..494b1d5a29 --- /dev/null +++ b/packages/core-internal/src/ext/tasks/types.ts @@ -0,0 +1,220 @@ +/* + * MCP Tasks extension wire types (extension id: io.modelcontextprotocol/tasks). + * + * Adapted from modelcontextprotocol/ext-tasks, pinned at commit dcc8d2b + * (SEP-2663 Final): `schema/draft/schema.ts`, re-based on + * `@modelcontextprotocol/server` v2 types — the modern SDK ships the MRTR + * `InputRequest`/`InputResponse` unions the upstream file's TODOs point at, so + * those are re-exported rather than re-declared. The JSON-RPC request shapes + * are declared standalone (not extending the SDK's `JSONRPCRequest`) so the + * wire contract stays pinned to this file. `notifications/tasks` and the + * subscription additions are omitted: v1 is polling-only (the engine contract)). + * + * This module (with ./schemas) is the ONLY import source for task wire types + * in this repo — the SDK's deprecated 2025-11-25 task exports (`Task`, + * `CreateTaskResult`, `TaskStatus`, `GetTaskRequest`, ...) carry the removed + * legacy wire shape and must not be used (the engine contract)). + * https://github.com/modelcontextprotocol/ext-tasks + * + * Copyright (c) Model Context Protocol contributors + */ + +import type { InputRequests, InputResponses, Result } from '../../types/index'; + +/** + * A single input request / response embedded in a task, re-based on the SDK + * v2 MRTR unions (sampling, roots, or elicitation). Keys in the containing + * maps MUST be unique over the lifetime of a single task. + */ +export type { InputRequest, InputRequests, InputResponse, InputResponses } from '../../types/index'; + +/** The MCP Tasks extension identifier. An empty-object capability declares support. */ +export const TASKS_EXTENSION_ID = 'io.modelcontextprotocol/tasks'; + +/** All task statuses, in lifecycle order (terminal states last). */ +export const TASK_STATUSES = ['working', 'input_required', 'completed', 'failed', 'cancelled'] as const; + +/** The status of a task. */ +export type TaskStatus = (typeof TASK_STATUSES)[number]; + +/** Data associated with a task. */ +export interface Task { + /** The task identifier. */ + taskId: string; + + /** Current task status. */ + status: TaskStatus; + + /** + * Optional human-readable message describing the current task state: + * progress descriptions for "working", blocked work for "input_required", + * reasons for "cancelled", summaries for "completed", diagnostics for + * "failed". + */ + statusMessage?: string; + + /** ISO 8601 timestamp when the task was created. */ + createdAt: string; + + /** ISO 8601 timestamp when the task was last updated. */ + lastUpdatedAt: string; + + /** + * Time-to-live duration from creation in integer milliseconds, null for + * unlimited. The server may discard the task after the TTL elapses. This + * value MAY change over the lifetime of a task. + */ + ttlMs: number | null; + + /** + * Suggested polling interval in integer milliseconds. Clients SHOULD honor + * this value to avoid overwhelming the server. This value MAY change over + * the lifetime of a task. + */ + pollIntervalMs?: number; +} + +/** A task that is in a normal working state. */ +export interface WorkingTask extends Task { + status: 'working'; +} + +/** A task that is waiting for input from the client. */ +export interface InputRequiredTask extends Task { + status: 'input_required'; + + /** + * Server-to-client requests that need to be fulfilled during task + * execution. Keys are arbitrary identifiers for matching requests to + * responses. + */ + inputRequests: InputRequests; +} + +/** A task that has completed successfully. */ +export interface CompletedTask extends Task { + status: 'completed'; + + /** + * The final result of the task. The structure matches the result type of + * the original request — for a `tools/call` task, the `CallToolResult` + * structure. + */ + result: { [key: string]: unknown }; +} + +/** A task that has failed due to a JSON-RPC error during execution. */ +export interface FailedTask extends Task { + status: 'failed'; + + /** The JSON-RPC error that caused the task to fail. */ + error: { [key: string]: unknown }; +} + +/** A task that has been cancelled. */ +export interface CancelledTask extends Task { + status: 'cancelled'; +} + +/** + * A task with status-specific fields inlined, as returned by `tasks/get`: + * terminal results or pending input requests ride on the snapshot itself. + */ +export type DetailedTask = WorkingTask | InputRequiredTask | CompletedTask | FailedTask | CancelledTask; + +/** + * The result returned by a server in lieu of a standard result shape when it + * elects to process a request asynchronously — flat `Result & Task`. The + * `resultType` field MUST be `"task"` on the wire; it is declared explicitly + * here (the upstream type leaves it to the old SDK `Result`'s index + * signature, which the v2 `Result` no longer has). + */ +export type CreateTaskResult = Result & Task & { resultType: 'task' }; + +/** Parameters of a `tasks/get` request. */ +export interface GetTaskParams { + /** The task identifier to query. */ + taskId: string; + + /** + * The modern (2026-07-28) per-request envelope: carries + * `io.modelcontextprotocol/clientCapabilities` among other keys. Not part + * of the upstream extension schema's params (which lists only `taskId`) — + * declared here because every modern request threads it. + */ + _meta?: Record; +} + +/** A request to retrieve the state of a task. */ +export interface GetTaskRequest { + jsonrpc: '2.0'; + id: string | number; + method: 'tasks/get'; + params: GetTaskParams; +} + +/** + * The response to `tasks/get`: the appropriate {@link DetailedTask} variant + * for the task's current status, with `resultType: "complete"` (MUST). + */ +export type GetTaskResult = Result & DetailedTask & { resultType: 'complete' }; + +/** Parameters of a `tasks/update` request. */ +export interface UpdateTaskParams { + /** The task identifier to update. */ + taskId: string; + + /** + * Responses to outstanding inputRequests previously surfaced by the + * server. Each key MUST correspond to a currently-outstanding inputRequest + * key (unknown keys are ignored; partial responses are accepted). + */ + inputResponses: InputResponses; + + /** The modern per-request envelope (see {@link GetTaskParams._meta}). */ + _meta?: Record; +} + +/** A request to provide input responses to a task in the input_required state. */ +export interface UpdateTaskRequest { + jsonrpc: '2.0'; + id: string | number; + method: 'tasks/update'; + params: UpdateTaskParams; +} + +/** + * The response to `tasks/update`: an empty acknowledgement (eventually + * consistent), with `resultType: "complete"` (MUST). + */ +export type UpdateTaskResult = Result & { resultType: 'complete' }; + +/** Parameters of a `tasks/cancel` request. */ +export interface CancelTaskParams { + /** The task identifier to cancel. */ + taskId: string; + + /** The modern per-request envelope (see {@link GetTaskParams._meta}). */ + _meta?: Record; +} + +/** A request to cancel a task. Cancellation is cooperative and eventually consistent. */ +export interface CancelTaskRequest { + jsonrpc: '2.0'; + id: string | number; + method: 'tasks/cancel'; + params: CancelTaskParams; +} + +/** + * The response to `tasks/cancel`: an empty acknowledgement (ack does not mean + * stopped), with `resultType: "complete"` (MUST). + */ +export type CancelTaskResult = Result & { resultType: 'complete' }; + +/** + * The extension capability declaration for the tasks extension. An empty + * object indicates support; no extension-specific settings are currently + * defined. + */ +export type TasksExtensionCapability = Record; diff --git a/packages/core-internal/test/packageTopologyPins.test.ts b/packages/core-internal/test/packageTopologyPins.test.ts index 5bc882e8e3..6716f30cfd 100644 --- a/packages/core-internal/test/packageTopologyPins.test.ts +++ b/packages/core-internal/test/packageTopologyPins.test.ts @@ -39,13 +39,22 @@ const PUBLIC_PACKAGES: Record void; +} + +interface TaskRecord { + taskId: string; + principal?: string; + context?: Record; + status: TaskStatus; + statusMessage?: string; + createdAt: number; + lastUpdatedAt: number; + ttlMs: number | null; + pollIntervalMs?: number; + result?: CallToolResult; + error?: { code: number; message: string; data?: unknown }; + pendingInput?: PendingInput; + abort: AbortController; + ttlTimer?: ReturnType; +} + +const TERMINAL: ReadonlySet = new Set(['completed', 'failed', 'cancelled']); + +/** + * The writer side of a task, for the code that does the work. Obtained from + * {@link InMemoryTaskStore.handle}; a persistent store exposes its own + * equivalent (or none, when the work reports through the store directly). + */ +export interface TaskHandle { + readonly taskId: string; + /** Aborted when `tasks/cancel` is acknowledged. */ + readonly signal: AbortSignal; + /** Writes `statusMessage` for pollers. */ + status(message: string): Promise; + /** + * Moves the task to `input_required` with `requests` outstanding and + * resolves once `tasks/update` has answered every key (partial updates + * accumulate). Rejects if the task is cancelled while waiting. + */ + requireInput(requests: InputRequests): Promise; + complete(result: CallToolResult): Promise; + fail(error: { code: number; message: string; data?: unknown }): Promise; +} + +/** Options for {@link InMemoryTaskStore}. */ +export interface InMemoryTaskStoreOptions { + /** Task id factory. Default `crypto.randomUUID()`. */ + createTaskId?: () => string; + /** Clock, for tests. Default `Date.now`. */ + now?: () => number; +} + +export class InMemoryTaskStore implements TaskStore { + readonly #tasks = new Map(); + readonly #createTaskId: () => string; + readonly #now: () => number; + + constructor(options?: InMemoryTaskStoreOptions) { + this.#createTaskId = options?.createTaskId ?? (() => crypto.randomUUID()); + this.#now = options?.now ?? (() => Date.now()); + } + + /** Clears every timer and forgets every task. For tests and shutdown. */ + close(): void { + for (const record of this.#tasks.values()) { + if (record.ttlTimer !== undefined) clearTimeout(record.ttlTimer); + record.abort.abort(); + } + this.#tasks.clear(); + } + + async create(params: CreateTaskParams): Promise { + if (params.ttlMs !== null && (!Number.isSafeInteger(params.ttlMs) || params.ttlMs < 0)) { + throw new RangeError(`ttlMs must be a non-negative integer or null, got ${params.ttlMs}`); + } + if (params.pollIntervalMs !== undefined && (!Number.isSafeInteger(params.pollIntervalMs) || params.pollIntervalMs < 0)) { + throw new RangeError(`pollIntervalMs must be a non-negative integer, got ${params.pollIntervalMs}`); + } + const now = this.#now(); + const record: TaskRecord = { + taskId: this.#createTaskId(), + ...(params.principal !== undefined && { principal: params.principal }), + ...(params.context !== undefined && { context: params.context }), + status: 'working', + createdAt: now, + lastUpdatedAt: now, + ttlMs: params.ttlMs, + ...(params.pollIntervalMs !== undefined && { pollIntervalMs: params.pollIntervalMs }), + abort: new AbortController() + }; + if (record.ttlMs !== null) { + record.ttlTimer = setTimeout(() => this.#purge(record), record.ttlMs); + record.ttlTimer.unref?.(); + } + this.#tasks.set(record.taskId, record); + return this.#snapshot(record); + } + + async get(taskId: string, access?: TaskAccess): Promise { + const record = this.#lookup(taskId, access); + return record === undefined ? undefined : this.#detailed(record); + } + + async update(taskId: string, inputResponses: InputResponses, access?: TaskAccess): Promise { + const record = this.#lookup(taskId, access); + if (record === undefined) return false; + const pending = record.pendingInput; + if (pending === undefined || TERMINAL.has(record.status)) return true; + for (const [key, response] of Object.entries(inputResponses)) { + // Unknown keys are ignored; the first answer to a key wins. + if (key in pending.requests && !(key in pending.responses)) pending.responses[key] = response; + } + this.#touch(record); + if (Object.keys(pending.requests).every(key => key in pending.responses)) { + record.pendingInput = undefined; + record.status = 'working'; + pending.resolve(pending.responses); + } + return true; + } + + async cancel(taskId: string, access?: TaskAccess): Promise { + const record = this.#lookup(taskId, access); + if (record === undefined) return false; + if (TERMINAL.has(record.status)) return true; + record.status = 'cancelled'; + record.pendingInput = undefined; + this.#touch(record); + record.abort.abort(); + return true; + } + + /** The context recorded at creation, for the server's execution to pick the work up. */ + context(taskId: string): Record | undefined { + return this.#tasks.get(taskId)?.context; + } + + /** The writer handle for a task the store holds. Throws for an unknown task. */ + handle(taskId: string): TaskHandle { + const record = this.#tasks.get(taskId); + if (record === undefined) throw new Error(`Task "${taskId}" not found`); + const live = (): TaskRecord | undefined => { + const current = this.#tasks.get(taskId); + return current === undefined || TERMINAL.has(current.status) ? undefined : current; + }; + return { + taskId, + signal: record.abort.signal, + status: async message => { + const current = live(); + if (current === undefined) return; + current.statusMessage = message; + this.#touch(current); + }, + requireInput: requests => { + const current = live(); + if (current === undefined) return Promise.reject(new Error(`Task "${taskId}" is no longer running`)); + return new Promise((resolve, reject) => { + current.pendingInput = { requests, responses: {}, resolve }; + current.status = 'input_required'; + this.#touch(current); + current.abort.signal.addEventListener('abort', () => reject(new Error(`Task "${taskId}" was cancelled`)), { + once: true + }); + }); + }, + complete: async result => { + const current = live(); + if (current === undefined) return; + current.status = 'completed'; + current.result = result; + this.#touch(current); + }, + fail: async error => { + const current = live(); + if (current === undefined) return; + current.status = 'failed'; + current.error = error; + this.#touch(current); + } + }; + } + + // ------------------------------------------------------------- internals -- + + #lookup(taskId: string, access: TaskAccess | undefined): TaskRecord | undefined { + const record = this.#tasks.get(taskId); + if (record === undefined) return undefined; + if (record.principal !== undefined && record.principal !== access?.principal) return undefined; + return record; + } + + #touch(record: TaskRecord): void { + record.lastUpdatedAt = this.#now(); + } + + #purge(record: TaskRecord): void { + record.abort.abort(); + this.#tasks.delete(record.taskId); + } + + #snapshot(record: TaskRecord): Task { + return { + taskId: record.taskId, + status: record.status, + ...(record.statusMessage !== undefined && { statusMessage: record.statusMessage }), + createdAt: new Date(record.createdAt).toISOString(), + lastUpdatedAt: new Date(record.lastUpdatedAt).toISOString(), + ttlMs: record.ttlMs, + ...(record.pollIntervalMs !== undefined && { pollIntervalMs: record.pollIntervalMs }) + }; + } + + #detailed(record: TaskRecord): DetailedTask { + const base = this.#snapshot(record); + switch (record.status) { + case 'working': { + return { ...base, status: 'working' }; + } + case 'input_required': { + const pending = record.pendingInput; + const inputRequests: InputRequests = {}; + for (const [key, request] of Object.entries(pending?.requests ?? {})) { + if (!(key in (pending?.responses ?? {}))) inputRequests[key] = request; + } + return { ...base, status: 'input_required', inputRequests }; + } + case 'completed': { + return { ...base, status: 'completed', result: (record.result ?? { content: [] }) as { [key: string]: unknown } }; + } + case 'failed': { + return { ...base, status: 'failed', error: record.error ?? { code: -32_603, message: 'Task failed' } }; + } + case 'cancelled': { + return { ...base, status: 'cancelled' }; + } + } + } +} diff --git a/packages/server/src/ext/tasks/index.ts b/packages/server/src/ext/tasks/index.ts new file mode 100644 index 0000000000..2cdcf9684a --- /dev/null +++ b/packages/server/src/ext/tasks/index.ts @@ -0,0 +1,68 @@ +/** + * `@modelcontextprotocol/server/ext/tasks` — the server side of the MCP + * Tasks extension (`io.modelcontextprotocol/tasks`). + * + * `TasksExtension` owns the wire: capability, `tasks/get` / `tasks/update` / + * `tasks/cancel`, the client-capability check, and the `tools/call` task + * handle. A `TaskStore` owns the task: `InMemoryTaskStore` is the + * in-process reference; durable stores implement the same four methods and + * live outside the SDK. How the work behind a task runs is the server's + * own. + */ + +export type { InMemoryTaskStoreOptions, TaskHandle } from './inMemoryStore'; +export { InMemoryTaskStore } from './inMemoryStore'; +export type { CreateTaskParams, TaskAccess, TaskStore } from './store'; +export type { CreateTaskOptions, TasksExtensionOptions, TaskToolResult } from './tasksExtension'; +export { declaresTasksExtension, TasksExtension } from './tasksExtension'; +export type { + CancelledTask, + CancelTaskParams, + CancelTaskRequest, + CancelTaskResult, + CompletedTask, + CreateTaskResult, + DetailedTask, + FailedTask, + GetTaskParams, + GetTaskRequest, + GetTaskResult, + InputRequest, + InputRequests, + InputRequiredTask, + InputResponse, + InputResponses, + Task, + TasksExtensionCapability, + TaskStatus, + UpdateTaskParams, + UpdateTaskRequest, + UpdateTaskResult, + WorkingTask +} from '@modelcontextprotocol/core-internal/ext/tasks'; +export { + cancelledTaskSchema, + cancelTaskParamsSchema, + cancelTaskRequestSchema, + cancelTaskResultSchema, + completedTaskSchema, + createTaskResultSchema, + detailedTaskSchema, + failedTaskSchema, + getTaskParamsSchema, + getTaskRequestSchema, + getTaskResultSchema, + inputRequestSchema, + inputRequestsSchema, + inputRequiredTaskSchema, + inputResponseSchema, + inputResponsesSchema, + taskSchema, + tasksExtensionCapabilitySchema, + taskStatusSchema, + updateTaskParamsSchema, + updateTaskRequestSchema, + updateTaskResultSchema, + workingTaskSchema +} from '@modelcontextprotocol/core-internal/ext/tasks'; +export { TASK_STATUSES, TASKS_EXTENSION_ID } from '@modelcontextprotocol/core-internal/ext/tasks'; diff --git a/packages/server/src/ext/tasks/store.ts b/packages/server/src/ext/tasks/store.ts new file mode 100644 index 0000000000..dc655f13c9 --- /dev/null +++ b/packages/server/src/ext/tasks/store.ts @@ -0,0 +1,54 @@ +/** + * The store interface of the Tasks extension: what the `tasks/*` request handlers + * and `TasksExtension.create` call. Everything here is request/response over + * JSON. How a task's work is executed — in-process, on a queue, in a + * durable-execution runtime — is the server's business, not the SDK's: the + * SDK only cares that a task can be created, read, answered, and cancelled. + */ + +import type { InputResponses } from '@modelcontextprotocol/core-internal'; +import type { DetailedTask, Task } from '@modelcontextprotocol/core-internal/ext/tasks'; + +/** What `TasksExtension.create` asks a store to durably create. */ +export interface CreateTaskParams { + /** Retention from creation, ms; `null` = unlimited. */ + ttlMs: number | null; + /** Suggested client polling interval, ms; omitted = no hint on the task. */ + pollIntervalMs?: number; + /** + * Auth binding: the principal the creating request was authenticated as, + * when the transport knows one. Stores MUST refuse `tasks/*` access from + * a different principal (fail closed, no existence leak) and MAY ignore + * it when absent. + */ + principal?: string; + /** + * Free-form, store-defined. The originating tool name and arguments, a + * workflow id — whatever the server's execution needs to find or start + * the work. Opaque to the SDK. + */ + context?: Record; +} + +/** The caller identity presented on a `tasks/*` request. */ +export interface TaskAccess { + principal?: string; +} + +/** + * A task store. Engine selection is configuration: the extension and the + * tool handlers that create tasks are byte-identical across stores. + * + * - `create` MUST NOT resolve before a subsequent `get(taskId)` would succeed + * (durable creation, a Tasks extension rule). + * - `get`, `update`, `cancel` resolve `undefined` / `false` for an unknown, + * expired, or foreign-principal task; the extension answers `-32602`. + * - `cancel` is cooperative: it resolves on acknowledgement, the task may + * still settle `completed` or `failed`. Idempotent on terminal tasks. + */ +export interface TaskStore { + create(params: CreateTaskParams): Promise; + get(taskId: string, access?: TaskAccess): Promise; + update(taskId: string, inputResponses: InputResponses, access?: TaskAccess): Promise; + cancel(taskId: string, access?: TaskAccess): Promise; +} diff --git a/packages/server/src/ext/tasks/tasksExtension.ts b/packages/server/src/ext/tasks/tasksExtension.ts new file mode 100644 index 0000000000..dd197864fd --- /dev/null +++ b/packages/server/src/ext/tasks/tasksExtension.ts @@ -0,0 +1,172 @@ +/** + * `TasksExtension` — the server side of the MCP Tasks extension + * (`io.modelcontextprotocol/tasks`) as a {@linkcode ServerExtension}. It + * owns the wire: the capability, the per-request client-capability check, + * the `tasks/get`, `tasks/update` and `tasks/cancel` methods, and the + * `tools/call` result shape. Everything about the task itself — where its + * state lives, how its work runs — is behind the {@link TaskStore} the + * server passes in. + * + * ```ts + * const store = new InMemoryTaskStore(); + * const tasks = new TasksExtension(store); + * const server = new McpServer(info, { extensions: [tasks] }); + * + * server.registerTool('send_report', { inputSchema }, async (input, ctx) => { + * const task = await tasks.create(ctx, { ttlMs: 3_600_000 }); + * void runReport(store.handle(task.taskId), input); // the server's own execution + * return task; + * }); + * ``` + */ + +import type { CallToolResult, InputResponses, JSONRPCRequest, Result, ServerContext } from '@modelcontextprotocol/core-internal'; +import { + CLIENT_CAPABILITIES_META_KEY, + MissingRequiredClientCapabilityError, + ProtocolError, + ProtocolErrorCode +} from '@modelcontextprotocol/core-internal'; +import type { CancelTaskResult, CreateTaskResult, GetTaskResult, UpdateTaskResult } from '@modelcontextprotocol/core-internal/ext/tasks'; +import { + cancelTaskParamsSchema, + getTaskParamsSchema, + inputResponsesSchema, + TASKS_EXTENSION_ID +} from '@modelcontextprotocol/core-internal/ext/tasks'; + +import type { ServerExtension, ServerExtensionContext } from '../../server/extension'; +import type { CreateTaskParams, TaskStore } from './store'; + +/** Options for {@link TasksExtension}. */ +export interface TasksExtensionOptions { + /** Default retention for `create` when the caller gives none, ms; `null` = unlimited. Default 86_400_000 (24h). */ + defaultTtlMs?: number | null; + /** Default polling hint for `create` when the caller gives none, ms. Default 5_000. */ + defaultPollIntervalMs?: number; +} + +/** What a tool handler passes to {@link TasksExtension.create}; every field optional. */ +export type CreateTaskOptions = Partial>; + +/** + * The task handle a tool handler returns in place of a `CallToolResult`. + * The 2026-07-28 result encoder forwards `resultType: "task"` for `tools/call` + * verbatim, so the flat handle is the wire result. + */ +export type TaskToolResult = CallToolResult & CreateTaskResult; + +const isRecord = (value: unknown): value is Record => value !== null && typeof value === 'object' && !Array.isArray(value); + +/** Whether this request's `_meta` envelope declared the tasks extension. */ +export const declaresTasksExtension = (ctx: ServerContext): boolean => { + const envelope = ctx.mcpReq.envelope as Record | undefined; + const capabilities = envelope?.[CLIENT_CAPABILITIES_META_KEY]; + if (!isRecord(capabilities)) return false; + const extensions = capabilities['extensions']; + return isRecord(extensions) && TASKS_EXTENSION_ID in extensions; +}; + +const requireTasksExtension = (ctx: ServerContext, what: string): void => { + if (declaresTasksExtension(ctx)) return; + throw new MissingRequiredClientCapabilityError( + { requiredCapabilities: { extensions: { [TASKS_EXTENSION_ID]: {} } } }, + `${what} requires the request to declare the "${TASKS_EXTENSION_ID}" extension capability` + ); +}; + +const principalOf = (ctx: ServerContext): string | undefined => ctx.http?.authInfo?.clientId; + +const notFound = (): never => { + throw new ProtocolError(ProtocolErrorCode.InvalidParams, 'Task not found'); +}; + +/** + * `tasks/update` params as the HANDLER sees them. `inputResponses` is a + * reserved multi-round-trip name on the 2026-07-28 revision: the protocol + * layer lifts it out of every client request's params before dispatch and + * surfaces it at `ctx.mcpReq.inputResponses`. The wire schema + * (`updateTaskParamsSchema`) keeps the field required; here it is optional + * and read back from the context. + */ +const updateTaskHandlerParamsSchema = getTaskParamsSchema.extend({ inputResponses: inputResponsesSchema.optional() }); + +export class TasksExtension implements ServerExtension { + readonly id = TASKS_EXTENSION_ID; + readonly store: TaskStore; + readonly #defaultTtlMs: number | null; + readonly #defaultPollIntervalMs: number; + + constructor(store: TaskStore, options?: TasksExtensionOptions) { + this.store = store; + this.#defaultTtlMs = options?.defaultTtlMs === undefined ? 86_400_000 : options.defaultTtlMs; + this.#defaultPollIntervalMs = options?.defaultPollIntervalMs ?? 5000; + } + + /** + * Creates a task for the current `tools/call` and returns the handle the + * tool handler answers with. Refuses (`-32021`) when the request did not + * declare the extension; binds the task to the request's principal when + * the transport knows one. + */ + async create(ctx: ServerContext, options?: CreateTaskOptions): Promise { + requireTasksExtension(ctx, `Tool "${ctx.mcpReq.method}" executes as a task and`); + const principal = principalOf(ctx); + const task = await this.store.create({ + ttlMs: options?.ttlMs === undefined ? this.#defaultTtlMs : options.ttlMs, + pollIntervalMs: options?.pollIntervalMs ?? this.#defaultPollIntervalMs, + ...(principal !== undefined && { principal }), + ...(options?.context !== undefined && { context: options.context }) + }); + const result: CreateTaskResult = { resultType: 'task', ...task }; + return result as TaskToolResult; + } + + /** + * Registers the extension's methods and its `tools/call` middleware. The + * server calls this for each entry in `extensions`. + * + * Experimental: the extension shape may change in a minor release. + * + * @experimental + */ + setup(ctx: ServerExtensionContext): void { + ctx.setRequestHandler('tasks/get', { params: getTaskParamsSchema }, async (params, requestCtx): Promise => { + requireTasksExtension(requestCtx, 'tasks/get'); + const task = await this.store.get(params.taskId, { principal: principalOf(requestCtx) }); + if (task === undefined) return notFound(); + return { resultType: 'complete', ...task }; + }); + + ctx.setRequestHandler( + 'tasks/update', + { params: updateTaskHandlerParamsSchema }, + async (params, requestCtx): Promise => { + requireTasksExtension(requestCtx, 'tasks/update'); + const inputResponses = (requestCtx.mcpReq.inputResponses ?? params.inputResponses ?? {}) as InputResponses; + const found = await this.store.update(params.taskId, inputResponses, { principal: principalOf(requestCtx) }); + if (!found) return notFound(); + return { resultType: 'complete' }; + } + ); + + ctx.setRequestHandler('tasks/cancel', { params: cancelTaskParamsSchema }, async (params, requestCtx): Promise => { + requireTasksExtension(requestCtx, 'tasks/cancel'); + const found = await this.store.cancel(params.taskId, { principal: principalOf(requestCtx) }); + if (!found) return notFound(); + return { resultType: 'complete' }; + }); + + // A task handle may only be answered to a client that can consume it. + // `create(ctx, …)` checks up front; this catches handles minted some + // other way (an external engine's own create) — the extension owns + // the wire regardless of where the task came from. + ctx.use('tools/call', async (request: JSONRPCRequest, requestCtx, next): Promise => { + const result = await next(request, requestCtx); + if ((result as { resultType?: unknown }).resultType === 'task') { + requireTasksExtension(requestCtx, `Tool "${String((request.params as { name?: unknown })?.name)}" executes as a task and`); + } + return result; + }); + } +} diff --git a/packages/server/test/ext/tasks/inMemoryStore.test.ts b/packages/server/test/ext/tasks/inMemoryStore.test.ts new file mode 100644 index 0000000000..59473ada47 --- /dev/null +++ b/packages/server/test/ext/tasks/inMemoryStore.test.ts @@ -0,0 +1,82 @@ +/** + * `InMemoryTaskStore`: the store semantics a persistent implementation must + * reproduce — durable create, partial input answers, first answer wins, + * cancel aborts a pending input wait, terminal writes are ignored, TTL purge, + * principal fail-closed. + */ +import { describe, expect, it } from 'vitest'; + +import { InMemoryTaskStore } from '../../../src/ext/tasks/index'; + +const tick = (ms = 0) => new Promise(resolve => setTimeout(resolve, ms)); + +describe('InMemoryTaskStore', () => { + it('creates durably and records context for the execution to pick up', async () => { + const store = new InMemoryTaskStore({ createTaskId: () => 'fixed' }); + const task = await store.create({ ttlMs: null, pollIntervalMs: 50, context: { tool: 'greet' } }); + expect(task).toMatchObject({ taskId: 'fixed', status: 'working', ttlMs: null, pollIntervalMs: 50 }); + expect(await store.get('fixed')).toMatchObject({ status: 'working' }); + expect(store.context('fixed')).toEqual({ tool: 'greet' }); + store.close(); + }); + + it('accumulates partial input answers; the first answer to a key wins; unknown keys are ignored', async () => { + const store = new InMemoryTaskStore(); + const { taskId } = await store.create({ ttlMs: null }); + const handle = store.handle(taskId); + const answers = handle.requireInput({ + a: { + method: 'elicitation/create', + params: { message: 'q', mode: 'form', requestedSchema: { type: 'object', properties: {} } } + }, + b: { method: 'elicitation/create', params: { message: 'q', mode: 'form', requestedSchema: { type: 'object', properties: {} } } } + }); + await tick(); + expect(await store.get(taskId)).toMatchObject({ status: 'input_required', inputRequests: { a: {}, b: {} } }); + expect(await store.update(taskId, { a: { action: 'accept' }, zzz: { action: 'accept' } })).toBe(true); + expect(await store.get(taskId)).toMatchObject({ status: 'input_required', inputRequests: { b: {} } }); + await store.update(taskId, { a: { action: 'decline' }, b: { action: 'cancel' } }); + expect(await answers).toEqual({ a: { action: 'accept' }, b: { action: 'cancel' } }); + expect((await store.get(taskId))?.status).toBe('working'); + store.close(); + }); + + it('cancel rejects a pending input wait, aborts the signal, and later writes are ignored', async () => { + const store = new InMemoryTaskStore(); + const { taskId } = await store.create({ ttlMs: null }); + const handle = store.handle(taskId); + const waiting = handle.requireInput({ + a: { method: 'elicitation/create', params: { message: 'q', mode: 'form', requestedSchema: { type: 'object', properties: {} } } } + }); + expect(await store.cancel(taskId)).toBe(true); + await expect(waiting).rejects.toThrow('cancelled'); + expect(handle.signal.aborted).toBe(true); + await handle.complete({ content: [] }); + await handle.status('late'); + expect(await store.get(taskId)).toMatchObject({ status: 'cancelled' }); + expect((await store.get(taskId))?.statusMessage).toBeUndefined(); + expect(await store.cancel(taskId)).toBe(true); + expect(await store.cancel('missing')).toBe(false); + store.close(); + }); + + it('purges at the TTL deadline and fails closed on a foreign principal', async () => { + const store = new InMemoryTaskStore(); + const { taskId } = await store.create({ ttlMs: 10, principal: 'alice' }); + expect(await store.get(taskId)).toBeUndefined(); + expect(await store.get(taskId, { principal: 'bob' })).toBeUndefined(); + expect((await store.get(taskId, { principal: 'alice' }))?.taskId).toBe(taskId); + expect(await store.update(taskId, {}, { principal: 'bob' })).toBe(false); + await tick(30); + expect(await store.get(taskId, { principal: 'alice' })).toBeUndefined(); + expect(() => store.handle(taskId)).toThrow('not found'); + store.close(); + }); + + it('rejects invalid ttlMs and pollIntervalMs', async () => { + const store = new InMemoryTaskStore(); + await expect(store.create({ ttlMs: -1 })).rejects.toBeInstanceOf(RangeError); + await expect(store.create({ ttlMs: null, pollIntervalMs: 1.5 })).rejects.toBeInstanceOf(RangeError); + store.close(); + }); +}); diff --git a/packages/server/test/ext/tasks/tasks.e2e.test.ts b/packages/server/test/ext/tasks/tasks.e2e.test.ts new file mode 100644 index 0000000000..7e19463ecf --- /dev/null +++ b/packages/server/test/ext/tasks/tasks.e2e.test.ts @@ -0,0 +1,196 @@ +/** + * End to end through a real `Client` against the stateless `createMcpHandler` + * (a fresh `McpServer` per request): nothing about a running task lives on + * the server instance that answered `tools/call`; the store is the only + * shared state, and the work runs wherever the server chooses — here, a + * plain async function driving the store's writer handle. + */ +import { Client, StreamableHTTPClientTransport } from '@modelcontextprotocol/client'; +import { TasksClientExtension } from '@modelcontextprotocol/client/ext/tasks'; +import { afterEach, describe, expect, it } from 'vitest'; +import * as z from 'zod/v4'; + +import type { DetailedTask, TaskHandle } from '../../../src/ext/tasks/index'; +import { InMemoryTaskStore, TASKS_EXTENSION_ID, TasksExtension } from '../../../src/ext/tasks/index'; + +const TASKS_CAPABILITY = { extensions: { [TASKS_EXTENSION_ID]: {} } }; +import { CLIENT_CAPABILITIES_META_KEY, createMcpHandler, McpServer, PROTOCOL_VERSION_META_KEY } from '../../../src/index'; + +type Work = (handle: TaskHandle, input: { name: string }) => Promise; + +function createHarness(work: Work, options?: { declareExtension?: boolean; plainTool?: boolean }) { + const store = new InMemoryTaskStore(); + const tasks = new TasksExtension(store, { defaultPollIntervalMs: 10 }); + const createServer = () => { + const server = new McpServer({ name: 'tasks-test', version: '1.0.0' }, { extensions: [tasks] }); + server.registerTool('greet', { inputSchema: z.object({ name: z.string() }) }, async (input, ctx) => { + const task = await tasks.create(ctx); + void work(store.handle(task.taskId), input); + return task; + }); + if (options?.plainTool) { + // A handle minted outside `tasks.create`: the tools/call override still gates it. + server.registerTool('sneaky', {}, async () => { + const task = await store.create({ ttlMs: null }); + return { resultType: 'task', ...task } as never; + }); + } + return server; + }; + const mcpHandler = createMcpHandler(createServer); + const declare = options?.declareExtension !== false; + /** A raw `tools/call` POST, for the cases where the client deliberately does not declare the extension. */ + const callTool = async (name: string, args: Record) => { + const body = { + jsonrpc: '2.0', + id: 1, + method: 'tools/call', + params: { + name, + arguments: args, + _meta: { [PROTOCOL_VERSION_META_KEY]: '2026-07-28', [CLIENT_CAPABILITIES_META_KEY]: declare ? TASKS_CAPABILITY : {} } + } + }; + const response = await mcpHandler.fetch( + new Request('http://test.local/mcp', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Accept: 'application/json, text/event-stream', + 'MCP-Protocol-Version': '2026-07-28', + 'Mcp-Method': 'tools/call', + 'Mcp-Name': name + }, + body: JSON.stringify(body) + }) + ); + const text = await response.text(); + const payload = response.headers.get('content-type')?.includes('text/event-stream') + ? text + .split('\n') + .filter(line => line.startsWith('data:')) + .map(line => line.slice(5).trim()) + .at(-1) + : text; + return JSON.parse(payload ?? '{}') as { result?: Record; error?: { code: number; message: string } }; + }; + const startTask = async (name: string) => { + const outcome = await clientTasks.callTool({ name: 'greet', arguments: { name } }); + if (outcome.kind !== 'task') throw new Error('expected a task handle'); + return outcome.task; + }; + const transport = new StreamableHTTPClientTransport(new URL('http://test.local/mcp'), { + fetch: (url, init) => mcpHandler.fetch(new Request(url, init)) + }); + const clientTasks = new TasksClientExtension(); + const client = new Client( + { name: 'harness', version: '1.0.0' }, + { versionNegotiation: { mode: 'auto' }, extensions: declare ? [clientTasks] : [] } + ); + return { store, client, transport, clientTasks, startTask, callTool }; +} + +async function pollUntil(tasks: TasksClientExtension, taskId: string, predicate: (task: DetailedTask) => boolean) { + for (let i = 0; i < 200; i++) { + const task = await tasks.get(taskId); + if (predicate(task)) return task; + await new Promise(resolve => setTimeout(resolve, 5)); + } + throw new Error(`task ${taskId} never reached the expected state`); +} + +describe('TasksExtension end to end (stateless handler, in-memory store)', () => { + let cleanup: (() => void) | undefined; + afterEach(() => cleanup?.()); + + it('advertises the extension, answers a task handle, and completes with status', async () => { + const h = createHarness(async (handle, { name }) => { + await handle.status(`greeting ${name}`); + await new Promise(resolve => setTimeout(resolve, 10)); + await handle.complete({ content: [{ type: 'text', text: `hello ${name}` }] }); + }); + cleanup = () => h.store.close(); + await h.client.connect(h.transport); + expect(h.client.getServerCapabilities()?.extensions).toEqual({ [TASKS_EXTENSION_ID]: {} }); + + const created = await h.startTask('ada'); + expect(created).toMatchObject({ resultType: 'task', status: 'working', ttlMs: 86_400_000, pollIntervalMs: 10 }); + + const seen: string[] = []; + const done = await h.clientTasks.waitFor(created.taskId, { onUpdate: task => seen.push(task.status) }); + expect(done.status === 'completed' && done.result).toEqual({ content: [{ type: 'text', text: 'hello ada' }] }); + expect(done.statusMessage).toBe('greeting ada'); + expect(seen.at(-1)).toBe('completed'); + }); + + it('surfaces input_required, resumes on tasks/update, and inlines the answer', async () => { + const h = createHarness(async (handle, { name }) => { + const answers = await handle.requireInput({ + confirm: { + method: 'elicitation/create', + params: { message: `greet ${name}?`, mode: 'form', requestedSchema: { type: 'object', properties: {} } } + } + }); + await handle.complete({ content: [{ type: 'text', text: JSON.stringify(answers) }] }); + }); + cleanup = () => h.store.close(); + await h.client.connect(h.transport); + const created = await h.startTask('bob'); + const waiting = await pollUntil(h.clientTasks, created.taskId, task => task.status === 'input_required'); + expect(waiting.status === 'input_required' && Object.keys(waiting.inputRequests)).toEqual(['confirm']); + + await h.clientTasks.update(created.taskId, { confirm: { action: 'accept', content: { ok: true } } }); + const done = await pollUntil(h.clientTasks, created.taskId, task => task.status === 'completed'); + expect(done.status === 'completed' && done.result).toEqual({ + content: [{ type: 'text', text: JSON.stringify({ confirm: { action: 'accept', content: { ok: true } } }) }] + }); + }); + + it('fails a task with the reported error', async () => { + const h = createHarness(async handle => { + await handle.fail({ code: -32_000, message: 'upstream down' }); + }); + cleanup = () => h.store.close(); + await h.client.connect(h.transport); + const created = await h.startTask('x'); + const done = await pollUntil(h.clientTasks, created.taskId, task => task.status === 'failed'); + expect(done.status === 'failed' && done.error).toEqual({ code: -32_000, message: 'upstream down' }); + }); + + it('cancels cooperatively: the handle signal aborts and later writes are ignored', async () => { + let aborted = false; + const h = createHarness(async handle => { + await new Promise(resolve => handle.signal.addEventListener('abort', () => resolve(), { once: true })); + aborted = true; + await handle.complete({ content: [] }); + }); + cleanup = () => h.store.close(); + await h.client.connect(h.transport); + const created = await h.startTask('x'); + await h.clientTasks.cancel(created.taskId); + const done = await pollUntil(h.clientTasks, created.taskId, task => task.status === 'cancelled'); + expect(done.status).toBe('cancelled'); + await new Promise(resolve => setTimeout(resolve, 5)); + expect(aborted).toBe(true); + expect((await h.clientTasks.get(created.taskId)).status).toBe('cancelled'); + await expect(h.clientTasks.cancel(created.taskId)).resolves.toBeUndefined(); // idempotent + }); + + it('answers -32602 for an unknown task and -32021 without the extension capability, as JSON-RPC errors', async () => { + const h = createHarness(async () => {}); + cleanup = () => h.store.close(); + await h.client.connect(h.transport); + await expect(h.clientTasks.get('nope')).rejects.toMatchObject({ code: -32_602 }); + + const plain = createHarness(async () => {}, { declareExtension: false, plainTool: true }); + await plain.client.connect(plain.transport); + await expect(plain.client.request({ method: 'tasks/get', params: { taskId: 'nope' } }, z.looseObject({}))).rejects.toMatchObject({ + code: -32_021 + }); + const refused = await plain.callTool('greet', { name: 'x' }); + expect(refused.error).toMatchObject({ code: -32_021 }); + const sneaky = await plain.callTool('sneaky', {}); + expect(sneaky.error).toMatchObject({ code: -32_021 }); + plain.store.close(); + }); +}); diff --git a/packages/server/test/ext/tasks/withSkills.e2e.test.ts b/packages/server/test/ext/tasks/withSkills.e2e.test.ts new file mode 100644 index 0000000000..37e48d9849 --- /dev/null +++ b/packages/server/test/ext/tasks/withSkills.e2e.test.ts @@ -0,0 +1,66 @@ +/** + * Two extensions on one server, end to end: the Skills and Tasks extensions, + * passed together in `extensions` and served by the stateless + * `createMcpHandler` to a real `Client`. + */ +import { Client, StreamableHTTPClientTransport } from '@modelcontextprotocol/client'; +import { SkillsClient } from '@modelcontextprotocol/client/ext/skills'; +import { TasksClientExtension } from '@modelcontextprotocol/client/ext/tasks'; +import { afterEach, describe, expect, it } from 'vitest'; +import * as z from 'zod/v4'; + +import { memorySkillFiles, SkillsExtension } from '../../../src/ext/skills/index'; +import { InMemoryTaskStore, TASKS_EXTENSION_ID, TasksExtension } from '../../../src/ext/tasks/index'; +import { createMcpHandler, McpServer } from '../../../src/index'; + +const SKILL_MD = '---\nname: release-notes\ndescription: Write release notes.\n---\n# Release notes\n'; + +describe('Skills and Tasks extensions on one server', () => { + let store: InMemoryTaskStore | undefined; + afterEach(() => store?.close()); + + it('serves skills and answers a tool with a task, through one client', async () => { + store = new InMemoryTaskStore(); + const taskStore = store; + const tasks = new TasksExtension(taskStore, { defaultPollIntervalMs: 10 }); + const skills = await SkillsExtension.load(memorySkillFiles({ 'release-notes/SKILL.md': SKILL_MD })); + + const mcpHandler = createMcpHandler(() => { + const server = new McpServer({ name: 'skills-and-tasks', version: '1.0.0' }, { extensions: [skills, tasks] }); + server.registerTool('draft_notes', { inputSchema: z.object({ version: z.string() }) }, async ({ version }, ctx) => { + const task = await tasks.create(ctx); + void taskStore.handle(task.taskId).complete({ content: [{ type: 'text', text: `notes for ${version}` }] }); + return task; + }); + return server; + }); + + const clientTasks = new TasksClientExtension(); + const client = new Client({ name: 'host', version: '1.0.0' }, { versionNegotiation: { mode: 'auto' }, extensions: [clientTasks] }); + await client.connect( + new StreamableHTTPClientTransport(new URL('http://test.local/mcp'), { + fetch: (url, init) => mcpHandler.fetch(new Request(url, init)) + }) + ); + + expect(client.getServerCapabilities()?.extensions).toEqual({ + 'io.modelcontextprotocol/skills': { directoryRead: true }, + [TASKS_EXTENSION_ID]: {} + }); + + const skillsClient = new SkillsClient(client); + const { skills: entries } = await skillsClient.list(); + expect(entries.map(entry => entry.uri)).toEqual(['skill://release-notes/SKILL.md']); + const [entry] = entries; + if (entry === undefined) throw new Error('expected one skill'); + const manifest = await skillsClient.read(entry); + expect(manifest).toMatchObject({ verified: true, text: SKILL_MD }); + + const outcome = await clientTasks.callTool({ name: 'draft_notes', arguments: { version: '2.4.0' } }); + if (outcome.kind !== 'task') throw new Error('expected a task handle'); + const done = await clientTasks.waitFor(outcome.task.taskId); + expect(done.status === 'completed' && done.result).toEqual({ content: [{ type: 'text', text: 'notes for 2.4.0' }] }); + + await client.close(); + }); +}); diff --git a/packages/server/tsconfig.json b/packages/server/tsconfig.json index 54f3389a5f..494526ddc8 100644 --- a/packages/server/tsconfig.json +++ b/packages/server/tsconfig.json @@ -21,7 +21,12 @@ "./node_modules/@modelcontextprotocol/core-internal/src/validators/cfWorkerProvider.ts" ], "@modelcontextprotocol/test-helpers": ["./node_modules/@modelcontextprotocol/test-helpers/src/index.ts"], - "@modelcontextprotocol/server/_shims": ["./src/shimsNode.ts"] + "@modelcontextprotocol/server/_shims": ["./src/shimsNode.ts"], + "@modelcontextprotocol/client": ["./node_modules/@modelcontextprotocol/client/src/index.ts"], + "@modelcontextprotocol/client/_shims": ["./node_modules/@modelcontextprotocol/client/src/shimsNode.ts"], + "@modelcontextprotocol/core-internal/ext/tasks": ["./node_modules/@modelcontextprotocol/core-internal/src/ext/tasks/index.ts"], + "@modelcontextprotocol/client/ext/tasks": ["./node_modules/@modelcontextprotocol/client/src/ext/tasks/index.ts"], + "@modelcontextprotocol/client/ext/skills": ["./node_modules/@modelcontextprotocol/client/src/ext/skills/index.ts"] } } } diff --git a/packages/server/tsdown.config.ts b/packages/server/tsdown.config.ts index 06137a4560..efacd7d4bd 100644 --- a/packages/server/tsdown.config.ts +++ b/packages/server/tsdown.config.ts @@ -5,6 +5,7 @@ export default defineConfig({ entry: [ 'src/index.ts', 'src/stdio.ts', + 'src/ext/tasks/index.ts', 'src/shimsNode.ts', 'src/shimsWorkerd.ts', 'src/shimsBrowser.ts', @@ -32,6 +33,7 @@ export default defineConfig({ '@modelcontextprotocol/core-internal': ['../core-internal/src/index.ts'], '@modelcontextprotocol/core-internal/public': ['../core-internal/src/exports/public/index.ts'], '@modelcontextprotocol/core-internal/ext/skills': ['../core-internal/src/ext/skills/index.ts'], + '@modelcontextprotocol/core-internal/ext/tasks': ['../core-internal/src/ext/tasks/index.ts'], '@modelcontextprotocol/core-internal/validators/ajv': ['../core-internal/src/validators/ajvProvider.ts'], '@modelcontextprotocol/core-internal/validators/cfWorker': ['../core-internal/src/validators/cfWorkerProvider.ts'] } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 5369810c7c..08da1a51dc 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1822,6 +1822,9 @@ importers: '@eslint/js': specifier: catalog:devTools version: 9.39.4 + '@modelcontextprotocol/client': + specifier: workspace:^ + version: link:../client '@modelcontextprotocol/core-internal': specifier: workspace:^ version: link:../core-internal