diff --git a/apps/website/content/docs/core-concepts/meta.json b/apps/website/content/docs/core-concepts/meta.json index b4908e58f..776748365 100644 --- a/apps/website/content/docs/core-concepts/meta.json +++ b/apps/website/content/docs/core-concepts/meta.json @@ -4,6 +4,7 @@ "tools", "prompts", "resources", + "tasks", "middlewares", "css", "external-clients" diff --git a/apps/website/content/docs/core-concepts/tasks.mdx b/apps/website/content/docs/core-concepts/tasks.mdx new file mode 100644 index 000000000..72c602bd7 --- /dev/null +++ b/apps/website/content/docs/core-concepts/tasks.mdx @@ -0,0 +1,183 @@ +--- +title: "Tasks" +metadataTitle: "Tasks | xmcp Documentation" +publishedAt: "2026-06-10" +summary: "Learn how to run tools as long-running, pollable tasks in your xmcp application." +description: "Tasks let a tool call run as a durable, pollable operation. The call returns immediately with a task id, and clients poll for status and fetch the result later. xmcp keeps this fully stateless by storing task state in an external store you provide." +--- + +`xmcp` enables tasks when you add a `src/task-store.ts` file. A task lets a tool call run as a long-running, pollable operation, following the [MCP tasks utility](https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/tasks). Instead of blocking until the tool finishes, the server returns a task immediately. Clients poll `tasks/get` for status and call `tasks/result` for the output once the task reaches a terminal state. + +Tasks are stateless. The HTTP transport builds a fresh server per request and never holds task state in memory. All state lives in a task store that you back with external storage such as Redis, a database, or a KV service. This makes tasks a good fit for serverless. + + + Tasks are experimental in the MCP specification and may evolve in future + protocol versions. + + +## The task store + +The task store is the single source of truth. Default-export one from `src/task-store.ts`. Its presence enables tasks and advertises the `tasks` capability during initialization. + +`createTaskStore` builds a complete store from four persistence methods. It handles id generation, timestamps, the status lifecycle, TTL expiry, and pagination for you: + +```ts title="src/task-store.ts" +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { createTaskStore, type TaskRecord } from "xmcp"; + +const dir = path.resolve(process.cwd(), ".tasks"); +const file = (taskId: string) => path.join(dir, `${taskId}.json`); + +export const store = createTaskStore({ + async get(taskId) { + try { + return JSON.parse(await fs.readFile(file(taskId), "utf-8")) as TaskRecord; + } catch { + return null; + } + }, + async set(taskId, record) { + await fs.mkdir(dir, { recursive: true }); + await fs.writeFile(file(taskId), JSON.stringify(record)); + }, + async delete(taskId) { + await fs.rm(file(taskId), { force: true }); + }, + async list() { + const names = (await fs.readdir(dir)).filter((n) => n.endsWith(".json")); + return Promise.all( + names.map((n) => + fs.readFile(path.join(dir, n), "utf-8").then((r) => JSON.parse(r)) + ) + ); + }, +}); + +export default store; +``` + +The adapter only stores and loads a `TaskRecord` (a JSON object) by id. `createTaskStore` accepts optional tuning as a second argument: + +**`defaultTtlMs`** + +- Lifetime in ms when the requestor doesn't ask for one +- Defaults to `null` (unlimited) + +**`pollIntervalMs`** + +- Suggested client poll interval, surfaced in task responses +- Defaults to `1000` + +**`pageSize`** + +- Page size for `tasks/list` +- Defaults to `50` + +## Marking a tool as a task + +A tool opts in with `taskSupport` in its `metadata`: + +```ts title="src/tools/long-job.ts" +import { z } from "zod"; +import { type InferSchema, type ToolExtraArguments, type ToolMetadata } from "xmcp"; +import { store } from "../task-store"; + +export const schema = { + label: z.string(), +}; + +export const metadata: ToolMetadata = { + name: "long_job", + description: "Runs a long job as a task.", + taskSupport: "required", +}; + +export default async function longJob( + { label }: InferSchema, + extra: ToolExtraArguments +) { + const taskId = extra.task?.taskId; + + // Kick off the work out-of-band and return nothing: the task stays "working" + // until your worker writes the result back through the store. + void enqueue({ taskId, label }); +} +``` + +`taskSupport` accepts: + +- **`"forbidden"`** (default): the tool is never task-augmented. +- **`"optional"`**: clients may call it as a task or as a normal call. +- **`"required"`**: clients must call it as a task. + +When a tool runs as a task, `extra.task.taskId` is available. Use it to key the work you enqueue and to write the result back to your store. + +## Execution + +xmcp creates the task and serves status and result reads. It does not run the background work. You decide how the tool executes: + +- **Return a result synchronously.** If your handler returns a `CallToolResult` (or a string or number), xmcp stores it as the terminal result immediately. This is good for quick work or long-lived Node servers. +- **Complete it out-of-band.** Return nothing to leave the task `working`, enqueue the work (queue, cron, worker), and have that worker write the result back through the store. This is the right model for serverless, where the function may freeze after returning. + + + If a tool is `"optional"` and a client calls it **without** a task, it is + expected to return a normal result synchronously. Use `"required"` for work + that only ever completes out-of-band. + + +### Returning a string from a worker + +An out-of-band worker can store a plain `string` or `number`. xmcp coerces it into a `CallToolResult` when the client reads it back, exactly like returning a string from a normal tool: + +```ts title="src/jobs/worker.ts" +await store.storeTaskResult(taskId, "completed", `Job "${label}" finished.`); +// the client receives: { content: [{ type: "text", text: "Job \"...\" finished." }] } +``` + +If you store an object instead, it must already be a valid `CallToolResult` (for example `{ content: [{ type: "text", text: "..." }] }` or `{ structuredContent: {...} }`). Objects are returned to the client unchanged. + +## Using another store + +The adapter is just `get` / `set` / `delete` / `list` over a JSON record, so any KV or database fits. Point the four methods at it: + +```ts title="src/task-store.ts" +import { createTaskStore } from "xmcp"; +import { redis } from "./redis"; + +export const store = createTaskStore({ + get: (taskId) => redis.get(`task:${taskId}`), + set: (taskId, record) => redis.set(`task:${taskId}`, record), + delete: (taskId) => redis.del(`task:${taskId}`), + list: () => redis.mget(`task:*`), +}); + +export default store; +``` + + + `list()` only powers `tasks/list`. `createTaskStore` paginates its result in + memory, which is fine for moderate volumes. For very large datasets, + [implement the store yourself](#implementing-the-store-yourself) and paginate + in the query. You can also let your store's native TTL (such as Redis + `EXPIRE`) expire records; `createTaskStore` already evicts expired tasks on + read. + + +### Implementing the store yourself + +`createTaskStore` is a convenience. When you need full control, such as query-level pagination, custom expiry, or multi-tenant keys, implement the `TaskStore` interface directly and default-export it from `src/task-store.ts`. It exposes `createTask`, `getTask`, `storeTaskResult`, `getTaskResult`, `updateTaskStatus`, and `listTasks`. Generate ids with `generateTaskId` (Web Crypto, cryptographically secure), keep a terminal task (`completed`, `failed`, or `cancelled`) terminal, and use ISO 8601 strings for `createdAt` and `lastUpdatedAt`. + +## Lifecycle + +1. The client calls the tool with a `task` field. xmcp creates a task (`status: "working"`) in your store and returns immediately. +2. The client polls `tasks/get` until the task reaches a terminal status (`completed`, `failed`, or `cancelled`), respecting the `pollInterval`. +3. The client calls `tasks/result` for the underlying tool result. A result with `isError: true` maps to `failed`. +4. `tasks/cancel` moves a non-terminal task to `cancelled`; cancelling a terminal task returns a `-32602` error. +5. After the `ttl` elapses, the store may delete the task. + +## Security + +Stateless HTTP has no requestor identity, so the task id is the only access control. xmcp generates cryptographically secure ids, and you should prefer short TTLs. If you need stronger isolation between callers, add [authentication](/docs/authentication/jwt). + +See the [`tasks-http` example](https://github.com/basementstudio/xmcp/tree/main/examples/tasks-http) for a complete, runnable server. diff --git a/examples/tasks-http/.gitignore b/examples/tasks-http/.gitignore new file mode 100644 index 000000000..23e4004f6 --- /dev/null +++ b/examples/tasks-http/.gitignore @@ -0,0 +1,5 @@ +.vercel +.xmcp +xmcp-env.d.ts +.tasks +.queue diff --git a/examples/tasks-http/README.md b/examples/tasks-http/README.md new file mode 100644 index 000000000..5436285a4 --- /dev/null +++ b/examples/tasks-http/README.md @@ -0,0 +1,119 @@ +# Tasks over HTTP + +This example shows how to run a tool as an [MCP task](https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/tasks) +in xmcp: a long-running, pollable operation that returns immediately and is +completed **out-of-band, by a separate process**. + +xmcp keeps this **stateless**. The HTTP transport builds a fresh server per +request and never holds task state in memory. All task state lives in a +**task store** that you implement against external storage. + +## Two processes, one shared store + +The whole point of tasks is that the work runs **somewhere other than the +request**. This example makes that boundary explicit with two processes that +share nothing but on-disk state: + +``` +MCP server (the tool) worker (separate process) + enqueue job ──────► .queue/ ──────► run job + │ + tasks/get / tasks/result ▼ + client ◄────────────── .tasks/ ◄── store result +``` + +1. A tool opts in with `taskSupport: "required"` in its `metadata` + (`src/tools/long-job.ts`). When a client calls it with a `task` field, xmcp + creates a task in the store and returns immediately. **The tool only drops a + job on the queue — it does not run the work.** +2. The worker (`src/worker.ts`) is a standalone process. It polls the queue, + runs the job, and writes the result back through the **same** task store. +3. Clients poll `tasks/get` for status and call `tasks/result` for the output. + They read from the store the worker wrote to — the server that created the + task is long gone. + +In production the queue is a real queue (SQS, a database table, Upstash) and the +worker is a queue consumer or cron; the shape is identical. + +The task store here writes one JSON file per task under `.tasks/`, and the queue +writes one job file under `.queue/`. Both are dependency-free and fine for +local/single-node use. For serverless or multi-instance deployments, back the +store with Redis, a database, or a KV service. + +> The worker logs each job as it runs. That console output is the point of this +> example — it makes execution happening *elsewhere* visible. + +## Run it + +```bash +pnpm dev +``` + +`dev` starts **both** processes — the MCP server and the worker — so tasks +complete out of the box. The worker is still a separate process (`tsx watch +src/worker.ts &`); it just shares the terminal. + +The server listens on `http://127.0.0.1:3001/mcp`. + +> **A task stuck on `working` means the worker isn't running.** The tool only +> enqueues; if nothing drains the queue, the task never completes. `pnpm dev` +> runs the worker for you — or run it yourself with `pnpm worker`. + +To see the two processes truly side by side, run them in separate terminals: + +```bash +# terminal 1 — server only +pnpm build && pnpm start + +# terminal 2 — worker only +pnpm worker +``` + +## Try the flow + +Create a task-augmented call (note the `task` field): + +```bash +curl -s http://127.0.0.1:3001/mcp \ + -H 'content-type: application/json' \ + -H 'accept: application/json, text/event-stream' \ + -d '{ + "jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": { "name": "long_job", "arguments": { "label": "demo", "seconds": 3 }, "task": { "ttl": 60000 } } + }' +# -> result.task = { taskId, status: "working", ... } +``` + +Watch terminal 2: the worker prints `▶ running "demo" (3s)…` and then +`✓ completed`. Now poll status and fetch the result: + +```bash +curl -s http://127.0.0.1:3001/mcp -H 'content-type: application/json' \ + -H 'accept: application/json, text/event-stream' \ + -d '{ "jsonrpc": "2.0", "id": 2, "method": "tasks/get", "params": { "taskId": "" } }' + +curl -s http://127.0.0.1:3001/mcp -H 'content-type: application/json' \ + -H 'accept: application/json, text/event-stream' \ + -d '{ "jsonrpc": "2.0", "id": 3, "method": "tasks/result", "params": { "taskId": "" } }' +``` + +Because all state lives in the store, you can kill and restart the server +between `tools/call` and `tasks/get` — the task still resolves. That is +statelessness. + +## Notes and caveats + +- **The tool never does the work.** It enqueues and returns; a separate process + completes the task and writes the result back through the store. This is what + makes tasks safe on serverless, where the request function may freeze the + moment it returns. +- **`required` vs `optional`.** Use `required` for work that completes + out-of-band. With `optional`, a client that calls the tool *without* a `task` + field expects a normal synchronous result, so an `optional` tool should return + a `CallToolResult` directly in that case. +- **Returning a string.** The worker stores a plain `string`; xmcp coerces it + into a `CallToolResult` when the client reads it back, exactly like returning + a string from a normal tool. +- **Security.** Stateless HTTP has no requestor identity, so the task ID is the + only access control. Use cryptographically secure IDs (see `generateTaskId`) + and short TTLs. Add authentication if you need stronger isolation. diff --git a/examples/tasks-http/package.json b/examples/tasks-http/package.json new file mode 100644 index 000000000..a71d009cb --- /dev/null +++ b/examples/tasks-http/package.json @@ -0,0 +1,26 @@ +{ + "name": "Tasks HTTP", + "description": "Run long-running tools as MCP tasks over stateless HTTP with an external task store", + "keywords": [ + "tasks", + "long-running", + "http", + "stateless" + ], + "scripts": { + "build": "xmcp build", + "dev": "tsx watch src/worker.ts & xmcp dev", + "worker": "tsx watch src/worker.ts", + "start": "node dist/http.js", + "type-check": "tsc --noEmit" + }, + "dependencies": { + "xmcp": "workspace:*", + "zod": "^4.0.10" + }, + "devDependencies": { + "@types/node": "^20.19.26", + "tsx": "^4.21.0", + "typescript": "^5.9.3" + } +} diff --git a/examples/tasks-http/src/queue.ts b/examples/tasks-http/src/queue.ts new file mode 100644 index 000000000..398bf9fad --- /dev/null +++ b/examples/tasks-http/src/queue.ts @@ -0,0 +1,81 @@ +import { promises as fs } from "node:fs"; +import path from "node:path"; + +/** + * A tiny filesystem job queue: the explicit boundary between the tool (which + * only *enqueues* work) and the worker (which *executes* it, in a separate + * process). The tool and the worker never share memory — only these job files + * and the task store on disk. This mirrors the real serverless shape: an HTTP + * function drops a job on a queue, and a separate worker/cron picks it up. + */ +export interface Job { + taskId: string; + label: string; + seconds: number; +} + +const dir = path.resolve(process.cwd(), ".queue"); +const file = (taskId: string) => path.join(dir, `${taskId}.json`); + +/** + * How often the worker checks the queue. A production worker would receive + * jobs through its queue's native delivery (SQS long-poll, Redis `BLPOP`, a + * webhook); this dependency-free example polls the directory instead. + */ +const POLL_INTERVAL_MS = 500; + +const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** Drop a job on the queue. Called by the tool; returns immediately. */ +export async function enqueue(job: Job): Promise { + await fs.mkdir(dir, { recursive: true }); + await fs.writeFile(file(job.taskId), JSON.stringify(job), "utf-8"); +} + +/** Read a job and remove it from the queue so it runs at most once. */ +async function claimJob(taskId: string): Promise { + try { + const raw = await fs.readFile(file(taskId), "utf-8"); + await fs.rm(file(taskId), { force: true }); + return JSON.parse(raw) as Job; + } catch { + return null; + } +} + +async function pendingTaskIds(): Promise { + try { + return (await fs.readdir(dir)) + .filter((n) => n.endsWith(".json")) + .map((n) => n.slice(0, -".json".length)); + } catch { + return []; + } +} + +/** + * Poll the queue forever, running `onJob` for each job exactly once. Jobs run + * concurrently, so a long one never blocks the rest of the queue. + */ +export async function processQueue( + onJob: (job: Job) => Promise +): Promise { + await fs.mkdir(dir, { recursive: true }); + const running = new Set(); + + for (;;) { + for (const taskId of await pendingTaskIds()) { + if (running.has(taskId)) continue; + running.add(taskId); + void (async () => { + try { + const job = await claimJob(taskId); + if (job) await onJob(job); + } finally { + running.delete(taskId); + } + })(); + } + await delay(POLL_INTERVAL_MS); + } +} diff --git a/examples/tasks-http/src/task-store.ts b/examples/tasks-http/src/task-store.ts new file mode 100644 index 000000000..d9f8c48c5 --- /dev/null +++ b/examples/tasks-http/src/task-store.ts @@ -0,0 +1,52 @@ +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { createTaskStore, type TaskRecord } from "xmcp"; + +/** + * A file-backed task store. + * + * Task state must live OUTSIDE the request: xmcp's HTTP transport is stateless + * and builds a fresh server per request, so `tasks/get`, `tasks/result`, + * `tasks/list` and `tasks/cancel` can only work if the state is persisted + * somewhere later requests can read. + * + * `createTaskStore` owns all the protocol logic (ids, timestamps, status + * transitions, TTL, pagination); we only provide get/set/delete/list over a + * JSON record. Swap these four methods for Redis, a database, or a KV service. + */ +const dir = path.resolve(process.cwd(), ".tasks"); +const file = (taskId: string) => path.join(dir, `${taskId}.json`); + +export const store = createTaskStore({ + async get(taskId) { + try { + return JSON.parse(await fs.readFile(file(taskId), "utf-8")) as TaskRecord; + } catch { + return null; + } + }, + async set(taskId, record) { + await fs.mkdir(dir, { recursive: true }); + await fs.writeFile(file(taskId), JSON.stringify(record), "utf-8"); + }, + async delete(taskId) { + await fs.rm(file(taskId), { force: true }); + }, + async list() { + try { + const names = (await fs.readdir(dir)).filter((n) => n.endsWith(".json")); + const records = await Promise.all( + names.map((n) => + fs + .readFile(path.join(dir, n), "utf-8") + .then((raw) => JSON.parse(raw) as TaskRecord) + ) + ); + return records; + } catch { + return []; + } + }, +}); + +export default store; diff --git a/examples/tasks-http/src/tools/long-job.ts b/examples/tasks-http/src/tools/long-job.ts new file mode 100644 index 000000000..dfdf09955 --- /dev/null +++ b/examples/tasks-http/src/tools/long-job.ts @@ -0,0 +1,53 @@ +import { z } from "zod"; +import { + type InferSchema, + type ToolExtraArguments, + type ToolMetadata, +} from "xmcp"; +import { enqueue } from "../queue"; + +export const schema = { + label: z.string().describe("A label for the job"), + seconds: z + .number() + .min(1) + .max(30) + .default(3) + .describe("How long the simulated job runs"), +}; + +export const metadata: ToolMetadata = { + name: "long_job", + description: + "Runs a long job as an MCP task. The call returns immediately with a task; " + + "poll tasks/get for status and tasks/result for the final output.", + // Require task augmentation: clients must call this tool with a `task` field. + // This is the right choice for async work that completes out-of-band, so the + // server never blocks waiting for a result it cannot produce synchronously. + taskSupport: "required", + annotations: { + title: "Long job", + readOnlyHint: true, + }, +}; + +export default async function longJob( + { label, seconds }: InferSchema, + extra: ToolExtraArguments +) { + const taskId = extra.task?.taskId; + + // The tool does NOT run the job. It only hands the work off to the queue and + // returns nothing, leaving the task in the "working" state. A separate + // process (src/worker.ts) picks the job up, runs it, and writes the result + // back through the shared task store — which is what later tasks/get and + // tasks/result read. This is the serverless model: enqueue here, execute + // elsewhere. + if (taskId) { + await enqueue({ taskId, label, seconds }); + return; + } + + // Fallback for a non-task invocation (only reachable if taskSupport changes). + return `Job "${label}" ran inline.`; +} diff --git a/examples/tasks-http/src/worker.ts b/examples/tasks-http/src/worker.ts new file mode 100644 index 000000000..4baa91b41 --- /dev/null +++ b/examples/tasks-http/src/worker.ts @@ -0,0 +1,23 @@ +import { store } from "./task-store"; +import { processQueue, type Job } from "./queue"; + +/** + * The worker: a process completely separate from the MCP server. It shares + * nothing with the request that created the task except the on-disk queue and + * task store. Run it in its own terminal (`pnpm worker`). + */ +async function run(job: Job) { + // The delay simulates real work; its duration is caller-supplied job data, + // not a fixed timer. + await new Promise((resolve) => setTimeout(resolve, job.seconds * 1000)); + + // Store a plain string: xmcp coerces it into a CallToolResult when the client + // reads it back, just like returning a string from a normal tool. + await store.storeTaskResult( + job.taskId, + "completed", + `Job "${job.label}" finished after ${job.seconds}s.` + ); +} + +void processQueue(run); diff --git a/examples/tasks-http/tsconfig.json b/examples/tasks-http/tsconfig.json new file mode 100644 index 000000000..369a04ea4 --- /dev/null +++ b/examples/tasks-http/tsconfig.json @@ -0,0 +1,11 @@ +{ + "compilerOptions": { + "target": "es2017", + "module": "commonjs", + "esModuleInterop": true, + "forceConsistentCasingInFileNames": true, + "strict": true, + "skipLibCheck": true + }, + "include": ["xmcp-env.d.ts", "src/**/*.ts"] +} diff --git a/examples/tasks-http/xmcp.config.ts b/examples/tasks-http/xmcp.config.ts new file mode 100644 index 000000000..9eba99a2e --- /dev/null +++ b/examples/tasks-http/xmcp.config.ts @@ -0,0 +1,14 @@ +import { XmcpConfig } from "xmcp"; + +const config: XmcpConfig = { + http: true, + paths: { + prompts: false, + resources: false, + }, + typescript: { + skipTypeCheck: true, + }, +}; + +export default config; diff --git a/packages/xmcp/src/compiler/compiler-context.ts b/packages/xmcp/src/compiler/compiler-context.ts index 4aef399b5..9903e1ae3 100644 --- a/packages/xmcp/src/compiler/compiler-context.ts +++ b/packages/xmcp/src/compiler/compiler-context.ts @@ -22,6 +22,8 @@ interface CompilerContext { resourcePaths: Set; /** Whether the middleware is enabled. */ hasMiddleware: boolean; + /** Whether a task store (src/task-store.ts) is present. */ + hasTaskStore: boolean; /** The parsed config. */ xmcpConfig?: XmcpConfigOutputSchema; /** Client bundles mapping for React (toolName -> bundlePath). */ @@ -36,7 +38,11 @@ export const compilerContext = createContext({ export const compilerContextProvider = async ( initialValue: Omit< CompilerContext, - "toolPaths" | "promptPaths" | "resourcePaths" | "hasMiddleware" + | "toolPaths" + | "promptPaths" + | "resourcePaths" + | "hasMiddleware" + | "hasTaskStore" >, callback: () => void ) => { @@ -47,6 +53,7 @@ export const compilerContextProvider = async ( promptPaths: new Set(), resourcePaths: new Set(), hasMiddleware: false, + hasTaskStore: false, }, () => Promise.resolve(callback()) ); diff --git a/packages/xmcp/src/compiler/generate-import-code.ts b/packages/xmcp/src/compiler/generate-import-code.ts index f3abcfeef..a0ab663ee 100644 --- a/packages/xmcp/src/compiler/generate-import-code.ts +++ b/packages/xmcp/src/compiler/generate-import-code.ts @@ -14,6 +14,7 @@ export function generateImportCode(): string { promptPaths, resourcePaths, hasMiddleware, + hasTaskStore, clientBundles, platforms, } = compilerContext.getContext(); @@ -28,6 +29,7 @@ export function generateImportCode(): string { promptPaths, resourcePaths, hasMiddleware, + hasTaskStore, clientBundles ); } @@ -37,6 +39,7 @@ export function generateImportCode(): string { promptPaths, resourcePaths, hasMiddleware, + hasTaskStore, clientBundles ); } @@ -50,6 +53,7 @@ function generateStaticImportCode( promptPaths: Set, resourcePaths: Set, hasMiddleware: boolean, + hasTaskStore: boolean, clientBundles?: Map ): string { // Generate static import statements at the top @@ -88,6 +92,12 @@ function generateStaticImportCode( middlewareCode = `export const middleware = () => Promise.resolve(_middleware);`; } + let taskStoreCode = ""; + if (hasTaskStore) { + staticImports.push(`import * as _taskStore from "../src/task-store.ts";`); + taskStoreCode = `export const taskStore = () => Promise.resolve(_taskStore);`; + } + // Generate client bundles mapping (empty object if none) const clientBundlesEntries = clientBundles && clientBundles.size > 0 @@ -115,6 +125,8 @@ ${clientBundlesEntries} }; ${middlewareCode} + +${taskStoreCode} `; } @@ -127,6 +139,7 @@ function generateDynamicImportCode( promptPaths: Set, resourcePaths: Set, hasMiddleware: boolean, + hasTaskStore: boolean, clientBundles?: Map ): string { const importToolsCode = Array.from(toolPaths) @@ -157,6 +170,10 @@ function generateDynamicImportCode( ? `export const middleware = () => import("../src/middleware.ts");` : ""; + const importTaskStoreCode = hasTaskStore + ? `export const taskStore = () => import("../src/task-store.ts");` + : ""; + // Generate client bundles mapping (empty object if none) const clientBundlesEntries = clientBundles && clientBundles.size > 0 @@ -183,5 +200,7 @@ ${clientBundlesEntries} }; ${importMiddlewareCode} + +${importTaskStoreCode} `; } diff --git a/packages/xmcp/src/compiler/get-bundler-config/index.ts b/packages/xmcp/src/compiler/get-bundler-config/index.ts index a9a63100f..528e593f8 100644 --- a/packages/xmcp/src/compiler/get-bundler-config/index.ts +++ b/packages/xmcp/src/compiler/get-bundler-config/index.ts @@ -243,6 +243,10 @@ export function getRspackConfig( path.resolve(processFolder, ".xmcp/import-map.js"), "middleware", ], + INJECTED_TASK_STORE: [ + path.resolve(processFolder, ".xmcp/import-map.js"), + "taskStore", + ], }; // add entry points based on config diff --git a/packages/xmcp/src/compiler/index.ts b/packages/xmcp/src/compiler/index.ts index 4d1f7abce..6450bef5c 100644 --- a/packages/xmcp/src/compiler/index.ts +++ b/packages/xmcp/src/compiler/index.ts @@ -199,6 +199,26 @@ export async function compile({ onBuild }: CompileOptions = {}) { }); } + // handle task store (src/task-store.ts) + watcher.watch("./src/task-store.ts", { + onAdd: async () => { + compilerContext.setContext({ + hasTaskStore: true, + }); + if (compilerStarted) { + await generateCode(); + } + }, + onUnlink: async () => { + compilerContext.setContext({ + hasTaskStore: false, + }); + if (compilerStarted) { + await generateCode(); + } + }, + }); + // start compiler watcher.onReady(async () => { let firstBuild = true; diff --git a/packages/xmcp/src/index.ts b/packages/xmcp/src/index.ts index c376be484..70d56759d 100644 --- a/packages/xmcp/src/index.ts +++ b/packages/xmcp/src/index.ts @@ -11,9 +11,21 @@ export type { ToolSchema, ToolOutputSchema, ToolExtraArguments, + TaskSupport, InferSchema, ElicitResult, } from "./types/tool"; +export type { + Task, + TaskStore, + TaskResult, + TaskRecord, + TaskStoreAdapter, + CreateTaskStoreOptions, + CreateTaskOptions, +} from "./types/task"; +export { generateTaskId } from "./types/task"; +export { createTaskStore } from "./utils/create-task-store"; export type { McpClientInfo } from "./types/client-info"; export type { PromptMetadata } from "./types/prompt"; export type { ResourceMetadata } from "./types/resource"; diff --git a/packages/xmcp/src/runtime/utils/server.ts b/packages/xmcp/src/runtime/utils/server.ts index 0cb362705..5663b5ebc 100644 --- a/packages/xmcp/src/runtime/utils/server.ts +++ b/packages/xmcp/src/runtime/utils/server.ts @@ -17,6 +17,9 @@ import { reportResourceLoadIssues, } from "./resource-loader"; import { loadToolModules, reportToolLoadIssues } from "./tool-loader"; +import type { TaskStore } from "@/types/task"; +import type { TaskStore as SdkTaskStore } from "@modelcontextprotocol/sdk/experimental/tasks/interfaces"; +import { coerceToolResponse } from "./transformers/tool"; export type ToolFile = { metadata: ToolMetadata; @@ -54,17 +57,64 @@ export const injectedResources = INJECTED_RESOURCES as Record< export const INJECTED_CONFIG = SERVER_INFO as Implementation & { instructions?: string }; +const injectedTaskStore = INJECTED_TASK_STORE; + +/** Loads the user-provided task store (src/task-store.ts), if present. */ +export async function loadTaskStore(): Promise { + if (!injectedTaskStore) return undefined; + const module = await injectedTaskStore(); + return module?.default ?? module?.taskStore; +} + +/** + * Wraps a task store so `getTaskResult` applies the same string/number → result + * coercion xmcp does on a synchronous tool return. An out-of-band worker can + * store a plain string and the client still receives a valid CallToolResult. + * Only primitives are coerced; objects/arrays (already-valid results, failed + * envelopes, etc.) pass through unchanged. + */ +export function wrapTaskStore(store: TaskStore): SdkTaskStore { + return { + createTask: (...args) => store.createTask(...args), + getTask: (...args) => store.getTask(...args), + storeTaskResult: (...args) => store.storeTaskResult(...args), + updateTaskStatus: (...args) => store.updateTaskStatus(...args), + listTasks: (...args) => store.listTasks(...args), + getTaskResult: async (taskId, sessionId) => { + const result = await store.getTaskResult(taskId, sessionId); + if (typeof result === "string" || typeof result === "number") { + return coerceToolResponse(result, {}, undefined, undefined, "task"); + } + return result; + }, + }; +} + /* Loads all modules and injects them into the server */ // would be better as a class and use dependency injection perhaps export async function configureServer( server: McpServer, toolModules: Map, promptModules: Map, - resourceModules: Map + resourceModules: Map, + hasTaskStore = false ): Promise { uIResourceRegistry.clear(); - addToolsToServer(server, toolModules); + if (hasTaskStore) { + // Advertise task support so clients may augment tools/call with a task. + // Security note: stateless HTTP has no requestor identity, so task IDs are + // the only access control. Use cryptographically secure IDs in the store. + server.server.registerCapabilities({ + tasks: { + list: {}, + cancel: {}, + requests: { tools: { call: {} } }, + }, + }); + } + + addToolsToServer(server, toolModules, hasTaskStore); addPromptsToServer(server, promptModules); addResourcesToServer(server, resourceModules); return server; @@ -94,14 +144,27 @@ export async function loadResources() { export async function createServer() { const { instructions, ...serverInfo } = INJECTED_CONFIG; - const server = new McpServer(serverInfo, { instructions }); const toolModulesPromise = loadTools(); const promptModulesPromise = loadPrompts(); const resourceModulesPromise = loadResources(); - const [toolModules, promptModules, resourceModules] = await Promise.all([ - toolModulesPromise, - promptModulesPromise, - resourceModulesPromise, - ]); - return configureServer(server, toolModules, promptModules, resourceModules); + const taskStorePromise = loadTaskStore(); + const [toolModules, promptModules, resourceModules, taskStore] = + await Promise.all([ + toolModulesPromise, + promptModulesPromise, + resourceModulesPromise, + taskStorePromise, + ]); + // Passing a task store enables the SDK's tasks/* request handlers. + const server = new McpServer(serverInfo, { + instructions, + taskStore: taskStore ? wrapTaskStore(taskStore) : undefined, + }); + return configureServer( + server, + toolModules, + promptModules, + resourceModules, + Boolean(taskStore) + ); } diff --git a/packages/xmcp/src/runtime/utils/tools.ts b/packages/xmcp/src/runtime/utils/tools.ts index b61049b20..aec2d2abb 100644 --- a/packages/xmcp/src/runtime/utils/tools.ts +++ b/packages/xmcp/src/runtime/utils/tools.ts @@ -3,7 +3,11 @@ import { z } from "zod"; import { ZodRawShape } from "zod/v3"; import { ToolFile } from "./server"; import { ToolMetadata } from "@/types/tool"; -import { transformToolHandler } from "./transformers/tool"; +import { + transformToolHandler, + createToolExtraArguments, + coerceToolResponse, +} from "./transformers/tool"; import { isReactFile } from "./react"; import { uIResourceRegistry } from "./ext-apps-registry"; import { flattenMeta, hasUIMeta } from "./ui/flatten-meta"; @@ -41,7 +45,8 @@ export function ensureAnnotations(toolConfig: Pick + toolModules: Map, + hasTaskStore = false ): McpServer { toolModules.forEach((toolModule, path) => { const defaultName = pathToName(path); @@ -171,12 +176,72 @@ export function addToolsToServer( _meta: flattenedToolMeta, // Use flattened metadata for MCP protocol }; - // server as any prevents infinite type recursion - (server as any).registerTool( - toolConfig.name, - toolConfigFormatted, - transformedHandler - ); + const taskSupport = toolConfig.taskSupport; + const isTaskTool = taskSupport === "optional" || taskSupport === "required"; + + if (isTaskTool) { + if (!hasTaskStore) { + throw new Error( + `Tool "${toolConfig.name}" declares taskSupport "${taskSupport}" but no task store was found. ` + + `Add a src/task-store.ts that exports a TaskStore implementation.` + ); + } + + // Register as a task-augmented tool. The SDK exposes tasks/get, + // tasks/result, tasks/list and tasks/cancel and routes task-augmented + // tools/call requests to this handler triple. + (server as any).experimental.tasks.registerToolTask( + toolConfig.name, + { ...toolConfigFormatted, execution: { taskSupport } }, + { + // Create the task record, let the tool kick off (or run) the work, + // and return immediately. The tool either returns a result + // synchronously (stored as the terminal result) or returns nothing + // and completes the task later via the store from its own worker. + createTask: async (args: any, extra: any) => { + const task = await extra.taskStore.createTask({ + ttl: extra.taskRequestedTtl ?? undefined, + }); + + const toolExtra = createToolExtraArguments(extra); + toolExtra.task = { taskId: task.taskId }; + + let response: any = handler(args, toolExtra); + if (response instanceof Promise) { + response = await response; + } + + if (response !== undefined && response !== null) { + const result = coerceToolResponse( + response, + args, + meta, + toolOutputSchema, + toolConfig.name + ); + await extra.taskStore.storeTaskResult( + task.taskId, + result.isError ? "failed" : "completed", + result + ); + } + + return { task }; + }, + getTask: (_args: any, extra: any) => + extra.taskStore.getTask(extra.taskId), + getTaskResult: (_args: any, extra: any) => + extra.taskStore.getTaskResult(extra.taskId), + } + ); + } else { + // server as any prevents infinite type recursion + (server as any).registerTool( + toolConfig.name, + toolConfigFormatted, + transformedHandler + ); + } }); return server; diff --git a/packages/xmcp/src/runtime/utils/transformers/tool.ts b/packages/xmcp/src/runtime/utils/transformers/tool.ts index 7cee5c17b..81912cec2 100644 --- a/packages/xmcp/src/runtime/utils/transformers/tool.ts +++ b/packages/xmcp/src/runtime/utils/transformers/tool.ts @@ -73,7 +73,7 @@ function hasUIMeta(meta?: Record): boolean { ); } -function createToolExtraArguments( +export function createToolExtraArguments( extra: RequestHandlerExtra ): ToolExtraArguments { let clientInfo = undefined; @@ -137,6 +137,28 @@ export function transformToolHandler( response = await response; } + return coerceToolResponse(response, args, meta, outputSchema, toolName); + }; +} + +/** + * Normalizes a user tool handler's return value into a valid `CallToolResult`. + * + * Shared by {@link transformToolHandler} and the task-tool wrapper so that + * string/number coercion, structured-content handling, widget metadata, and + * outputSchema validation behave identically whether a tool runs synchronously + * or as a task. + * + * @throws Error if the value cannot be turned into a valid CallToolResult. + */ +export function coerceToolResponse( + response: any, + args: ZodRawShape, + meta?: Record, + outputSchema?: ZodRawShape, + toolName = "unknown-tool" +): CallToolResult { + { if (typeof response === "string" || typeof response === "number") { if (outputSchema) { const outputSchemaEntries = Object.entries(outputSchema); diff --git a/packages/xmcp/src/types/injected.d.ts b/packages/xmcp/src/types/injected.d.ts index 281fe0a71..c7b2a6223 100644 --- a/packages/xmcp/src/types/injected.d.ts +++ b/packages/xmcp/src/types/injected.d.ts @@ -25,6 +25,13 @@ declare const INJECTED_MIDDLEWARE: }>) | undefined; +declare const INJECTED_TASK_STORE: + | (() => Promise<{ + default?: import("./task").TaskStore; + taskStore?: import("./task").TaskStore; + }>) + | undefined; + // ─── DefinePlugin — config objects ──────────────────────────────────────────── declare const HTTP_CONFIG: Omit< diff --git a/packages/xmcp/src/types/task.ts b/packages/xmcp/src/types/task.ts new file mode 100644 index 000000000..6aad32a60 --- /dev/null +++ b/packages/xmcp/src/types/task.ts @@ -0,0 +1,92 @@ +/** + * Public contract for MCP tasks support in xmcp. + * + * Tasks let a tool call run as a long-running, pollable operation (the MCP + * "tasks" utility). xmcp keeps this fully stateless: it never holds task state + * in server memory. Instead, you provide a `TaskStore` implementation backed by + * external storage (Redis, a database, a KV store, etc.) in `src/task-store.ts`. + * + * The underlying protocol handlers (`tasks/get`, `tasks/result`, `tasks/list`, + * `tasks/cancel`) and the `tools/call` task negotiation are implemented by the + * MCP SDK; xmcp wires your store into them. + * + * Security note: in unauthenticated stateless HTTP there is no requestor + * identity to bind tasks to, so task IDs are the only access control. Generate + * cryptographically secure IDs (see {@link generateTaskId}) and prefer short + * TTLs. If you need stronger isolation, add authentication. + */ +import type { + TaskStore as SdkTaskStore, + CreateTaskOptions, +} from "@modelcontextprotocol/sdk/experimental/tasks/interfaces"; +import type { Result, Task } from "@modelcontextprotocol/sdk/types"; + +export type { Task, CreateTaskOptions }; + +/** + * A task result a tool or worker may produce. A bare `string` or `number` is + * coerced into a `CallToolResult` when the client reads it back (same as + * returning a string from a normal tool). An object must already be a valid + * result (`content` / `structuredContent` / `isError`). + */ +export type TaskResult = Result | string | number; + +/** + * Contract for external task storage. Implement this in `src/task-store.ts`. + * + * Identical to the MCP SDK's task store, except `storeTaskResult` accepts — and + * `getTaskResult` may return — a bare `string`/`number`, which xmcp coerces. + */ +export interface TaskStore + extends Omit { + storeTaskResult( + taskId: string, + status: "completed" | "failed", + result: TaskResult, + sessionId?: string + ): Promise; + getTaskResult(taskId: string, sessionId?: string): Promise; +} + +/** + * A persisted task: the public `Task` plus the stored result. This is the + * opaque record a {@link TaskStoreAdapter} reads and writes; xmcp owns its + * shape and handles the protocol logic around it. + */ +export type TaskRecord = Task & { result?: TaskResult }; + +/** + * Minimal persistence primitives for {@link createTaskStore}. Each method just + * stores or loads a JSON-serializable {@link TaskRecord} by id — no protocol + * logic. Maps directly onto any KV or database (Redis, Upstash, a table, …). + */ +export interface TaskStoreAdapter { + /** Load a record by id, or `null`/`undefined` if it does not exist. */ + get(taskId: string): Promise; + /** Create or overwrite a record. */ + set(taskId: string, record: TaskRecord): Promise; + /** Remove a record. */ + delete(taskId: string): Promise; + /** Return every stored record (used only for `tasks/list`). */ + list(): Promise; +} + +/** Options for {@link createTaskStore}. All optional. */ +export interface CreateTaskStoreOptions { + /** Task lifetime in ms when the requestor does not ask for one. `null` (default) = unlimited. */ + defaultTtlMs?: number | null; + /** Suggested client poll interval in ms, surfaced in task responses. Default `1000`. */ + pollIntervalMs?: number; + /** Page size for `tasks/list`. Default `50`. */ + pageSize?: number; +} + +/** + * Generates a cryptographically secure task identifier. + * + * Convenience for `TaskStore.createTask` implementations. Uses the platform Web + * Crypto API, available on Node.js 20+ and Cloudflare Workers. + */ +export function generateTaskId(): string { + return globalThis.crypto.randomUUID(); +} diff --git a/packages/xmcp/src/types/tool.ts b/packages/xmcp/src/types/tool.ts index d5bcfe0e0..33312b06e 100644 --- a/packages/xmcp/src/types/tool.ts +++ b/packages/xmcp/src/types/tool.ts @@ -18,6 +18,16 @@ export interface ToolAnnotations { [key: string]: any; } +/** + * Whether a tool can be invoked as an MCP task (long-running, pollable). + * - "forbidden" (default): the tool is never task-augmented. + * - "optional": clients may invoke it as a task or as a normal call. + * - "required": clients must invoke it as a task. + * + * Requires a `src/task-store.ts` task store implementation to be present. + */ +export type TaskSupport = "required" | "optional" | "forbidden"; + export interface ToolMetadata { /** Unique identifier for the tool */ name: string; @@ -25,6 +35,12 @@ export interface ToolMetadata { description: string; /** Optional hints about tool behavior */ annotations?: ToolAnnotations; + /** + * Opt the tool into MCP task execution. When set to "optional" or + * "required", xmcp registers the tool as a task-augmented tool. Requires a + * `src/task-store.ts` task store implementation. + */ + taskSupport?: TaskSupport; /** Metadata for the tool. */ _meta?: { ui?: UIMetadata; @@ -157,6 +173,16 @@ export interface ToolExtraArguments { /** The JSON-RPC ID of the request being handled */ requestId: string | number; + /** + * Present only when the tool is being executed as an MCP task. Use + * `task.taskId` to key external work (e.g. enqueue a job) and to write the + * result back to your task store from a worker. + */ + task?: { + /** The receiver-generated unique identifier for this task */ + taskId: string; + }; + /** The original HTTP request information */ requestInfo?: { /** The headers of the request */ diff --git a/packages/xmcp/src/utils/create-task-store.ts b/packages/xmcp/src/utils/create-task-store.ts new file mode 100644 index 000000000..1cc3a2e38 --- /dev/null +++ b/packages/xmcp/src/utils/create-task-store.ts @@ -0,0 +1,133 @@ +import { + generateTaskId, + type CreateTaskStoreOptions, + type Task, + type TaskRecord, + type TaskStore, + type TaskStoreAdapter, +} from "../types/task"; + +const TERMINAL_STATUSES: ReadonlySet = new Set([ + "completed", + "failed", + "cancelled", +]); + +function isExpired(record: TaskRecord): boolean { + if (record.ttl == null) return false; + return Date.now() > new Date(record.createdAt).getTime() + record.ttl; +} + +/** Strips the stored result so only the public `Task` is returned. */ +function toPublicTask({ result: _result, ...task }: TaskRecord): Task { + return task; +} + +function encodeCursor(offset: number): string { + return Buffer.from(String(offset)).toString("base64"); +} + +function decodeCursor(cursor: string | undefined): number { + if (!cursor) return 0; + const offset = Number(Buffer.from(cursor, "base64").toString()); + return Number.isFinite(offset) && offset >= 0 ? offset : 0; +} + +/** + * Builds a full {@link TaskStore} from minimal persistence primitives. + * + * The adapter only stores and loads {@link TaskRecord}s; xmcp handles all of + * the protocol logic: id generation, timestamps, the working → terminal status + * machine (with terminal guards), TTL expiry, the public-task projection, and + * cursor pagination for `tasks/list`. + * + * ```ts + * export default createTaskStore({ + * get: (id) => kv.get(id), + * set: (id, record) => kv.set(id, record), + * delete: (id) => kv.del(id), + * list: () => kv.values(), + * }); + * ``` + * + * Need cursor pagination pushed into the query, or anything else bespoke? + * Implement {@link TaskStore} directly instead. + */ +export function createTaskStore( + adapter: TaskStoreAdapter, + options: CreateTaskStoreOptions = {} +): TaskStore { + const defaultTtlMs = options.defaultTtlMs ?? null; + const pollIntervalMs = options.pollIntervalMs ?? 1000; + const pageSize = options.pageSize ?? 50; + + /** Loads a live record, evicting and dropping it if its TTL has elapsed. */ + async function readLive(taskId: string): Promise { + const record = await adapter.get(taskId); + if (!record) return null; + if (isExpired(record)) { + await adapter.delete(taskId); + return null; + } + return record; + } + + return { + async createTask(taskParams) { + const now = new Date().toISOString(); + const record: TaskRecord = { + taskId: generateTaskId(), + status: "working", + ttl: taskParams.ttl ?? defaultTtlMs, + createdAt: now, + lastUpdatedAt: now, + pollInterval: taskParams.pollInterval ?? pollIntervalMs, + }; + await adapter.set(record.taskId, record); + return toPublicTask(record); + }, + + async getTask(taskId) { + const record = await readLive(taskId); + return record ? toPublicTask(record) : null; + }, + + async storeTaskResult(taskId, status, result) { + const record = await readLive(taskId); + if (!record || TERMINAL_STATUSES.has(record.status)) return; + record.status = status; + record.result = result; + record.lastUpdatedAt = new Date().toISOString(); + await adapter.set(taskId, record); + }, + + async getTaskResult(taskId) { + const record = await readLive(taskId); + return record?.result ?? {}; + }, + + async updateTaskStatus(taskId, status, statusMessage) { + const record = await readLive(taskId); + if (!record || TERMINAL_STATUSES.has(record.status)) return; + record.status = status; + if (statusMessage !== undefined) record.statusMessage = statusMessage; + record.lastUpdatedAt = new Date().toISOString(); + await adapter.set(taskId, record); + }, + + async listTasks(cursor) { + const all = (await adapter.list()) + .filter((record) => !isExpired(record)) + .sort((a, b) => a.createdAt.localeCompare(b.createdAt)); + + const start = decodeCursor(cursor); + const page = all.slice(start, start + pageSize); + const next = start + pageSize; + + return { + tasks: page.map(toPublicTask), + nextCursor: next < all.length ? encodeCursor(next) : undefined, + }; + }, + }; +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e0138e4c2..e76c81f7e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -716,6 +716,25 @@ importers: specifier: ^4.0.10 version: 4.1.13 + examples/tasks-http: + dependencies: + xmcp: + specifier: workspace:* + version: link:../../packages/xmcp + zod: + specifier: ^4.0.10 + version: 4.4.3 + devDependencies: + '@types/node': + specifier: ^20.19.26 + version: 20.19.26 + tsx: + specifier: ^4.21.0 + version: 4.21.0 + typescript: + specifier: ^5.9.3 + version: 5.9.3 + examples/template-config: dependencies: xmcp: @@ -13688,8 +13707,8 @@ snapshots: '@babel/parser': 7.29.2 eslint: 9.39.1(jiti@2.6.1) hermes-parser: 0.25.1 - zod: 4.1.13 - zod-validation-error: 4.0.2(zod@4.1.13) + zod: 4.4.3 + zod-validation-error: 4.0.2(zod@4.4.3) transitivePeerDependencies: - supports-color @@ -16014,7 +16033,7 @@ snapshots: openid-client@6.8.1: dependencies: - jose: 6.1.3 + jose: 6.2.3 oauth4webapi: 3.8.3 optionator@0.9.4: @@ -18167,9 +18186,9 @@ snapshots: dependencies: zod: 4.4.3 - zod-validation-error@4.0.2(zod@4.1.13): + zod-validation-error@4.0.2(zod@4.4.3): dependencies: - zod: 4.1.13 + zod: 4.4.3 zod@3.25.76: {}