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

[WIP] Refactor everything server to be more modular and use recommended APIs.

Browse source 
* Adding static resources, move server instructions to
the new docs folder, and add code formatting

* Add docs folder

* Add docs/architecture.md which describes the architecture of the project thus far.

* Refactor moved instructions.md to docs/server-instructions.md

* Add resources/static.ts
  - in addStaticResources()
    - read the file entries from the docs folder
    - register each file as a resource (no template), with a readResource function that reads the file and returns it in a contents block with the appropriate mime type and contents
  - getMimeType helper function gets the mime type for a filename
  - readSafe helper function reads the file synchronously as utf-8 or returns an error string

* Add resources/index.ts
  - import addStaticResources
  - export registerResources function
  - in registerResources()
    - call addStaticResources

* In package.json
  - add prettier devDependency
  - add prettier:check script
  - add prettier:fix script
  - in build script, copy docs folder to dist

* All other changes were prettier formatting
This commit is contained in:
cliffhall 2025-12-05 13:26:08 -05:00
commit 1c64b36c78
2 changed files with 915 additions and 538 deletions
Download patch file Download diff file Expand all files Collapse all files

36
src/everything/transports/sse.ts
View file

@ -1,27 +1,35 @@
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import express from "express";
import { createServer } from "../server/index.js";
import cors from 'cors';
import cors from "cors";
console.error('Starting SSE server...');
console.error("Starting SSE server...");
const app = express();
app.use(cors({
"origin": "*", // use "*" with caution in production
"methods": "GET,POST",
"preflightContinue": false,
"optionsSuccessStatus": 204,
})); // Enable CORS for all routes so Inspector can connect
const transports: Map<string, SSEServerTransport> = new Map<string, SSEServerTransport>();
app.use(
cors({
origin: "*", // use "*" with caution in production
methods: "GET,POST",
preflightContinue: false,
optionsSuccessStatus: 204,
})
); // Enable CORS for all routes so Inspector can connect
const transports: Map<string, SSEServerTransport> = new Map<
string,
SSEServerTransport
>();
app.get("/sse", async (req, res) => {
let transport: SSEServerTransport;
const { server, cleanup, startNotificationIntervals } = createServer();
if (req?.query?.sessionId) {
const sessionId = (req?.query?.sessionId as string);
const sessionId = req?.query?.sessionId as string;
transport = transports.get(sessionId) as SSEServerTransport;
console.error("Client Reconnecting? This shouldn't happen; when client has a sessionId, GET /sse should not be called again.", transport.sessionId);
console.error(
"Client Reconnecting? This shouldn't happen; when client has a sessionId, GET /sse should not be called again.",
transport.sessionId
);
} else {
// Create and store transport for new session
transport = new SSEServerTransport("/message", res);
@ -40,19 +48,17 @@ app.get("/sse", async (req, res) => {
transports.delete(transport.sessionId);
await cleanup();
};
}
});
app.post("/message", async (req, res) => {
const sessionId = (req?.query?.sessionId as string);
const sessionId = req?.query?.sessionId as string;
const transport = transports.get(sessionId);
if (transport) {
console.error("Client Message from", sessionId);
await transport.handlePostMessage(req, res);
} else {
console.error(`No transport found for sessionId ${sessionId}`)
console.error(`No transport found for sessionId ${sessionId}`);
}
});

306
src/everything/transports/streamableHttp.ts
View file

@ -9,198 +9,198 @@ console.log("Starting Streamable HTTP server...");
const app = express();
app.use(
cors({
origin: "*", // use "*" with caution in production
methods: "GET,POST,DELETE",
preflightContinue: false,
optionsSuccessStatus: 204,
exposedHeaders: ["mcp-session-id", "last-event-id", "mcp-protocol-version"],
}),
cors({
origin: "*", // use "*" with caution in production
methods: "GET,POST,DELETE",
preflightContinue: false,
optionsSuccessStatus: 204,
exposedHeaders: ["mcp-session-id", "last-event-id", "mcp-protocol-version"],
})
); // Enable CORS for all routes so Inspector can connect
const transports: Map<string, StreamableHTTPServerTransport> = new Map<
string,
StreamableHTTPServerTransport
string,
StreamableHTTPServerTransport
>();
app.post("/mcp", async (req: Request, res: Response) => {
console.log("Received MCP POST request");
try {
// Check for existing session ID
const sessionId = req.headers["mcp-session-id"] as string | undefined;
console.log("Received MCP POST request");
try {
// Check for existing session ID
const sessionId = req.headers["mcp-session-id"] as string | undefined;
let transport: StreamableHTTPServerTransport;
let transport: StreamableHTTPServerTransport;
if (sessionId && transports.has(sessionId)) {
// Reuse existing transport
transport = transports.get(sessionId)!;
} else if (!sessionId) {
const { server } = createServer();
if (sessionId && transports.has(sessionId)) {
// Reuse existing transport
transport = transports.get(sessionId)!;
} else if (!sessionId) {
const { server } = createServer();
// New initialization request
const eventStore = new InMemoryEventStore();
transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => randomUUID(),
eventStore, // Enable resumability
onsessioninitialized: (sessionId: string) => {
// Store the transport by session ID when session is initialized
// This avoids race conditions where requests might come in before the session is stored
console.log(`Session initialized with ID: ${sessionId}`);
transports.set(sessionId, transport);
},
});
// New initialization request
const eventStore = new InMemoryEventStore();
transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => randomUUID(),
eventStore, // Enable resumability
onsessioninitialized: (sessionId: string) => {
// Store the transport by session ID when session is initialized
// This avoids race conditions where requests might come in before the session is stored
console.log(`Session initialized with ID: ${sessionId}`);
transports.set(sessionId, transport);
},
});
// Set up onclose handler to clean up transport when closed
server.server.onclose = async () => {
const sid = transport.sessionId;
if (sid && transports.has(sid)) {
console.log(
`Transport closed for session ${sid}, removing from transports map`,
);
transports.delete(sid);
}
};
// Connect the transport to the MCP server BEFORE handling the request
// so responses can flow back through the same transport
await server.connect(transport);
await transport.handleRequest(req, res);
return;
} else {
// Invalid request - no session ID or not initialization request
res.status(400).json({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Bad Request: No valid session ID provided",
},
id: req?.body?.id,
});
return;
// Set up onclose handler to clean up transport when closed
server.server.onclose = async () => {
const sid = transport.sessionId;
if (sid && transports.has(sid)) {
console.log(
`Transport closed for session ${sid}, removing from transports map`
);
transports.delete(sid);
}
};
// Handle the request with existing transport - no need to reconnect
// The existing transport is already connected to the server
await transport.handleRequest(req, res);
} catch (error) {
console.log("Error handling MCP request:", error);
if (!res.headersSent) {
res.status(500).json({
jsonrpc: "2.0",
error: {
code: -32603,
message: "Internal server error",
},
id: req?.body?.id,
});
return;
}
// Connect the transport to the MCP server BEFORE handling the request
// so responses can flow back through the same transport
await server.connect(transport);
await transport.handleRequest(req, res);
return;
} else {
// Invalid request - no session ID or not initialization request
res.status(400).json({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Bad Request: No valid session ID provided",
},
id: req?.body?.id,
});
return;
}
// Handle the request with existing transport - no need to reconnect
// The existing transport is already connected to the server
await transport.handleRequest(req, res);
} catch (error) {
console.log("Error handling MCP request:", error);
if (!res.headersSent) {
res.status(500).json({
jsonrpc: "2.0",
error: {
code: -32603,
message: "Internal server error",
},
id: req?.body?.id,
});
return;
}
}
});
// Handle GET requests for SSE streams (using built-in support from StreamableHTTP)
app.get("/mcp", async (req: Request, res: Response) => {
console.log("Received MCP GET request");
const sessionId = req.headers["mcp-session-id"] as string | undefined;
if (!sessionId || !transports.has(sessionId)) {
res.status(400).json({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Bad Request: No valid session ID provided",
},
id: req?.body?.id,
});
return;
}
console.log("Received MCP GET request");
const sessionId = req.headers["mcp-session-id"] as string | undefined;
if (!sessionId || !transports.has(sessionId)) {
res.status(400).json({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Bad Request: No valid session ID provided",
},
id: req?.body?.id,
});
return;
}
// Check for Last-Event-ID header for resumability
const lastEventId = req.headers["last-event-id"] as string | undefined;
if (lastEventId) {
console.log(`Client reconnecting with Last-Event-ID: ${lastEventId}`);
} else {
console.log(`Establishing new SSE stream for session ${sessionId}`);
}
// Check for Last-Event-ID header for resumability
const lastEventId = req.headers["last-event-id"] as string | undefined;
if (lastEventId) {
console.log(`Client reconnecting with Last-Event-ID: ${lastEventId}`);
} else {
console.log(`Establishing new SSE stream for session ${sessionId}`);
}
const transport = transports.get(sessionId);
await transport!.handleRequest(req, res);
const transport = transports.get(sessionId);
await transport!.handleRequest(req, res);
});
// Handle DELETE requests for session termination (according to MCP spec)
app.delete("/mcp", async (req: Request, res: Response) => {
const sessionId = req.headers["mcp-session-id"] as string | undefined;
if (!sessionId || !transports.has(sessionId)) {
res.status(400).json({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Bad Request: No valid session ID provided",
},
id: req?.body?.id,
});
return;
}
const sessionId = req.headers["mcp-session-id"] as string | undefined;
if (!sessionId || !transports.has(sessionId)) {
res.status(400).json({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Bad Request: No valid session ID provided",
},
id: req?.body?.id,
});
return;
}
console.log(`Received session termination request for session ${sessionId}`);
console.log(`Received session termination request for session ${sessionId}`);
try {
const transport = transports.get(sessionId);
await transport!.handleRequest(req, res);
} catch (error) {
console.log("Error handling session termination:", error);
if (!res.headersSent) {
res.status(500).json({
jsonrpc: "2.0",
error: {
code: -32603,
message: "Error handling session termination",
},
id: req?.body?.id,
});
return;
}
try {
const transport = transports.get(sessionId);
await transport!.handleRequest(req, res);
} catch (error) {
console.log("Error handling session termination:", error);
if (!res.headersSent) {
res.status(500).json({
jsonrpc: "2.0",
error: {
code: -32603,
message: "Error handling session termination",
},
id: req?.body?.id,
});
return;
}
}
});
// Start the server
const PORT = process.env.PORT || 3001;
const server = app.listen(PORT, () => {
console.error(`MCP Streamable HTTP Server listening on port ${PORT}`);
console.error(`MCP Streamable HTTP Server listening on port ${PORT}`);
});
server.on("error", (err: unknown) => {
const code =
typeof err === "object" && err !== null && "code" in err
? (err as { code?: unknown }).code
: undefined;
if (code === "EADDRINUSE") {
console.error(
`Failed to start: Port ${PORT} is already in use. Set PORT to a free port or stop the conflicting process.`,
);
} else {
console.error("HTTP server encountered an error while starting:", err);
}
// Ensure a non-zero exit so npm reports the failure instead of silently exiting
process.exit(1);
const code =
typeof err === "object" && err !== null && "code" in err
? (err as { code?: unknown }).code
: undefined;
if (code === "EADDRINUSE") {
console.error(
`Failed to start: Port ${PORT} is already in use. Set PORT to a free port or stop the conflicting process.`
);
} else {
console.error("HTTP server encountered an error while starting:", err);
}
// Ensure a non-zero exit so npm reports the failure instead of silently exiting
process.exit(1);
});
// Handle server shutdown
process.on("SIGINT", async () => {
console.log("Shutting down server...");
console.log("Shutting down server...");
// Close all active transports to properly clean up resources
for (const sessionId in transports) {
try {
console.log(`Closing transport for session ${sessionId}`);
await transports.get(sessionId)!.close();
transports.delete(sessionId);
} catch (error) {
console.log(`Error closing transport for session ${sessionId}:`, error);
}
// Close all active transports to properly clean up resources
for (const sessionId in transports) {
try {
console.log(`Closing transport for session ${sessionId}`);
await transports.get(sessionId)!.close();
transports.delete(sessionId);
} catch (error) {
console.log(`Error closing transport for session ${sessionId}:`, error);
}
}
console.log("Server shutdown complete");
process.exit(0);
console.log("Server shutdown complete");
process.exit(0);
});