a5c-ai/mcp-transport-sse-setup

MCP Transport SSE Setup

Configure HTTP/SSE transport for web-based MCP servers.

Capabilities

• Configure SSE transport for MCP servers
• Set up HTTP endpoints for MCP communication
• Implement authentication middleware
• Configure CORS for browser clients
• Set up health check endpoints
• Implement connection management

Usage

Invoke this skill when you need to:

• Set up web-based MCP server transport
• Configure SSE for real-time communication
• Implement authentication for MCP endpoints
• Enable CORS for browser-based clients

Inputs

| Parameter | Type | Required | Description | |-----------|------|----------|-------------| | language | string | Yes | Target language (typescript, python) | | framework | string | No | Web framework (express, fastify, fastapi) | | auth | object | No | Authentication configuration | | cors | object | No | CORS configuration |

Generated Patterns

TypeScript Express SSE Transport

import express from 'express';
import cors from 'cors';
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse.js';

const app = express();

// CORS configuration
app.use(cors({
  origin: process.env.ALLOWED_ORIGINS?.split(',') || '*',
  credentials: true,
}));

app.use(express.json());

// Store active connections
const connections = new Map<string, SSEServerTransport>();

// SSE endpoint
app.get('/sse', async (req, res) => {
  const connectionId = req.query.connectionId as string || crypto.randomUUID();

  // Set SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.setHeader('X-Accel-Buffering', 'no');

  // Create transport
  const transport = new SSEServerTransport('/message', res);
  connections.set(connectionId, transport);

  // Create server instance for this connection
  const server = new Server(
    { name: 'my-mcp-server', version: '1.0.0' },
    { capabilities: { tools: {}, resources: {} } }
  );

  // Register handlers...
  registerToolHandlers(server);

  // Handle connection close
  req.on('close', () => {
    connections.delete(connectionId);
    transport.close();
  });

  // Connect transport
  await server.connect(transport);
});

// Message endpoint
app.post('/message', async (req, res) => {
  const connectionId = req.query.connectionId as string;
  const transport = connections.get(connectionId);

  if (!transport) {
    res.status(404).json({ error: 'Connection not found' });
    return;
  }

  try {
    await transport.handlePostMessage(req, res);
  } catch (error) {
    res.status(500).json({ error: 'Failed to process message' });
  }
});

// Health check
app.get('/health', (req, res) => {
  res.json({
    status: 'healthy',
    connections: connections.size,
    timestamp: new Date().toISOString(),
  });
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`MCP Server listening on port ${PORT}`);
});

Python FastAPI SSE Transport

from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from sse_starlette.sse import EventSourceResponse
from mcp.server import Server
from mcp.server.sse import SseServerTransport
import uuid
import asyncio

app = FastAPI()

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Store connections
connections: dict[str, SseServerTransport] = {}

@app.get("/sse")
async def sse_endpoint(request: Request):
    connection_id = request.query_params.get("connectionId") or str(uuid.uuid4())

    async def event_generator():
        transport = SseServerTransport("/message")
        connections[connection_id] = transport

        server = Server("my-mcp-server")
        register_handlers(server)

        try:
            async for event in transport.events():
                yield event
        finally:
            connections.pop(connection_id, None)

    return EventSourceResponse(event_generator())

@app.post("/message")
async def message_endpoint(request: Request):
    connection_id = request.query_params.get("connectionId")
    transport = connections.get(connection_id)

    if not transport:
        return Response(status_code=404, content="Connection not found")

    body = await request.json()
    await transport.handle_message(body)
    return Response(status_code=200)

@app.get("/health")
async def health():
    return {
        "status": "healthy",
        "connections": len(connections),
    }

Workflow

1. Configure framework - Set up web framework
2. Add CORS - Configure cross-origin requests
3. Create SSE endpoint - Set up event stream
4. Add message handler - POST endpoint for messages
5. Implement auth - Optional authentication
6. Add health check - Monitoring endpoint

Target Processes

• mcp-transport-layer
• mcp-server-bootstrap
• mcp-server-monitoring-debugging