|
1 | | -import type { IncomingMessage, ServerResponse } from "node:http"; |
2 | 1 | import type { Server } from "@modelcontextprotocol/sdk/server/index.js"; |
3 | 2 | import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; |
4 | | -import { type RequestHandlers, createBaseHttpServer } from "../utils"; |
5 | | -import { Logger } from "../utils/logger"; |
| 3 | +import type { Request, Response } from "express"; |
| 4 | +import express from "express"; |
6 | 5 |
|
7 | 6 | export const startSSEMcpServer = async ( |
8 | 7 | server: Server, |
9 | 8 | endpoint = "/sse", |
10 | 9 | port = 3033, |
11 | 10 | ): Promise<void> => { |
12 | | - const activeTransports: Record<string, SSEServerTransport> = {}; |
| 11 | + const app = express(); |
| 12 | + app.use(express.json()); |
13 | 13 |
|
14 | | - // Define the request handler for SSE-specific logic |
15 | | - const handleRequest: RequestHandlers["handleRequest"] = async ( |
16 | | - req: IncomingMessage, |
17 | | - res: ServerResponse, |
18 | | - ) => { |
19 | | - if (!req.url) { |
20 | | - res.writeHead(400).end("No URL"); |
21 | | - return; |
22 | | - } |
23 | | - |
24 | | - const reqUrl = new URL(req.url, "http://localhost"); |
| 14 | + const transports: Record<string, SSEServerTransport> = {}; |
25 | 15 |
|
26 | | - // Handle GET requests to the SSE endpoint |
27 | | - if (req.method === "GET" && reqUrl.pathname === endpoint) { |
| 16 | + app.get(endpoint, async (req: Request, res: Response) => { |
| 17 | + try { |
28 | 18 | const transport = new SSEServerTransport("/messages", res); |
29 | | - |
30 | | - activeTransports[transport.sessionId] = transport; |
31 | | - |
32 | | - let closed = false; |
33 | | - |
34 | | - res.on("close", async () => { |
35 | | - closed = true; |
36 | | - |
37 | | - try { |
38 | | - await server.close(); |
39 | | - } catch (error) { |
40 | | - Logger.error("Error closing server", error); |
41 | | - } |
42 | | - |
43 | | - delete activeTransports[transport.sessionId]; |
44 | | - }); |
45 | | - |
46 | | - try { |
47 | | - await server.connect(transport); |
48 | | - |
49 | | - await transport.send({ |
50 | | - jsonrpc: "2.0", |
51 | | - method: "sse/connection", |
52 | | - params: { message: "SSE Connection established" }, |
53 | | - }); |
54 | | - } catch (error) { |
55 | | - if (!closed) { |
56 | | - Logger.error("Error connecting to server", error); |
57 | | - res.writeHead(500).end("Error connecting to server"); |
58 | | - } |
59 | | - } |
60 | | - |
61 | | - return; |
| 19 | + transports[transport.sessionId] = transport; |
| 20 | + transport.onclose = () => delete transports[transport.sessionId]; |
| 21 | + await server.connect(transport); |
| 22 | + } catch (error) { |
| 23 | + if (!res.headersSent) |
| 24 | + res.status(500).send("Error establishing SSE stream"); |
62 | 25 | } |
| 26 | + }); |
63 | 27 |
|
64 | | - // Handle POST requests to the messages endpoint |
65 | | - if (req.method === "POST" && req.url?.startsWith("/messages")) { |
66 | | - const sessionId = new URL( |
67 | | - req.url, |
68 | | - "https://example.com", |
69 | | - ).searchParams.get("sessionId"); |
70 | | - |
71 | | - if (!sessionId) { |
72 | | - res.writeHead(400).end("No sessionId"); |
73 | | - return; |
74 | | - } |
75 | | - |
76 | | - const activeTransport: SSEServerTransport | undefined = |
77 | | - activeTransports[sessionId]; |
78 | | - |
79 | | - if (!activeTransport) { |
80 | | - res.writeHead(400).end("No active transport"); |
81 | | - return; |
82 | | - } |
83 | | - |
84 | | - await activeTransport.handlePostMessage(req, res); |
85 | | - return; |
86 | | - } |
| 28 | + app.post("/messages", async (req: Request, res: Response) => { |
| 29 | + const sessionId = req.query.sessionId as string; |
| 30 | + if (!sessionId) return res.status(400).send("Missing sessionId parameter"); |
87 | 31 |
|
88 | | - // If we reach here, no handler matched |
89 | | - res.writeHead(404).end("Not found"); |
90 | | - }; |
| 32 | + const transport = transports[sessionId]; |
| 33 | + if (!transport) return res.status(404).send("Session not found"); |
91 | 34 |
|
92 | | - // Custom cleanup for SSE server |
93 | | - const cleanup = () => { |
94 | | - // Close all active transports |
95 | | - for (const transport of Object.values(activeTransports)) { |
96 | | - try { |
97 | | - transport.close(); |
98 | | - } catch (error) { |
99 | | - Logger.error("Error closing SSE transport", error); |
100 | | - } |
| 35 | + try { |
| 36 | + await transport.handlePostMessage(req, res, req.body); |
| 37 | + } catch (error) { |
| 38 | + if (!res.headersSent) res.status(500).send("Error handling request"); |
101 | 39 | } |
102 | | - }; |
| 40 | + }); |
103 | 41 |
|
104 | | - // Create the HTTP server using our factory |
105 | | - createBaseHttpServer(port, endpoint, { |
106 | | - handleRequest, |
107 | | - cleanup, |
108 | | - serverType: "SSE Server", |
| 42 | + app.listen(port, () => { |
| 43 | + console.log(`SSE Server listening on http://localhost:${port}${endpoint}`); |
109 | 44 | }); |
110 | 45 | }; |
0 commit comments