modelcontextprotocol/typescript-sdk
Watch
1
0
Fork
You've already forked typescript-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

share implementation of in memory store between examples and integration tests

Browse source
This commit is contained in:
ihrpr 2025-04-16 14:48:47 +01:00
commit c3f08567a2
No known key found for this signature in database
4 changed files with 86 additions and 213 deletions
Download patch file Download diff file Expand all files Collapse all files

80
src/examples/server/simpleStreamableHttp.ts
View file

@ -1,84 +1,10 @@
import express, { Request, Response } from 'express';
import { randomUUID } from 'node:crypto';
import { McpServer } from '../../server/mcp.js';
import { EventStore, StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
import { z } from 'zod';
import { CallToolResult, GetPromptResult, isInitializeRequest, JSONRPCMessage, ReadResourceResult } from '../../types.js';
// Create a simple in-memory EventStore for resumability
class InMemoryEventStore implements EventStore {
private events: Map<string, { streamId: string, message: JSONRPCMessage }> = new Map();
/**
* Generates a unique event ID for a given stream ID
*/
private generateEventId(streamId: string): string {
return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`;
}
private getStreamIdFromEventId(eventId: string): string {
const parts = eventId.split('_');
return parts.length > 0 ? parts[0] : '';
}
/**
* Stores an event with a generated event ID
* Implements EventStore.storeEvent
*/
async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
const eventId = this.generateEventId(streamId);
console.log(`Storing event ${eventId} for stream ${streamId}`);
this.events.set(eventId, { streamId, message });
return eventId;
}
/**
* Replays events that occurred after a specific event ID
* Implements EventStore.replayEventsAfter
*/
async replayEventsAfter(lastEventId: string,
{ send }: { send: (eventId: string, message: JSONRPCMessage) => Promise<void> }
): Promise<string> {
if (!lastEventId || !this.events.has(lastEventId)) {
console.log(`No events found for lastEventId: ${lastEventId}`);
return '';
}
// Extract the stream ID from the event ID
const streamId = this.getStreamIdFromEventId(lastEventId);
if (!streamId) {
console.log(`Could not extract streamId from lastEventId: ${lastEventId}`);
return '';
}
let foundLastEvent = false;
let eventCount = 0;
// Sort events by eventId for chronological ordering
const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0]));
for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) {
// Only include events from the same stream
if (eventStreamId !== streamId) {
continue;
}
// Start sending events after we find the lastEventId
if (eventId === lastEventId) {
foundLastEvent = true;
continue;
}
if (foundLastEvent) {
await send(eventId, message);
eventCount++;
}
}
console.log(`Replayed ${eventCount} events after ${lastEventId} for stream ${streamId}`);
return streamId;
}
}
import { CallToolResult, GetPromptResult, isInitializeRequest, ReadResourceResult } from '../../types.js';
import { InMemoryEventStore } from '../shared/inMemoryEventStore.js';
// Create an MCP server with implementation details
const server = new McpServer({

79
src/examples/server/sseAndStreamableHttpCompatibleServer.ts
View file

@ -1,10 +1,11 @@
import express, { Request, Response } from 'express';
import { randomUUID } from "node:crypto";
import { McpServer } from '../../server/mcp.js';
import { EventStore, StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
import { SSEServerTransport } from '../../server/sse.js';
import { z } from 'zod';
import { CallToolResult, isInitializeRequest, JSONRPCMessage } from '../../types.js';
import { CallToolResult, isInitializeRequest } from '../../types.js';
import { InMemoryEventStore } from '../shared/inMemoryEventStore.js';
/**
* This example server demonstrates backwards compatibility with both:
@ -17,81 +18,7 @@ import { CallToolResult, isInitializeRequest, JSONRPCMessage } from '../../types
* - /request: The deprecated POST endpoint for older clients (POST to send messages)
*/
// Simple in-memory event store for resumability
class InMemoryEventStore implements EventStore {
private events: Map<string, { streamId: string, message: JSONRPCMessage }> = new Map();
/**
* Generates a unique event ID for a given stream ID
*/
private generateEventId(streamId: string): string {
return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`;
}
private getStreamIdFromEventId(eventId: string): string {
const parts = eventId.split('_');
return parts.length > 0 ? parts[0] : '';
}
/**
* Stores an event with a generated event ID
* Implements EventStore.storeEvent
*/
async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
const eventId = this.generateEventId(streamId);
console.log(`Storing event ${eventId} for stream ${streamId}`);
this.events.set(eventId, { streamId, message });
return eventId;
}
/**
* Replays events that occurred after a specific event ID
* Implements EventStore.replayEventsAfter
*/
async replayEventsAfter(lastEventId: string,
{ send }: { send: (eventId: string, message: JSONRPCMessage) => Promise<void> }
): Promise<string> {
if (!lastEventId || !this.events.has(lastEventId)) {
console.log(`No events found for lastEventId: ${lastEventId}`);
return '';
}
const streamId = this.getStreamIdFromEventId(lastEventId);
if (!streamId) {
console.log(`Could not extract streamId from lastEventId: ${lastEventId}`);
return '';
}
let foundLastEvent = false;
let eventCount = 0;
// Sort events by eventId for chronological ordering
const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0]));
for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) {
// Only include events from the same stream
if (eventStreamId !== streamId) {
continue;
}
// Start sending events after we find the lastEventId
if (eventId === lastEventId) {
foundLastEvent = true;
continue;
}
if (foundLastEvent) {
await send(eventId, message);
eventCount++;
}
}
console.log(`Replayed ${eventCount} events after ${lastEventId} for stream ${streamId}`);
return streamId;
}
}
// Create a shared MCP server instance
const server = new McpServer({
name: 'backwards-compatible-server',
version: '1.0.0',

77
src/examples/shared/inMemoryEventStore.ts Normal file
View file

@ -0,0 +1,77 @@
import { JSONRPCMessage } from '../../types.js';
import { EventStore } from '../../server/streamableHttp.js';
/**
* Simple in-memory implementation of the EventStore interface for resumability
* This is primarily intended for examples and testing, not for production use
* where a persistent storage solution would be more appropriate.
*/
export class InMemoryEventStore implements EventStore {
private events: Map<string, { streamId: string, message: JSONRPCMessage }> = new Map();
/**
* Generates a unique event ID for a given stream ID
*/
private generateEventId(streamId: string): string {
return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`;
}
/**
* Extracts the stream ID from an event ID
*/
private getStreamIdFromEventId(eventId: string): string {
const parts = eventId.split('_');
return parts.length > 0 ? parts[0] : '';
}
/**
* Stores an event with a generated event ID
* Implements EventStore.storeEvent
*/
async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
const eventId = this.generateEventId(streamId);
this.events.set(eventId, { streamId, message });
return eventId;
}
/**
* Replays events that occurred after a specific event ID
* Implements EventStore.replayEventsAfter
*/
async replayEventsAfter(lastEventId: string,
{ send }: { send: (eventId: string, message: JSONRPCMessage) => Promise<void> }
): Promise<string> {
if (!lastEventId || !this.events.has(lastEventId)) {
return '';
}
// Extract the stream ID from the event ID
const streamId = this.getStreamIdFromEventId(lastEventId);
if (!streamId) {
return '';
}
let foundLastEvent = false;
// Sort events by eventId for chronological ordering
const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0]));
for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) {
// Only include events from the same stream
if (eventStreamId !== streamId) {
continue;
}
// Start sending events after we find the lastEventId
if (eventId === lastEventId) {
foundLastEvent = true;
continue;
}
if (foundLastEvent) {
await send(eventId, message);
}
}
return streamId;
}
}

63
src/integration-tests/taskResumability.test.ts
View file

@ -4,68 +4,11 @@ import { randomUUID } from 'node:crypto';
import { Client } from '../client/index.js';
import { StreamableHTTPClientTransport } from '../client/streamableHttp.js';
import { McpServer } from '../server/mcp.js';
import { EventStore, StreamableHTTPServerTransport } from '../server/streamableHttp.js';
import { CallToolResultSchema, JSONRPCMessage, LoggingMessageNotificationSchema } from '../types.js';
import { StreamableHTTPServerTransport } from '../server/streamableHttp.js';
import { CallToolResultSchema, LoggingMessageNotificationSchema } from '../types.js';
import { z } from 'zod';
import { InMemoryEventStore } from '../examples/shared/inMemoryEventStore.js';
/**
* Simple in-memory event store implementation for resumability
*/
class InMemoryEventStore implements EventStore {
private events: Map<string, { streamId: string, message: JSONRPCMessage }> = new Map();
private generateEventId(streamId: string): string {
return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`;
}
private getStreamIdFromEventId(eventId: string): string {
const parts = eventId.split('_');
return parts.length > 0 ? parts[0] : '';
}
async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
const eventId = this.generateEventId(streamId);
this.events.set(eventId, { streamId, message });
return eventId;
}
async replayEventsAfter(lastEventId: string,
{ send }: { send: (eventId: string, message: JSONRPCMessage) => Promise<void> }
): Promise<string> {
if (!lastEventId || !this.events.has(lastEventId)) {
return '';
}
// Extract the stream ID from the event ID
const streamId = this.getStreamIdFromEventId(lastEventId);
if (!streamId) {
return '';
}
let foundLastEvent = false;
// Sort events by eventId for chronological ordering
const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0]));
for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) {
// Only include events from the same stream
if (eventStreamId !== streamId) {
continue;
}
// Start collecting events after we find the lastEventId
if (eventId === lastEventId) {
foundLastEvent = true;
continue;
}
if (foundLastEvent) {
await send(eventId, message);
}
}
return streamId;
}
}
describe('Transport resumability', () => {