MCP server STDIO transport: build custom tools with Python

You’re debugging why Claude Desktop won’t discover your MCP server, and the logs just say “connection failed” with zero protocol-level visibility. STDIO transport adds roughly 5ms latency compared to HTTP but eliminates port conflicts and authentication complexity for local tools. This guide builds a working MCP server STDIO transport in Python using raw JSON-RPC, diagnoses real transport failures, and graduates to maintainable SDK-based patterns.

What is MCP and why use STDIO transport?

MCP STDIO transport is a Model Context Protocol transport that uses standard input and output streams for JSON-RPC communication. It’s designed for local tools that run as child processes of AI clients like Claude Desktop, eliminating network configuration. Each message is a newline-terminated JSON object, making it ideal for CLI tools and desktop integrations where the client spawns your server as a subprocess.

The Model Context Protocol (MCP) enables AI assistants to discover and invoke external tools through a standardized interface. STDIO transport uses standard input/output streams for JSON-RPC communication, which works well for local CLI tools and desktop integrations. This approach avoids network overhead entirely.

HTTP vs STDIO: when to choose each

Choose STDIO when your tool runs locally alongside the AI client. The client spawns your server as a child process, communicates over pipes, and handles lifecycle management automatically. STDIO eliminates port conflicts, firewall configuration, and authentication.

Choose HTTP with SSE transport when you need remote server access, language decoupling, or multiple clients connecting to one server. HTTP requires infrastructure but separates server lifecycle from any single client.

Factor	STDIO	HTTP/SSE
Latency	~5ms lower	Network RTT added
Authentication	None needed	Required
Deployment	Single binary	HTTP server
Multi-client	No	Yes
Remote access	No	Yes

The JSON-RPC message framing you must get right

MCP uses JSON-RPC 2.0 with newline-delimited messages according to the [spec.modelcontextprotocol.io] specification. Each request and response is a single JSON object terminated by a newline character (\n). There are no Content-Length headers or chunked encoding.

This framing means your server must never write partial JSON objects. Every write() to stdout must contain exactly one complete JSON-RPC message followed by a newline. Missing newlines cause the client to hang indefinitely waiting for message boundaries.

How does MCP STDIO protocol framing work?

MCP over STDIO uses newline-delimited JSON-RPC 2.0 messages. Each request and response is a single JSON object terminated by a newline, with no Content-Length headers or chunked encoding. The protocol requires strict message boundaries because the stream has no inherent framing like HTTP.

Understanding JSON-RPC batching limits

JSON-RPC 2.0 supports batch requests where multiple calls arrive in a single array. MCP servers must handle batched requests, though most clients send individual requests. Your parser should detect arrays at the top level and process each element.

The MCP specification recommends against arbitrary batching limits. However, practical implementations should enforce reasonable bounds to prevent memory exhaustion from malformed requests.

Message boundary detection without HTTP headers

Without HTTP headers, the newline character defines message boundaries. Your server reads from stdin until it encounters \n, then parses the accumulated bytes as JSON. This sounds simple, but two issues break it:

1. Unflushed buffers: Your response sits in the stdout buffer and never reaches the client before timeout.
2. Platform newlines: Windows translates \n to \r\n in text mode, corrupting message boundaries.

Build a raw MCP server: Python STDIO implementation

Implement a minimal MCP server using only the standard library: read lines from stdin, parse JSON-RPC, route to handlers, and write responses to stdout with mandatory newline termination. This approach requires no dependencies beyond Python 3.10+ and teaches the protocol mechanics that SDKs abstract away.

Function calling is typically vendor-specific and embedded in chat APIs. MCP is an open protocol for tool discovery and invocation that works across clients and servers, with pluggable transports (STDIO, SSE, WebSocket) and standardized capability negotiation.

Setting up non-blocking stdin on Windows and Unix

Cross-platform stdin handling requires attention to platform-specific behavior. Windows needs binary mode to prevent newline translation.

# mcp_server_raw.py
import sys
import json
import asyncio
import logging
from typing import Any

# Configure logging to stderr (never stdout - that breaks JSON-RPC)
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s',
    stream=sys.stderr
)
logger = logging.getLogger(__name__)

# Windows requires binary mode for correct newline handling
if sys.platform == 'win32':
    import msvcrt
    msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)
    msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)

The code above redirects all logs to stderr. Writing logs to stdout would interleave with JSON-RPC messages and break the protocol.

Implementing initialize and tools/list handlers

The MCP handshake starts with an initialize request. The client sends protocol version and capabilities, and your server responds with its own capabilities.

# mcp_server_raw.py (continued)

# MCP Protocol Version (2025-03-26 spec)
PROTOCOL_VERSION = "2025-03-26"

SERVER_INFO = {
    "name": "nilesh-blog-mcp-server",
    "version": "1.0.0"
}

CAPABILITIES = {
    "tools": {}
}

def create_response(request_id: Any, result: Any) -> dict:
    return {
        "jsonrpc": "2.0",
        "id": request_id,
        "result": result
    }

def create_error(request_id: Any, code: int, message: str) -> dict:
    return {
        "jsonrpc": "2.0",
        "id": request_id,
        "error": {
            "code": code,
            "message": message
        }
    }

def handle_initialize(params: dict) -> dict:
    return {
        "protocolVersion": PROTOCOL_VERSION,
        "capabilities": CAPABILITIES,
        "serverInfo": SERVER_INFO
    }

def handle_tools_list(params: dict) -> dict:
    return {
        "tools": [
            {
                "name": "get_current_time",
                "description": "Returns the current UTC time in ISO 8601 format",
                "inputSchema": {
                    "type": "object",
                    "properties": {},
                    "required": []
                }
            },
            {
                "name": "add_numbers",
                "description": "Adds two integers and returns the result",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "a": {"type": "integer", "description": "First number"},
                        "b": {"type": "integer", "description": "Second number"}
                    },
                    "required": ["a", "b"]
                }
            }
        ]
    }

Each tool definition must include a JSON Schema-compliant inputSchema. Returning non-compliant schemas causes silent discovery failures where the client accepts the server but never shows the tools.

Implementing tools/call with error handling

The tools/call method invokes a specific tool by name. Your handler extracts parameters, executes the tool logic, and returns the result.

# mcp_server_raw.py (continued)
from datetime import datetime, timezone

def handle_tool_call(params: dict) -> dict:
    tool_name = params.get("name")
    arguments = params.get("arguments", {})
    
    logger.info(f"Tool call: {tool_name} with args {arguments}")
    
    if tool_name == "get_current_time":
        result = datetime.now(timezone.utc).isoformat()
        return {"content": [{"type": "text", "text": result}]}
    
    elif tool_name == "add_numbers":
        try:
            a = arguments.get("a")
            b = arguments.get("b")
            if not isinstance(a, int) or not isinstance(b, int):
                raise ValueError("Both arguments must be integers")
            result = a + b
            return {"content": [{"type": "text", "text": str(result)}]}
        except Exception as e:
            return {
                "content": [{"type": "text", "text": f"Error: {str(e)}"}],
                "isError": True
            }
    
    else:
        raise ValueError(f"Unknown tool: {tool_name}")

Error responses inside tools/call should set isError: true rather than returning JSON-RPC errors. This distinction lets the client present the error to the user while keeping the connection alive.

Main request router and message loop

The main loop reads from stdin, routes requests, and writes responses with explicit flushing.

# mcp_server_raw.py (continued)

METHOD_HANDLERS = {
    "initialize": handle_initialize,
    "tools/list": handle_tools_list,
    "tools/call": handle_tool_call,
}

def process_request(request: dict) -> dict:
    request_id = request.get("id")
    method = request.get("method")
    params = request.get("params", {})
    
    logger.debug(f"Processing: method={method}, id={request_id}")
    
    if method not in METHOD_HANDLERS:
        return create_error(request_id, -32601, f"Method not found: {method}")
    
    try:
        result = METHOD_HANDLERS[method](params)
        return create_response(request_id, result)
    except Exception as e:
        logger.exception(f"Handler error for {method}")
        return create_error(request_id, -32603, f"Internal error: {str(e)}")

def write_response(response: dict) -> None:
    """Write response to stdout with newline and flush."""
    json_str = json.dumps(response, separators=(',', ':'))
    sys.stdout.write(json_str + '\n')
    sys.stdout.flush()
    logger.debug(f"Sent response for id={response.get('id')}")

def main():
    logger.info("MCP server starting (raw STDIO implementation)")
    logger.info(f"Python version: {sys.version}")
    
    try:
        for line in sys.stdin:
            line = line.strip()
            if not line:
                continue
            
            logger.debug(f"Received: {line[:200]}...")
            
            try:
                request = json.loads(line)
            except json.JSONDecodeError as e:
                logger.error(f"JSON parse error: {e}")
                write_response(create_error(None, -32700, f"Parse error: {str(e)}"))
                continue
            
            # Handle batch requests
            if isinstance(request, list):
                for req in request:
                    response = process_request(req)
                    write_response(response)
            else:
                response = process_request(request)
                write_response(response)
                
    except KeyboardInterrupt:
        logger.info("Server shutting down")
    except BrokenPipeError:
        logger.info("Client disconnected")
    except Exception as e:
        logger.exception(f"Fatal error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

The explicit sys.stdout.flush() after each write prevents buffering delays. Missing flushes are a primary cause of “server hangs on startup” issues.

What breaks at the transport layer? Debugging MCP STDIO

Common failures include missing newlines causing hangs, unflushed stdout buffers, and stdin EOF on Windows due to CRT mode. These manifest as “Server disconnected” or infinite loading in clients like Claude Desktop.

Enable verbose logging to a file (not stdout), verify JSON-RPC messages end with newlines, check that your process doesn’t buffer stdout, and use MCP Inspector or manual stdin/stdout testing to isolate transport from protocol issues.

Reproducing and fixing the hanging read bug

The most common cause is blocking on stdin without proper newline handling, or stdout buffering preventing the initialize response from reaching the client. Ensure unbuffered binary mode on Windows and explicitly flush after each write.

Here is a script that reproduces the bug:

# buggy_server.py - DO NOT USE IN PRODUCTION
import sys
import json

# BUG: No flush after write
def main():
    for line in sys.stdin:
        request = json.loads(line)
        response = {"jsonrpc": "2.0", "id": request["id"], "result": {}}
        json_str = json.dumps(response)
        sys.stdout.write(json_str + '\n')
        # BUG: Missing flush - response stays in buffer
        # sys.stdout.flush()  # Uncomment to fix

if __name__ == "__main__":
    main()

Run this server and send a request manually. The response never arrives because Python’s stdout buffer holds it.

# Terminal 1: Start buggy server
python buggy_server.py

# Terminal 2: Send request (hangs waiting for response)
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}' | python buggy_server.py
# No output - the response is buffered

Add the flush call and the server works correctly.

Windows-specific: binary mode and codepage issues

Windows defaults to text mode for stdin/stdout, which translates \n to \r\n. This corrupts JSON parsing. The fix sets binary mode at startup.

# Windows fix for binary mode
if sys.platform == 'win32':
    import os
    import msvcrt
    # Set stdin/stdout to binary mode
    msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)
    msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)

Windows codepage issues can also corrupt UTF-8 characters. Set the console codepage explicitly:

if sys.platform == 'win32':
    import ctypes
    kernel32 = ctypes.windll.kernel32
    kernel32.SetConsoleOutputCP(65001)  # UTF-8
    kernel32.SetConsoleCP(65001)

Capturing traffic with Process Monitor or strace

On Unix systems, strace captures system calls including read/write operations on stdin/stdout.

# Trace a Python MCP server
strace -f -e trace=read,write -s 1000 python mcp_server_raw.py 2> trace.log

# In another terminal, send a request
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{}}}' | python mcp_server_raw.py

The trace shows raw bytes flowing through stdin and stdout. Look for partial writes or missing newlines.

On Windows, use Process Monitor with filters for ReadFile and WriteFile operations on the Python process.

MCP protocol debugger script

This diagnostic script sits between the client and server, logging all traffic without modifying the stream.

# mcp_debug_proxy.py
import sys
import json
import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(direction)s] %(message)s',
    stream=sys.stderr
)

def log_message(direction: str, data: str):
    record = logging.LogRecord(
        name='mcp-proxy',
        level=logging.INFO,
        pathname='',
        lineno=0,
        msg=data,
        args=(),
        exc_info=None
    )
    record.direction = direction
    logging.getLogger('mcp-proxy').handle(record)

def main():
    logger = logging.getLogger('mcp-proxy')
    logger.info("MCP Debug Proxy started")
    
    for line in sys.stdin:
        timestamp = datetime.now().isoformat()
        log_message('IN', line.strip())
        
        try:
            request = json.loads(line)
            logger.info(f"[IN] Parsed: method={request.get('method')}, id={request.get('id')}")
        except json.JSONDecodeError:
            logger.error(f"[IN] Invalid JSON: {line[:100]}")
        
        # Pass through to actual server would happen here
        # For testing, just echo back
        sys.stdout.write(line)
        sys.stdout.flush()
        log_message('OUT', line.strip())

if __name__ == "__main__":
    main()

Graduate to the official SDK: mcp[cli] patterns

The official Python SDK (mcp package) abstracts transport details. Use mcp.server.Server with the StdioServerTransport for production servers with automatic lifecycle management. The SDK handles capability negotiation, request routing, and error formatting according to the [github.com/modelcontextprotocol/python-sdk] specification.

Yes, you can implement initialize, tools/list, and tools/call handlers using only Python’s standard library. The protocol is open and based on JSON-RPC 2.0. However, the SDK provides conveniences like automatic capability negotiation.

Installing the official MCP SDK

The SDK requires Python 3.10 or higher and uses anyio for async support.

pip install 'mcp[cli]>=1.6.0'

[VERIFY: Is mcp Python SDK 1.6.0 still latest, or has 2.x released?]

Decorator-based tool registration vs manual routing

The SDK uses decorators for tool registration, eliminating manual routing tables.

# mcp_server_sdk.py
import asyncio
from datetime import datetime, timezone
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# Create server instance
server = Server("nilesh-blog-mcp-sdk")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_current_time",
            description="Returns the current UTC time in ISO 8601 format",
            inputSchema={
                "type": "object",
                "properties": {}
            }
        ),
        Tool(
            name="add_numbers",
            description="Adds two integers and returns the result",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {"type": "integer", "description": "First number"},
                    "b": {"type": "integer", "description": "Second number"}
                },
                "required": ["a", "b"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "get_current_time":
        result = datetime.now(timezone.utc).isoformat()
        return [TextContent(type="text", text=result)]
    
    elif name == "add_numbers":
        a = arguments.get("a")
        b = arguments.get("b")
        result = a + b
        return [TextContent(type="text", text=str(result))]
    
    else:
        raise ValueError(f"Unknown tool: {name}")

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

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

The SDK version is roughly 50 lines shorter than the raw implementation. It handles JSON-RPC parsing, error formatting, and stream lifecycle automatically.

Server capabilities negotiation

The SDK automatically declares capabilities based on registered handlers. If you register @server.list_tools(), the SDK declares the `tools