docs(servers): document resources/subscribe serving; exercise subscriptions in the resources example (#2413)
This commit is contained in:
Felix Weinberger
GitHub
parent
35f6856a17
commit
ddf6cc1d73
7 changed files with 221 additions and 11 deletions
|
|
@ -87,7 +87,7 @@ On stdio, [`serveStdio`](../serving/stdio.md) routes the instance's own `send*Li
|
|||
:::
|
||||
|
||||
::: info Coming from 2025-era subscriptions
|
||||
A 2025-era connection delivers per-resource updates without a listen stream — [Subscriptions](../clients/subscriptions.md#fall-back-to-legacy-per-resource-subscribe) covers that path, and [Protocol versions](../protocol-versions.md) the era split.
|
||||
A 2025-era connection delivers per-resource updates without a listen stream — [Resources](./resources.md#serve-per-resource-subscriptions) covers the server's subscription bookkeeping, [Subscriptions](../clients/subscriptions.md#fall-back-to-legacy-per-resource-subscribe) the client call, and [Protocol versions](../protocol-versions.md) the era split.
|
||||
:::
|
||||
|
||||
## Pick an event bus for multi-process deployments
|
||||
|
|
|
|||
|
|
@ -212,6 +212,51 @@ server.sendResourceListChanged();
|
|||
|
||||
The notification tells connected clients to call `resources/list` again. A change to one resource's content is a different signal, `notifications/resources/updated` — [Notifications](./notifications.md) covers both.
|
||||
|
||||
## Serve per-resource subscriptions
|
||||
|
||||
A 2025-era client opts into `notifications/resources/updated` for one URI with `resources/subscribe`. The SDK routes the verb; the bookkeeping is yours: advertise the capability, track the URIs per connection, and send the notification to subscribers only.
|
||||
|
||||
```ts source="../../examples/guides/servers/resources.examples.ts#sendResourceUpdated_subscribers"
|
||||
let deployStatus = 'idle';
|
||||
|
||||
const deploys = new McpServer({ name: 'deploys', version: '1.0.0' }, { capabilities: { resources: { subscribe: true } } });
|
||||
|
||||
deploys.registerResource(
|
||||
'deploy-status',
|
||||
'deploys://status',
|
||||
{ description: 'The current deploy state', mimeType: 'text/plain' },
|
||||
async uri => ({ contents: [{ uri: uri.href, text: deployStatus }] })
|
||||
);
|
||||
|
||||
// The SDK routes the two verbs; which URIs this connection watches is yours to track.
|
||||
const subscribedUris = new Set<string>();
|
||||
deploys.server.setRequestHandler('resources/subscribe', request => {
|
||||
subscribedUris.add(request.params.uri);
|
||||
return {};
|
||||
});
|
||||
deploys.server.setRequestHandler('resources/unsubscribe', request => {
|
||||
subscribedUris.delete(request.params.uri);
|
||||
return {};
|
||||
});
|
||||
|
||||
async function setDeployStatus(status: string): Promise<void> {
|
||||
deployStatus = status;
|
||||
if (subscribedUris.has('deploys://status')) {
|
||||
await deploys.server.sendResourceUpdated({ uri: 'deploys://status' });
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The `Set` belongs to one server instance, and each connection gets its own instance from your factory — a subscription never leaks across connections. Send `resources/updated` only to connections that subscribed; unsolicited per-resource updates are wrong on 2025-era connections.
|
||||
|
||||
The pattern needs a connection that outlives the subscribe call: over stdio (and any sessionful wiring) the instance and its `Set` live as long as the connection. Behind `createMcpHandler`'s stateless legacy fallback each POST gets a fresh instance, so `resources/subscribe` succeeds and the `Set` is discarded with it — no update can ever be delivered on that posture. [Support legacy clients](../serving/legacy-clients.md) covers the serving postures.
|
||||
|
||||
::: info
|
||||
On [2026-07-28](../protocol-versions.md) connections the verb does not exist: clients name resource URIs in their `subscriptions/listen` filter, and the entry filters delivery itself — `serveStdio` routes the instance's own `sendResourceUpdated` call onto matching streams, and `createMcpHandler` delivers what you publish on its notifier ([Notifications](./notifications.md#publish-a-resource-update-through-the-handler)).
|
||||
:::
|
||||
|
||||
A dual-era server therefore still calls `sendResourceUpdated` on 2026-07-28 connections, where the subscribe set is always empty — gate on the connection's era as well as the set. The [`resources` example](https://github.com/modelcontextprotocol/typescript-sdk/tree/main/examples/resources) guards with `reqCtx.era === 'modern' || subscribedUris.has(uri)` in its factory and runs as a self-verifying pair: delivery is asserted over stdio on both eras and over HTTP on the 2026-07-28 listen path; the stateless legacy HTTP leg asserts only that the subscribe calls succeed.
|
||||
|
||||
## Recap
|
||||
|
||||
- `registerResource(name, uri, config, readCallback)` registers a resource at a fixed URI.
|
||||
|
|
@ -220,3 +265,4 @@ The notification tells connected clients to call `resources/list` again. A chang
|
|||
- A template's `list` callback is what makes its instances appear in `resources/list`.
|
||||
- Resolve file-backed paths to their real location and reject anything outside the root before reading.
|
||||
- Registration changes emit `notifications/resources/list_changed` automatically.
|
||||
- `resources/subscribe` bookkeeping is the server's: advertise `resources: { subscribe: true }`, track URIs per connection, send `resources/updated` to subscribers only.
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ The one exception to the generic commands is the reference pair: [`cli-client/`]
|
|||
| ------------------------------------- | ------------------------------------------------------------------------ |
|
||||
| [`tools/`](./tools/README.md) | Register tools, infer input/output schemas, call them, structured output |
|
||||
| [`prompts/`](./prompts/README.md) | Prompts + argument completion |
|
||||
| [`resources/`](./resources/README.md) | Static + templated resources, list/read |
|
||||
| [`resources/`](./resources/README.md) | Static + templated resources, list/read, era-split subscriptions |
|
||||
| [`dual-era/`](./dual-era/README.md) | One factory, both protocol eras, both transports |
|
||||
|
||||
## Feature stories
|
||||
|
|
|
|||
|
|
@ -117,6 +117,41 @@ server.registerResource(
|
|||
// Imported dynamically so the page's lead region stays self-contained.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// "Serve per-resource subscriptions"
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
//#region sendResourceUpdated_subscribers
|
||||
let deployStatus = 'idle';
|
||||
|
||||
const deploys = new McpServer({ name: 'deploys', version: '1.0.0' }, { capabilities: { resources: { subscribe: true } } });
|
||||
|
||||
deploys.registerResource(
|
||||
'deploy-status',
|
||||
'deploys://status',
|
||||
{ description: 'The current deploy state', mimeType: 'text/plain' },
|
||||
async uri => ({ contents: [{ uri: uri.href, text: deployStatus }] })
|
||||
);
|
||||
|
||||
// The SDK routes the two verbs; which URIs this connection watches is yours to track.
|
||||
const subscribedUris = new Set<string>();
|
||||
deploys.server.setRequestHandler('resources/subscribe', request => {
|
||||
subscribedUris.add(request.params.uri);
|
||||
return {};
|
||||
});
|
||||
deploys.server.setRequestHandler('resources/unsubscribe', request => {
|
||||
subscribedUris.delete(request.params.uri);
|
||||
return {};
|
||||
});
|
||||
|
||||
async function setDeployStatus(status: string): Promise<void> {
|
||||
deployStatus = status;
|
||||
if (subscribedUris.has('deploys://status')) {
|
||||
await deploys.server.sendResourceUpdated({ uri: 'deploys://status' });
|
||||
}
|
||||
}
|
||||
//#endregion sendResourceUpdated_subscribers
|
||||
|
||||
const { Client, InMemoryTransport } = await import('@modelcontextprotocol/client');
|
||||
|
||||
const client = new Client({ name: 'resources-docs-harness', version: '1.0.0' });
|
||||
|
|
@ -153,5 +188,29 @@ if (uris.some(uri => uri.startsWith('users://')) || !uris.includes('teams://core
|
|||
server.sendResourceListChanged();
|
||||
//#endregion sendResourceListChanged
|
||||
|
||||
// "Serve per-resource subscriptions" — a 2025-era client subscribes, the status
|
||||
// changes, exactly one notifications/resources/updated arrives, and none after
|
||||
// unsubscribe.
|
||||
const watcher = new Client({ name: 'resources-docs-watcher', version: '1.0.0' }, { versionNegotiation: { mode: 'legacy' } });
|
||||
const [watcherTransport, deploysTransport] = InMemoryTransport.createLinkedPair();
|
||||
const updates: string[] = [];
|
||||
watcher.setNotificationHandler('notifications/resources/updated', notification => {
|
||||
updates.push(notification.params.uri);
|
||||
});
|
||||
await deploys.connect(deploysTransport);
|
||||
await watcher.connect(watcherTransport);
|
||||
|
||||
await watcher.subscribeResource({ uri: 'deploys://status' });
|
||||
await setDeployStatus('deploying');
|
||||
await watcher.unsubscribeResource({ uri: 'deploys://status' });
|
||||
await setDeployStatus('done');
|
||||
await new Promise(resolve => setTimeout(resolve, 50));
|
||||
console.log('updates:', updates);
|
||||
if (updates.length !== 1 || updates[0] !== 'deploys://status') {
|
||||
throw new Error(`resources.md subscription claim failed: ${JSON.stringify(updates)}`);
|
||||
}
|
||||
await watcher.close();
|
||||
await deploys.close();
|
||||
|
||||
await client.close();
|
||||
await server.close();
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
# resources
|
||||
|
||||
Direct resources (a fixed URI string) and templated resources (`ResourceTemplate('greeting://{name}')`). The client lists both, reads the direct config, and reads a templated greeting.
|
||||
Direct resources (a fixed URI string), templated resources (`ResourceTemplate('greeting://{name}')`), and per-resource subscriptions. The client lists both kinds, reads the direct config and a templated greeting, then subscribes to `counter://value` — `subscriptions/listen` on 2026-07-28, `resources/subscribe` on 2025 — calls the `increment` tool, and asserts the `notifications/resources/updated` it produces. Per-request legacy HTTP has no delivery channel, so that leg skips the delivery assertion.
|
||||
|
||||
```bash
|
||||
pnpm tsx examples/resources/client.ts
|
||||
|
|
|
|||
|
|
@ -1,5 +1,11 @@
|
|||
/**
|
||||
* Drives the resources example: list, list templates, read direct + templated.
|
||||
* Drives the resources example: list, list templates, read direct + templated,
|
||||
* then subscribe to `counter://value` and assert the update notification. The
|
||||
* subscription sender is era-split — `subscriptions/listen` on 2026-07-28,
|
||||
* `resources/subscribe` on 2025 — while the notification handler is one
|
||||
* registration either way. Per-request legacy HTTP has no channel to deliver
|
||||
* notifications, so that leg asserts the calls succeed and skips the delivery
|
||||
* assertion.
|
||||
*/
|
||||
import { check, parseExampleArgs, siblingPath } from '@mcp-examples/shared';
|
||||
import { Client, StreamableHTTPClientTransport } from '@modelcontextprotocol/client';
|
||||
|
|
@ -18,6 +24,7 @@ await (transport === 'stdio'
|
|||
|
||||
const list = await client.listResources();
|
||||
check.ok(list.resources.some(r => r.uri === 'config://app'));
|
||||
check.ok(list.resources.some(r => r.uri === 'counter://value'));
|
||||
|
||||
const templates = await client.listResourceTemplates();
|
||||
check.ok(templates.resourceTemplates.some(t => t.uriTemplate === 'greeting://{name}'));
|
||||
|
|
@ -30,4 +37,52 @@ const hello = await client.readResource({ uri: 'greeting://world' });
|
|||
const helloContent = hello.contents[0];
|
||||
check.equal(helloContent && 'text' in helloContent ? helloContent.text : '', 'Hello, world!');
|
||||
|
||||
// --- Subscriptions ---------------------------------------------------------
|
||||
|
||||
// One handler serves both delivery paths: the 2026-07-28 listen stream and a
|
||||
// 2025-era connection's unsolicited notification dispatch the same way.
|
||||
let resolveUpdated: ((uri: string) => void) | undefined;
|
||||
client.setNotificationHandler('notifications/resources/updated', notification => {
|
||||
resolveUpdated?.(notification.params.uri);
|
||||
});
|
||||
|
||||
// Per-request legacy HTTP answers each POST in isolation: subscribing succeeds,
|
||||
// but there is no stream to deliver the notification on.
|
||||
const deliverable = !(era === 'legacy' && transport === 'http');
|
||||
|
||||
const updated = new Promise<string>(resolve => {
|
||||
resolveUpdated = resolve;
|
||||
});
|
||||
|
||||
const subscription = era === 'modern' ? await client.listen({ resourceSubscriptions: ['counter://value'] }) : undefined;
|
||||
if (era === 'legacy') {
|
||||
await client.subscribeResource({ uri: 'counter://value' });
|
||||
}
|
||||
|
||||
const bumped = await client.callTool({ name: 'increment', arguments: {} });
|
||||
check.ok(!bumped.isError);
|
||||
const bumpedContent = bumped.content[0];
|
||||
const bumpedValue = bumpedContent && bumpedContent.type === 'text' ? bumpedContent.text : '';
|
||||
|
||||
if (deliverable) {
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
const timeout = new Promise<never>((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error('no resources/updated within 8s')), 8000);
|
||||
});
|
||||
const updatedUri = await Promise.race([updated, timeout]).finally(() => clearTimeout(timer));
|
||||
check.equal(updatedUri, 'counter://value');
|
||||
|
||||
// The resource and the tool observe the same state: the re-read matches the
|
||||
// value increment returned (not a literal, so a long-lived server re-runs).
|
||||
const counter = await client.readResource({ uri: 'counter://value' });
|
||||
const counterContent = counter.contents[0];
|
||||
check.equal(counterContent && 'text' in counterContent ? counterContent.text : '', bumpedValue);
|
||||
}
|
||||
|
||||
if (subscription) {
|
||||
await subscription.close();
|
||||
} else if (era === 'legacy') {
|
||||
await client.unsubscribeResource({ uri: 'counter://value' });
|
||||
}
|
||||
|
||||
await client.close();
|
||||
|
|
|
|||
|
|
@ -1,19 +1,33 @@
|
|||
/**
|
||||
* Resources primitive — direct + templated.
|
||||
* Resources primitive — direct, templated, and subscribable.
|
||||
*
|
||||
* `McpServer.registerResource` accepts either a fixed URI string (direct
|
||||
* resource) or a `ResourceTemplate` (URI template with substitution). One
|
||||
* binary, either transport — selected from argv below.
|
||||
* resource) or a `ResourceTemplate` (URI template with substitution). The
|
||||
* `counter://value` resource is mutable: the `increment` tool bumps it and
|
||||
* announces the change — in-band to this connection (2025-era subscribers
|
||||
* tracked per connection, 2026-07-28 listen streams routed by the entry), and
|
||||
* on the handler's notifier for clients on other requests. One binary, either
|
||||
* transport — selected from argv below.
|
||||
*/
|
||||
import { createServer } from 'node:http';
|
||||
|
||||
import { parseExampleArgs } from '@mcp-examples/shared';
|
||||
import { toNodeHandler } from '@modelcontextprotocol/node';
|
||||
import type { McpRequestContext } from '@modelcontextprotocol/server';
|
||||
import { createMcpHandler, McpServer, ResourceTemplate } from '@modelcontextprotocol/server';
|
||||
import { serveStdio } from '@modelcontextprotocol/server/stdio';
|
||||
|
||||
function buildServer(): McpServer {
|
||||
const server = new McpServer({ name: 'resources-example', version: '1.0.0' });
|
||||
const COUNTER_URI = 'counter://value';
|
||||
|
||||
// The counter is application state, shared by every connection; which URIs a
|
||||
// connection subscribed to is connection state, tracked inside buildServer.
|
||||
let counter = 0;
|
||||
|
||||
function buildServer(reqCtx: McpRequestContext, publishUpdated?: (uri: string) => void): McpServer {
|
||||
const server = new McpServer(
|
||||
{ name: 'resources-example', version: '1.0.0' },
|
||||
{ capabilities: { resources: { subscribe: true, listChanged: true } } }
|
||||
);
|
||||
|
||||
// A direct resource at a fixed URI.
|
||||
server.registerResource(
|
||||
|
|
@ -31,16 +45,52 @@ function buildServer(): McpServer {
|
|||
async (uri, vars) => ({ contents: [{ uri: uri.href, text: `Hello, ${vars.name}!` }] })
|
||||
);
|
||||
|
||||
// A mutable resource — the subscription story's subject.
|
||||
server.registerResource(
|
||||
'counter',
|
||||
COUNTER_URI,
|
||||
{ mimeType: 'text/plain', description: 'A number the increment tool bumps' },
|
||||
async uri => ({ contents: [{ uri: uri.href, mimeType: 'text/plain', text: String(counter) }] })
|
||||
);
|
||||
|
||||
// resources/subscribe bookkeeping is the application's: the SDK routes the
|
||||
// two verbs, and which URIs THIS connection watches lives here.
|
||||
const subscribedUris = new Set<string>();
|
||||
server.server.setRequestHandler('resources/subscribe', request => {
|
||||
subscribedUris.add(request.params.uri);
|
||||
return {};
|
||||
});
|
||||
server.server.setRequestHandler('resources/unsubscribe', request => {
|
||||
subscribedUris.delete(request.params.uri);
|
||||
return {};
|
||||
});
|
||||
|
||||
server.registerTool('increment', { description: `Bump ${COUNTER_URI} by one` }, async () => {
|
||||
counter += 1;
|
||||
if (publishUpdated) {
|
||||
// Per-request serving: this instance answers one request and is gone,
|
||||
// so the change is published on the entry's notifier for the listen
|
||||
// streams other requests hold open.
|
||||
publishUpdated(COUNTER_URI);
|
||||
} else if (reqCtx.era === 'modern' || subscribedUris.has(COUNTER_URI)) {
|
||||
// Connection serving (stdio): announce in-band. The entry routes it
|
||||
// onto 2026-07-28 listen streams; on a 2025-era connection it goes
|
||||
// only to subscribers — unsolicited per-resource updates are wrong.
|
||||
await server.server.sendResourceUpdated({ uri: COUNTER_URI }).catch(() => {});
|
||||
}
|
||||
return { content: [{ type: 'text', text: String(counter) }] };
|
||||
});
|
||||
|
||||
return server;
|
||||
}
|
||||
|
||||
const { transport, port } = parseExampleArgs();
|
||||
|
||||
if (transport === 'stdio') {
|
||||
void serveStdio(buildServer);
|
||||
void serveStdio(reqCtx => buildServer(reqCtx));
|
||||
console.error('[server] serving over stdio');
|
||||
} else {
|
||||
const handler = createMcpHandler(buildServer);
|
||||
const handler = createMcpHandler(reqCtx => buildServer(reqCtx, uri => handler.notify.resourceUpdated(uri)));
|
||||
createServer(toNodeHandler(handler)).listen(port, () => {
|
||||
console.error(`[server] listening on http://127.0.0.1:${port}/mcp`);
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in a new issue