How to add streaming responses to an MCP server

1// TypeScript
2import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
3import { z } from "zod";
4
5const server = new McpServer({ name: "streaming-server", version: "1.0.0" });
6
7server.registerTool("process-files", {
8  description: "Process a batch of files with progress updates",
9  inputSchema: {
10    directory: z.string().describe("Directory to process"),
11  },
12}, async ({ directory }, extra) => {
13  const files = await getFileList(directory);
14  const total = files.length;
15  const results: string[] = [];
16
17  for (let i = 0; i < files.length; i++) {
18    // Send progress notification
19    if (extra.sendNotification) {
20      await extra.sendNotification({
21        method: "notifications/progress",
22        params: {
23          progressToken: extra._meta?.progressToken,
24          progress: i,
25          total,
26          message: `Processing ${files[i]}...`,
27        },
28      });
29    }
30
31    const result = await processFile(files[i]);
32    results.push(result);
33  }
34
35  return {
36    content: [{
37      type: "text",
38      text: `Processed ${total} files:\n${results.join("\n")}`,
39    }],
40  };
41});

1// TypeScript
2import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
3import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
4import express from "express";
5
6const app = express();
7const server = new McpServer({ name: "http-streaming-server", version: "1.0.0" });
8
9// Register tools...
10
11const transport = new StreamableHTTPServerTransport({
12  sessionIdGenerator: () => crypto.randomUUID(),
13});
14
15app.post("/mcp", async (req, res) => {
16  await transport.handleRequest(req, res);
17});
18
19app.get("/mcp", async (req, res) => {
20  // SSE endpoint for server-to-client streaming
21  await transport.handleRequest(req, res);
22});
23
24app.delete("/mcp", async (req, res) => {
25  // Session termination
26  await transport.handleRequest(req, res);
27});
28
29await server.connect(transport);
30
31app.listen(3000, () => {
32  console.error("Streamable HTTP MCP server on port 3000");
33});

1// TypeScript
2server.registerTool("multi-source-search", {
3  description: "Search across multiple data sources with incremental results",
4  inputSchema: {
5    query: z.string().describe("Search query"),
6  },
7}, async ({ query }, extra) => {
8  const sources = ["database", "documents", "api", "cache"];
9  const allResults: any[] = [];
10
11  for (let i = 0; i < sources.length; i++) {
12    const source = sources[i];
13
14    // Progress notification
15    if (extra.sendNotification) {
16      await extra.sendNotification({
17        method: "notifications/progress",
18        params: {
19          progressToken: extra._meta?.progressToken,
20          progress: i,
21          total: sources.length,
22          message: `Searching ${source}...`,
23        },
24      });
25
26      // Log intermediate results
27      await extra.sendNotification({
28        method: "notifications/message",
29        params: {
30          level: "info",
31          data: `Found results in ${source}`,
32        },
33      });
34    }
35
36    const results = await searchSource(source, query);
37    allResults.push(...results);
38  }
39
40  return {
41    content: [{
42      type: "text",
43      text: JSON.stringify({ total: allResults.length, results: allResults }, null, 2),
44    }],
45  };
46});

1# Python
2from mcp.server.fastmcp import FastMCP, Context
3
4mcp = FastMCP("streaming-python-server")
5
6@mcp.tool()
7async def process_batch(items: list[str], ctx: Context) -> str:
8    """Process a batch of items with progress reporting.
9
10    Args:
11        items: List of items to process
12        ctx: MCP context (injected automatically)
13    """
14    results = []
15    total = len(items)
16
17    for i, item in enumerate(items):
18        # Send progress
19        await ctx.report_progress(i, total)
20        await ctx.info(f"Processing item {i+1}/{total}: {item}")
21
22        result = await process_item(item)
23        results.append(result)
24
25    await ctx.report_progress(total, total)
26    return f"Processed {total} items: {', '.join(results)}"

1// claude_desktop_config.json
2{
3  "mcpServers": {
4    "my-streaming-server": {
5      "url": "http://localhost:3000/mcp"
6    }
7  }
8}
9
10// For remote servers with auth:
11{
12  "mcpServers": {
13    "my-remote-server": {
14      "url": "https://mcp.example.com/mcp"
15    }
16  }
17}