add mcp server with find user tool

Author: kentcdodds

app/routes/mcp+/fetch-transport.server.ts

@@ -0,0 +1,154 @@
+import { randomUUID } from 'node:crypto'
+import { type Transport } from '@modelcontextprotocol/sdk/shared/transport.js'
+import {
+	type JSONRPCMessage,
+	JSONRPCMessageSchema,
+} from '@modelcontextprotocol/sdk/types.js'
+
+/**
+ * Server transport for SSE using Standard Fetch: this will send messages over
+ * an SSE connection and receive messages from HTTP POST requests.
+ */
+export class FetchSSEServerTransport implements Transport {
+	#stream?: ReadableStreamDefaultController<string>
+	#sessionId: string
+	#endpoint: string
+
+	onclose?: () => void
+	onerror?: (error: Error) => void
+	onmessage?: (message: JSONRPCMessage) => void
+
+	/**
+	 * Creates a new SSE server transport, which will direct the client to POST
+	 * messages to the relative or absolute URL identified by `endpoint`.
+	 */
+	constructor(endpoint: string, sessionId?: string | null) {
+		this.#endpoint = endpoint
+		this.#sessionId = sessionId ?? randomUUID()
+	}
+
+	/**
+	 * Starts processing messages on the transport.
+	 * This is called by the Server class and should not be called directly.
+	 */
+	async start(): Promise<void> {
+		if (this.#stream) {
+			throw new Error(
+				'FetchSSEServerTransport already started! If using Server class, note that connect() calls start() automatically.',
+			)
+		}
+	}
+
+	/**
+	 * Handles the initial SSE connection request.
+	 * This should be called from your Remix loader to establish the SSE stream.
+	 */
+	async handleSSERequest(request: Request): Promise<Response> {
+		const stream = new ReadableStream<string>({
+			start: (controller) => {
+				this.#stream = controller
+
+				// Send headers
+				controller.enqueue(': ping\n\n') // Keep connection alive
+
+				// Send the endpoint event
+				controller.enqueue(
+					`event: endpoint\ndata: ${encodeURI(
+						this.#endpoint,
+					)}?sessionId=${this.#sessionId}\n\n`,
+				)
+
+				// Handle cleanup when the connection closes
+				request.signal.addEventListener('abort', () => {
+					controller.close()
+					this.#stream = undefined
+					this.onclose?.()
+				})
+			},
+			cancel: () => {
+				this.#stream = undefined
+				this.onclose?.()
+			},
+		})
+
+		return new Response(stream, {
+			headers: {
+				'Content-Type': 'text/event-stream',
+				'Cache-Control': 'no-cache',
+				Connection: 'keep-alive',
+				'Mcp-Session-Id': this.#sessionId,
+			},
+		})
+	}
+
+	/**
+	 * Handles incoming POST messages.
+	 * This should be called from your Remix action to handle incoming messages.
+	 */
+	async handlePostMessage(request: Request): Promise<Response> {
+		if (!this.#stream) {
+			const message = 'SSE connection not established'
+			return new Response(message, { status: 500 })
+		}
+
+		let body: unknown
+		try {
+			const contentType = request.headers.get('content-type')
+			if (contentType !== 'application/json') {
+				throw new Error(`Unsupported content-type: ${contentType}`)
+			}
+
+			body = await request.json()
+		} catch (error) {
+			this.onerror?.(error as Error)
+			return new Response(String(error), { status: 400 })
+		}
+
+		try {
+			await this.handleMessage(body)
+		} catch (error) {
+			console.error(error)
+			return new Response(`Invalid message: ${body}`, { status: 400 })
+		}
+
+		return new Response('Accepted', { status: 202 })
+	}
+
+	/**
+	 * Handle a client message, regardless of how it arrived.
+	 */
+	async handleMessage(message: unknown): Promise<void> {
+		let parsedMessage: JSONRPCMessage
+		try {
+			parsedMessage = JSONRPCMessageSchema.parse(message)
+		} catch (error) {
+			this.onerror?.(error as Error)
+			throw error
+		}
+
+		this.onmessage?.(parsedMessage)
+	}
+
+	async close(): Promise<void> {
+		this.#stream?.close()
+		this.#stream = undefined
+		this.onclose?.()
+	}
+
+	async send(message: JSONRPCMessage): Promise<void> {
+		if (!this.#stream) {
+			throw new Error('Not connected')
+		}
+
+		// Send the message through the event stream
+		this.#stream.enqueue(`event: message\ndata: ${JSON.stringify(message)}\n\n`)
+	}
+
+	/**
+	 * Returns the session ID for this transport.
+	 * This can be used to route incoming POST requests.
+	 */
+	get sessionId(): string {
+		return this.#sessionId
+	}
+}

app/routes/mcp+/index.ts

@@ -0,0 +1,22 @@
+import { type Route } from './+types/index.ts'
+import { connect, getTransport } from './mcp.server.ts'
+
+export async function loader({ request }: Route.LoaderArgs) {
+	const url = new URL(request.url)
+	const sessionId = url.searchParams.get('sessionId')
+	const transport = await connect(sessionId)
+	return transport.handleSSERequest(request)
+}
+
+export async function action({ request }: Route.ActionArgs) {
+	const url = new URL(request.url)
+	const sessionId = url.searchParams.get('sessionId')
+	if (!sessionId) {
+		return new Response('No session ID', { status: 400 })
+	}
+	const transport = await getTransport(sessionId)
+	if (!transport) {
+		return new Response('Not found', { status: 404 })
+	}
+	return transport.handlePostMessage(request)
+}

app/routes/mcp+/mcp.server.ts

@@ -0,0 +1,71 @@
+import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'
+import { type CallToolResult } from '@modelcontextprotocol/sdk/types.js'
+import { searchUsers } from '@prisma/client/sql'
+import { z } from 'zod'
+import { prisma } from '#app/utils/db.server.ts'
+import { getSignedGetRequestInfo } from '#app/utils/storage.server.ts'
+import { FetchSSEServerTransport } from './fetch-transport.server.ts'
+
+export const server = new McpServer(
+	{
+		name: 'epic-mcp-a25d',
+		version: '1.0.0',
+	},
+	{
+		capabilities: {
+			tools: {},
+		},
+	},
+)
+
+server.tool(
+	'Find User',
+	{ query: z.string().describe('The query to search for') },
+	async ({ query }) => {
+		const like = `%${query ?? ''}%`
+		const users = await prisma.$queryRawTyped(searchUsers(like))
+
+		const content: CallToolResult['content'] = []
+		for (const user of users) {
+			content.push({
+				type: 'text',
+				text: `${user.name} (${user.username})`,
+			})
+
+			if (user.imageObjectKey) {
+				content.push({
+					type: 'image',
+					data: await getUserBase64Image(user.imageObjectKey),
+					mimeType: 'image/png',
+				})
+			}
+		}
+
+		return { content }
+	},
+)
+
+async function getUserBase64Image(imageObjectKey: string) {
+	const { url: signedUrl, headers: signedHeaders } =
+		getSignedGetRequestInfo(imageObjectKey)
+	const response = await fetch(signedUrl, { headers: signedHeaders })
+	const blob = await response.blob()
+	const buffer = await blob.arrayBuffer()
+	return Buffer.from(buffer).toString('base64')
+}
+
+const transports = new Map<string, FetchSSEServerTransport>()
+
+export async function connect(sessionId?: string | null) {
+	const transport = new FetchSSEServerTransport('/mcp', sessionId)
+	transport.onclose = () => {
+		transports.delete(transport.sessionId)
+	}
+	await server.connect(transport)
+	transports.set(transport.sessionId, transport)
+	return transport
+}
+
+export async function getTransport(sessionId: string) {
+	return transports.get(sessionId)
+}