Building a stdio MCP server

Kashish Hora

Co-founder of AgentCat

The Quick Answer

Create a stdio MCP server that communicates through standard input/output streams. Install the SDK and implement your server:

$npm install @modelcontextprotocol/sdk zod

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new McpServer({
  name: "my-stdio-server",
  version: "1.0.0"
});

const transport = new StdioServerTransport();
await server.connect(transport);

The stdio transport enables subprocess communication where the client launches your server and exchanges JSON-RPC messages through stdin/stdout streams.

Prerequisites

Node.js v18 or higher (for TypeScript implementation)
Python 3.10+ with asyncio support (for Python implementation) — required by mcp 2.0.0 (requires_python >=3.10) and by the X | None annotations used in the handler signatures below
Basic understanding of JSON-RPC protocol
Familiarity with subprocess communication patterns

Installation

TypeScript

# Create new project
$mkdir my-mcp-server && cd my-mcp-server
$npm init -y
 
# Install dependencies
$npm install @modelcontextprotocol/sdk zod
$npm install -D typescript @types/node tsx
 
# Initialize TypeScript
$npx tsc --init

This installs @modelcontextprotocol/sdk 1.30.0, the legacy (2025-11-25-era) TypeScript line; the packages implementing the 2026-07-28 revision are @modelcontextprotocol/server/client/core 2.0.0, which require Node 20+. The samples below use the 1.x API, including registerTool/registerResource, which are also the recommended calls on that line — the older variadic server.tool(...)/server.resource(...) forms are deprecated in 1.30.0 and removed in the 2.0.0 packages.

Python

# Create virtual environment
$python -m venv venv
$source venv/bin/activate  # On Windows: venv\Scripts\activate
 
# Install MCP SDK
$pip install mcp

This installs mcp 2.0.0, the current stable release (shipped 2026-07-28). The lowlevel Server examples below target that release's handler API. Before that release (mcp<2, final 1.x: 1.29.0), the lowlevel Server registered handlers with decorators (@app.list_tools(), @app.call_tool()) instead of the constructor keyword arguments shown below.

Configuration

MCP stdio servers require specific configuration in the client application. The client spawns your server as a subprocess and manages the communication channels.

Client Configuration (Claude Desktop)

{
  "mcpServers": {
    "my-server": {
      "command": "node",
      "args": ["path/to/server.js"],
      "env": {
        "API_KEY": "your-api-key"
      }
    }
  }
}

The command specifies the executable, args contains command-line arguments, and env sets environment variables. The client automatically connects stdin/stdout streams for bidirectional communication.

TypeScript Server Configuration

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new McpServer(
  {
    name: "my-stdio-server",
    version: "1.0.0",
    description: "A stdio-based MCP server"
  },
  { capabilities: { tools: {} } }
);

McpServer takes two arguments: an Implementation object identifying the server (name and version are required; title, description, websiteUrl, and icons are optional) and an optional ServerOptions object for capabilities and instructions. There's no vendor field, and ServerOptions itself isn't exported from server/mcp.js — it's just the type of that second argument, not something you construct and pass as the whole config. Include descriptive names and semantic versioning for better compatibility tracking.

Usage

Building a functional stdio MCP server involves registering tools, resources, and prompts that clients can discover and invoke. The server processes incoming JSON-RPC requests and returns structured responses.

Implementing Tools

import { z } from "zod";

server.registerTool(
  "calculate",
  {
    description: "Perform arithmetic operations",
    inputSchema: {
      operation: z.enum(["add", "subtract", "multiply", "divide"]),
      a: z.number(),
      b: z.number()
    }
  },
  async ({ operation, a, b }) => {
    let result: number;
    switch (operation) {
      case "add": result = a + b; break;
      case "subtract": result = a - b; break;
      case "multiply": result = a * b; break;
      case "divide": 
        if (b === 0) throw new Error("Division by zero");
        result = a / b; 
        break;
    }
    
    return {
      content: [{ type: "text", text: `Result: ${result}` }]
    };
  }
);

Tools expose executable functions to the client. Use Zod schemas for input validation, ensuring type safety and automatic error handling for malformed requests.

Python Implementation

from mcp.server import Server, ServerRequestContext
from mcp.server.stdio import stdio_server
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)
import asyncio


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[
        Tool(
            name="calculate",
            description="Perform arithmetic operations",
            input_schema={
                "type": "object",
                "properties": {
                    "operation": {"type": "string", "enum": ["add", "subtract"]},
                    "a": {"type": "number"},
                    "b": {"type": "number"}
                },
                "required": ["operation", "a", "b"]
            }
        )
    ])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    arguments = params.arguments or {}
    if params.name == "calculate":
        try:
            op = arguments["operation"]
            a, b = arguments["a"], arguments["b"]
        except KeyError as e:
            return CallToolResult(
                content=[TextContent(type="text", text=f"Missing required argument: {e}")],
                is_error=True,
            )
        result = a + b if op == "add" else a - b
        return CallToolResult(content=[TextContent(type="text", text=f"Result: {result}")])
    return CallToolResult(
        content=[TextContent(type="text", text=f"Unknown tool: {params.name}")],
        is_error=True,
    )


app = Server("my-stdio-server", on_list_tools=list_tools, on_call_tool=call_tool)


async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

Python's async context managers handle stream lifecycle automatically. Server.run() is unchanged since before the rename — it still requires the initialization options as a third argument, built with create_initialization_options() (pass NotificationOptions from mcp.server if you want list-changed notifications). Handlers are passed to the Server constructor as on_list_tools/on_call_tool keyword arguments — the mcp 2.0.0 lowlevel Server API replaced decorator-based registration with constructor parameters, and handlers now take (ctx, params) and return the full result type rather than a raw dict. On mcp<2, the same server used the @app.list_tools()/@app.call_tool() decorators shown in older tutorials, and handlers could return plain dicts.

Two more changes from that migration affect the code above, not just its shape. First, the old @app.call_tool() decorator validated arguments against the tool's input_schema for you; mcp 2.0.0 removed that automatic validation with no built-in replacement, which is why call_tool above wraps the argument lookups in try/except and returns is_error=True for a missing argument instead of letting a bare KeyError escape. Second, handler exceptions are no longer auto-wrapped into a CallToolResult — an unhandled exception now surfaces to the client as a JSON-RPC protocol error rather than a recoverable tool error, so catch what you expect to fail. (The unknown-tool fallback below returns a CallToolResult with is_error=True, a legal, model-recoverable answer; the spec's own example for an unrecognized tool name instead returns a JSON-RPC -32602 error. Both are valid — they're just not the same shape, so pick one convention and apply it consistently.)

Handling Resources

server.registerResource(
  "settings",
  "config://settings",
  { title: "Application configuration", mimeType: "application/json" },
  async (uri) => {
    const config = {
      debug: process.env.DEBUG === "true",
      maxConnections: parseInt(process.env.MAX_CONN || "10"),
      timeout: parseInt(process.env.TIMEOUT || "30000")
    };

    return {
      contents: [{
        uri: uri.href,
        mimeType: "application/json",
        text: JSON.stringify(config, null, 2)
      }]
    };
  }
);

Resources provide read-only data access. Use URI schemes to categorize resources (e.g., file://, config://, data://) and return appropriate MIME types for content negotiation. Note the argument order on registerResource: name first, then the uri (or a ResourceTemplate), then metadata, then the read callback — mixing up name and uri registers the resource under the wrong identifier.

Common Issues

Error: "Invalid JSON-RPC message"

The stdio transport requires precise message formatting with newline delimiters. Logging to stdout corrupts the protocol stream.

// WRONG - corrupts stdout
console.log("Debug info");

// CORRECT - use stderr
console.error("Debug info");

// BETTER - use proper logging
import { writeFileSync } from "fs";
function log(message: string) {
  writeFileSync("server.log", `${new Date().toISOString()} ${message}\n`, { flag: "a" });
}

Always redirect debug output to stderr or log files. The client expects only JSON-RPC messages on stdout, and any other content causes parsing failures.

Error: "Server process terminated unexpectedly"

Unhandled exceptions crash the subprocess, breaking the client connection. Implement global error handlers to maintain stability.

process.on("uncaughtException", (error) => {
  console.error("Uncaught exception:", error);
  // Log error details but don't exit
});

process.on("unhandledRejection", (reason, promise) => {
  console.error("Unhandled rejection at:", promise, "reason:", reason);
});

// Graceful shutdown
process.on("SIGTERM", async () => {
  await transport.close();
  process.exit(0);
});

Proper error boundaries prevent cascading failures. Log errors for debugging while keeping the server operational for subsequent requests.

Error: "Message framing error"

The stdio protocol expects exactly one JSON object per line. It's easy to assume that means you have to hand-escape newlines inside your own string values, but JSON.stringify already does that for you: a template literal that spans several source lines serializes to a single line with \n escapes baked into the string, so it never breaks framing. A multi-line template literal and a string with the newlines already escaped by hand produce byte-identical JSON either way.

What actually breaks framing is writing a JSON document that is itself spread across multiple lines on the wire — most commonly by pretty-printing the response you send to stdout:

// WRONG - pretty-printed output spans multiple lines on the wire
process.stdout.write(JSON.stringify(response, null, 2) + "\n");

// CORRECT - compact, single-line JSON
process.stdout.write(JSON.stringify(response) + "\n");

JSON.stringify(response, null, 2) inserts real newlines between the JSON's own tokens (after every {, ,, and so on) — those are unescaped, structural newlines, not characters inside a string value, so the stdio reader sees several incomplete lines instead of one message. Reserve null, 2 pretty-printing for logs and files. Anything written to stdout should go through a plain JSON.stringify(value) call, which always emits a single line no matter how much text — or how many embedded newlines — the content itself contains.

Examples

File System MCP Server

This example demonstrates a production-ready file system server with proper error handling and security considerations:

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { readFile, readdir, stat } from "fs/promises";
import { join, resolve, relative } from "path";
import { z } from "zod";

const server = new McpServer({
  name: "filesystem-server",
  version: "1.0.0"
});

// Security: restrict to specific directory
const ALLOWED_ROOT = process.env.FS_ROOT || process.cwd();

function validatePath(requestedPath: string): string {
  const resolved = resolve(ALLOWED_ROOT, requestedPath);
  const rel = relative(ALLOWED_ROOT, resolved);
  if (rel.startsWith("..")) {
    throw new Error("Access denied: Path outside allowed directory");
  }
  return resolved;
}

server.registerTool(
  "readFile",
  { description: "Read a file's contents", inputSchema: { path: z.string() } },
  async ({ path }) => {
    try {
      const safePath = validatePath(path);
      const content = await readFile(safePath, "utf-8");
      return {
        content: [{ type: "text", text: content }]
      };
    } catch (error) {
      return {
        content: [{ 
          type: "text", 
          text: `Error reading file: ${error.message}` 
        }],
        isError: true
      };
    }
  }
);

server.registerTool(
  "listDirectory",
  { description: "List entries in a directory", inputSchema: { path: z.string().default(".") } },
  async ({ path }) => {
    const safePath = validatePath(path);
    const entries = await readdir(safePath, { withFileTypes: true });
    
    const formatted = await Promise.all(
      entries.map(async (entry) => {
        const fullPath = join(safePath, entry.name);
        const stats = await stat(fullPath);
        return {
          name: entry.name,
          type: entry.isDirectory() ? "directory" : "file",
          size: stats.size,
          modified: stats.mtime.toISOString()
        };
      })
    );
    
    return {
      content: [{ 
        type: "text", 
        text: JSON.stringify(formatted, null, 2) 
      }]
    };
  }
);

const transport = new StdioServerTransport();
await server.connect(transport);

This implementation validates all paths against a root directory, preventing directory traversal attacks. Error handling ensures graceful failures without exposing system internals. The server returns structured data suitable for further processing by AI models.

Database Query Server

A practical example showing database integration with connection pooling and prepared statements:

from mcp.server import Server, ServerRequestContext
from mcp.server.stdio import stdio_server
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)
import asyncio
import asyncpg
import json
import os

class DatabasePool:
    def __init__(self):
        self.pool = None
    
    async def initialize(self):
        self.pool = await asyncpg.create_pool(
            host="localhost",
            database="myapp",
            user="readonly",
            password=os.environ.get("DB_PASSWORD"),
            min_size=1,
            max_size=10,
            command_timeout=10
        )
    
    async def execute_query(self, query: str, params: list = None):
        async with self.pool.acquire() as conn:
            # Set readonly transaction
            async with conn.transaction(readonly=True):
                rows = await conn.fetch(query, *(params or []))
                return [dict(row) for row in rows]

db = DatabasePool()

# Same mcp 2.0.0 on_list_tools/on_call_tool pattern as the calculate example above.
async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[
        Tool(
            name="query",
            description="Execute a SELECT query",
            input_schema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "params": {"type": "array", "items": {"type": "string"}}
                },
                "required": ["query"]
            }
        )
    ])

async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    arguments = params.arguments or {}
    if params.name == "query":
        try:
            query = arguments["query"]
        except KeyError:
            return CallToolResult(
                content=[TextContent(type="text", text="Missing required argument: query")],
                is_error=True,
            )
        query_params = arguments.get("params", [])

        # Validate query is SELECT only
        if not query.strip().upper().startswith("SELECT"):
            return CallToolResult(
                content=[TextContent(type="text", text="Error: Only SELECT queries are allowed")],
                is_error=True,
            )

        try:
            results = await db.execute_query(query, query_params)
            return CallToolResult(
                content=[TextContent(type="text", text=json.dumps(results, default=str, indent=2))]
            )
        except Exception as e:
            return CallToolResult(
                content=[TextContent(type="text", text=f"Query error: {str(e)}")],
                is_error=True,
            )
    return CallToolResult(
        content=[TextContent(type="text", text=f"Unknown tool: {params.name}")],
        is_error=True,
    )

app = Server("database-server", on_list_tools=list_tools, on_call_tool=call_tool)

async def main():
    await db.initialize()
    async with stdio_server() as (reader, writer):
        await app.run(reader, writer, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

Connection pooling optimizes database resource usage across multiple requests. Read-only transactions and query validation provide security layers. The parameterized queries prevent SQL injection while maintaining flexibility for dynamic queries.

Integration with External APIs

This example shows how to wrap external services as MCP tools with rate limiting and caching:

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import fetch from "node-fetch";
import NodeCache from "node-cache";

const server = new McpServer({
  name: "weather-server",
  version: "1.0.0"
});

// Cache responses for 10 minutes
const cache = new NodeCache({ stdTTL: 600 });

// Simple rate limiter
const rateLimiter = {
  requests: new Map<string, number[]>(),
  
  check(key: string, limit: number, window: number): boolean {
    const now = Date.now();
    const requests = this.requests.get(key) || [];
    const recent = requests.filter(t => now - t < window);
    
    if (recent.length >= limit) {
      return false;
    }
    
    recent.push(now);
    this.requests.set(key, recent);
    return true;
  }
};

server.registerTool(
  "getWeather",
  {
    description: "Get the current weather for a location",
    inputSchema: {
      location: z.string(),
      units: z.enum(["metric", "imperial"]).default("metric")
    }
  },
  async ({ location, units }) => {
    // Check cache first
    const cacheKey = `${location}-${units}`;
    const cached = cache.get(cacheKey);
    if (cached) {
      return {
        content: [{ 
          type: "text", 
          text: `${cached} (cached)` 
        }]
      };
    }
    
    // Rate limit: 10 requests per minute
    if (!rateLimiter.check("weather-api", 10, 60000)) {
      return {
        content: [{ 
          type: "text", 
          text: "Rate limit exceeded. Please try again later." 
        }],
        isError: true
      };
    }
    
    try {
      const apiKey = process.env.WEATHER_API_KEY;
      const response = await fetch(
        `https://api.openweathermap.org/data/2.5/weather?q=${location}&units=${units}&appid=${apiKey}`
      );
      
      if (!response.ok) {
        throw new Error(`API error: ${response.status}`);
      }
      
      const data = await response.json();
      const result = `Weather in ${data.name}: ${data.main.temp}°, ${data.weather[0].description}`;
      
      // Cache the result
      cache.set(cacheKey, result);
      
      return {
        content: [{ type: "text", text: result }]
      };
    } catch (error) {
      return {
        content: [{ 
          type: "text", 
          text: `Failed to fetch weather: ${error.message}` 
        }],
        isError: true
      };
    }
  }
);

const transport = new StdioServerTransport();
await server.connect(transport);

External API integration requires careful consideration of rate limits and failure modes. Caching reduces API calls and improves response times. The rate limiter prevents abuse while maintaining service availability for legitimate requests.

Related Guides

Comparing stdio vs. SSE vs. Streamable HTTP

How to choose an MCP transport: stdio for local subprocesses, Streamable HTTP for remote services, and why the old HTTP+SSE transport is deprecated, not a third option.

Building an MCP server in TypeScript

Build an MCP server with the TypeScript SDK: the @modelcontextprotocol/sdk 1.x McpServer API pinned to 1.30.0, registerTool with structured output, and both transports (stdio and Streamable HTTP), plus a note on the newer split @modelcontextprotocol/server 2.0 line for the 2026-07-28 spec revision.

Configuring MCP transport protocols for Docker containers

Configure MCP servers in Docker containers with proper transport protocols and networking.

Keep an eye on AI.

Get rich user analytics and tracing on every user interacting with your MCP server.

Get started