SSE MCP Server for GEO Analytics

Overview

The Model Context Protocol (MCP) defines a standard interface through which LLM clients — such as ChatGPT plugins, Claude desktop, or custom agent runtimes — can discover and invoke tools exposed by a server. Rather than baking tool logic directly into a prompt or agent framework, MCP separates the tool surface from the client: any compliant client can connect to any compliant server and immediately access its capabilities. This separation is particularly valuable in analytics contexts where the same database-backed tools should be reachable from multiple AI clients without duplicating integration code.

Server-Sent Events (SSE) is the HTTP transport layer that MCP uses to stream messages from server to client over a persistent connection. Unlike WebSockets, SSE is unidirectional (server to client) and works over plain HTTP/1.1 without an upgrade handshake, making it easy to host behind standard reverse proxies and load balancers. The client sends tool invocation requests via regular HTTP POST, and the server pushes results back over the open SSE stream. This architecture lets MCP servers run as ordinary Express applications that any LLM client can reach over HTTPS.

As an example, consider an Express-based MCP server that exposes tools for searching geographic event records, fetching individual entries, retrieving prompt templates, and querying dashboard statistics — all backed by postgres() connections to the live database. LLM clients streaming monitoring data receive structured results over SSE without knowing anything about the underlying schema. This document covers how to build this kind of SSE MCP server: the connection lifecycle, tool registration patterns, Postgres-backed handler design, and response streaming mechanics.

Core Concepts

How It Works

The server processes each client interaction through three phases: SSE connection setup, tool dispatch, and response streaming.

SSE Connection Lifecycle

When an MCP client starts a session it issues a GET request to the /sse endpoint. Express sets the response headers to Content-Type: text/event-stream, Cache-Control: no-cache, and Connection: keep-alive, then hands the open response object to the MCP SDK’s SSEServerTransport. The transport assigns a unique session ID and sends an initial endpoint event telling the client which URL to POST tool requests to (typically /messages?sessionId=<id>). The connection stays open until the client disconnects or the server closes it. A close event listener on the request triggers transport cleanup.

Tool Dispatch

When the client POSTs a tools/call JSON-RPC message to /messages, Express routes it to the transport associated with the session ID from the query string. The MCP SDK deserializes the message, validates the tool name against the registered tool list, runs the input JSON Schema validation, and invokes the registered handler. If the tool name is unknown or the input fails schema validation, the SDK returns a JSON-RPC error response over the SSE stream without calling any handler. Successful dispatch invokes the handler with the validated arguments object.

Response Streaming

The handler performs its Postgres query and returns an array of MCP content blocks — typically a single { type: "text", text: JSON.stringify(rows) } block for structured data, or multiple blocks for mixed content. The SDK serializes the tools/call JSON-RPC response containing these content blocks and writes it to the open SSE connection as a data: event. The client reads the event off the stream, parses the JSON-RPC response, and extracts the content blocks for use in the LLM context. The SSE connection remains open for subsequent tool calls within the same session.

Implementation Patterns

Express SSE Server Setup

Initialize the Express app, create the MCP server instance, and wire SSE and message endpoints to the transport layer.

import express from "express";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import postgres from "postgres";

const app = express();
app.use(express.json());

const sql = postgres(process.env.DATABASE_URL!);

// One MCP server instance shared across all sessions.
const mcpServer = new McpServer({
  name: "geo-analytics-mcp",
  version: "1.0.0",
});

// Track active transports by session ID.
const transports = new Map();

// SSE endpoint — client connects here to open a session.
app.get("/sse", async (req, res) => {
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");

  const transport = new SSEServerTransport("/messages", res);
  transports.set(transport.sessionId, transport);

  req.on("close", () => {
    transports.delete(transport.sessionId);
    transport.close();
  });

  await mcpServer.connect(transport);
});

// Message endpoint — client POSTs tool requests here.
app.post("/messages", async (req, res) => {
  const sessionId = req.query.sessionId as string;
  const transport = transports.get(sessionId);

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

  await transport.handlePostMessage(req, res, req.body);
});

app.listen(3000, () => {
  console.log("GEO Analytics MCP server listening on :3000");
});

MCP Tool Handler Registration

Register each tool with its name, input schema, and a Postgres-backed async handler that returns MCP content blocks.

import { z } from "zod";

// Tool: search GEO event records by region and time window.
mcpServer.tool(
  "search",
  "Search GEO monitoring events by region code and optional time range.",
  {
    region: z.string().describe("ISO 3166-1 alpha-2 region code, e.g. 'US'"),
    from: z.string().optional().describe("ISO 8601 start timestamp"),
    to: z.string().optional().describe("ISO 8601 end timestamp"),
    limit: z.number().int().min(1).max(200).default(50),
  },
  async ({ region, from, to, limit }) => {
    const rows = await sql`
      SELECT id, region, event_type, severity, occurred_at, payload
      FROM geo_events
      WHERE region = ${region}
        AND (${from}::timestamptz IS NULL OR occurred_at >= ${from}::timestamptz)
        AND (${to}::timestamptz   IS NULL OR occurred_at <= ${to}::timestamptz)
      ORDER BY occurred_at DESC
      LIMIT ${limit}
    `;
    return {
      content: [{ type: "text", text: JSON.stringify(rows, null, 2) }],
    };
  }
);

// Tool: fetch a single GEO event by ID.
mcpServer.tool(
  "fetch",
  "Retrieve a single GEO event record by its UUID.",
  { id: z.string().uuid() },
  async ({ id }) => {
    const [row] = await sql`
      SELECT * FROM geo_events WHERE id = ${id}
    `;
    if (!row) {
      return {
        content: [{ type: "text", text: JSON.stringify({ error: "Not found" }) }],
        isError: true,
      };
    }
    return {
      content: [{ type: "text", text: JSON.stringify(row, null, 2) }],
    };
  }
);

// Tool: get_prompts — return prompt templates stored in the database.
mcpServer.tool(
  "get_prompts",
  "List available LLM prompt templates for GEO analysis workflows.",
  { category: z.string().optional() },
  async ({ category }) => {
    const rows = await sql`
      SELECT slug, title, template, variables
      FROM prompt_templates
      WHERE (${category}::text IS NULL OR category = ${category})
      ORDER BY title
    `;
    return {
      content: [{ type: "text", text: JSON.stringify(rows, null, 2) }],
    };
  }
);

// Tool: dashboard_stats — aggregate metrics for the monitoring dashboard.
mcpServer.tool(
  "dashboard_stats",
  "Return aggregate GEO event statistics for the monitoring dashboard.",
  {
    window_hours: z
      .number()
      .int()
      .min(1)
      .max(720)
      .default(24)
      .describe("Look-back window in hours"),
  },
  async ({ window_hours }) => {
    const [stats] = await sql`
      SELECT
        COUNT(*)                                        AS total_events,
        COUNT(*) FILTER (WHERE severity = 'critical')  AS critical_count,
        COUNT(*) FILTER (WHERE severity = 'warning')   AS warning_count,
        COUNT(DISTINCT region)                         AS active_regions,
        MAX(occurred_at)                               AS latest_event_at
      FROM geo_events
      WHERE occurred_at >= NOW() - (${window_hours} || ' hours')::interval
    `;
    return {
      content: [{ type: "text", text: JSON.stringify(stats, null, 2) }],
    };
  }
);

Graceful Shutdown and Connection Cleanup

Close all active SSE transports and the Postgres pool before the process exits to avoid dangling connections.

async function shutdown(signal: string) {
  console.log(`Received ${signal}, shutting down gracefully...`);

  // Close all active SSE sessions.
  for (const [sessionId, transport] of transports) {
    try {
      await transport.close();
    } catch {
      // Ignore errors on already-closed connections.
    }
    transports.delete(sessionId);
  }

  // Close the Postgres connection pool.
  await sql.end({ timeout: 5 });

  console.log("Shutdown complete.");
  process.exit(0);
}

process.on("SIGTERM", () => shutdown("SIGTERM"));
process.on("SIGINT",  () => shutdown("SIGINT"));

Common Pitfalls

• Missing keep-alive headers — Without Cache-Control: no-cache and Connection: keep-alive, reverse proxies (nginx, Caddy, AWS ALB) will buffer or close the SSE stream before the client receives events. Always set these headers explicitly and configure proxy read timeouts to exceed the expected session duration.
• Session ID mismatch on POST — The client must include the correct sessionId query parameter when POSTing to /messages. If the transport map lookup fails, the server returns 404 and the client’s tool call is silently lost. Log session ID mismatches explicitly and ensure the client reads the endpoint event from the SSE stream before sending any POST requests.
• Unclosed Postgres connections on disconnect — If a client disconnects mid-query without triggering the close event (e.g., due to a network timeout), the Postgres query may continue running server-side. Use statement timeouts (SET statement_timeout = '10s') in query handlers and monitor pg_stat_activity for long-running queries left by abandoned sessions.
• Blocking the event loop with large result sets — JSON.stringify() on thousands of rows runs synchronously and can stall the Node.js event loop, delaying SSE events for other sessions. For large datasets, limit query results at the SQL level, paginate via cursor, or stream rows through a Transform stream that serializes and emits chunks incrementally.
• Schema validation mismatch between Zod and JSON Schema — The MCP SDK serializes Zod schemas to JSON Schema for the tools/list response. Some Zod constructs (e.g., z.union with more than two members, z.record with complex value types) produce JSON Schema that some MCP clients cannot parse. Stick to plain object schemas with string, number, and boolean fields for maximum client compatibility.
• Not handling isError in client code — MCP tool responses include an isError flag for application-level errors (e.g., record not found, permission denied). LLM clients that ignore this flag will treat error messages as normal tool output, potentially misleading the model. Always check isError in client handling and surface errors distinctly in the LLM context.