Lesson 7 - Step 2 of 5

Reading from stdin

Learn to read and parse JSON-RPC messages from standard input.

Line-Based Reading

MCP STDIO transport uses line-based message framing. Each JSON-RPC message is exactly one line, terminated by a newline character. This makes reading messages straightforward:

python

import sys
import json

def read_message():
    """Read a single JSON-RPC message from stdin."""
    line = sys.stdin.readline()

    # Empty line means EOF (client disconnected)
    if not line:
        return None

    # Strip the trailing newline and parse JSON
    return json.loads(line.strip())

The Main Read Loop

An MCP server continuously reads messages until the client disconnects. Here's the basic pattern:

python

import sys
import json

def main():
    """Main server loop - read messages until EOF."""
    while True:
        # Read one line from stdin
        line = sys.stdin.readline()

        # EOF - client closed the connection
        if not line:
            break

        # Skip empty lines (shouldn't happen, but be safe)
        line = line.strip()
        if not line:
            continue

        try:
            # Parse the JSON message
            message = json.loads(line)

            # Process the message
            handle_message(message)

        except json.JSONDecodeError as e:
            # Log error to stderr, NOT stdout!
            sys.stderr.write(f"JSON parse error: {e}\n")

if __name__ == "__main__":
    main()

Critical: Handle EOF Correctly

When sys.stdin.readline() returns an empty string, the client has closed the connection. Your server should exit gracefully, not hang or crash.

Parsing JSON-RPC Messages

Once you've read and parsed the JSON, you need to identify the message type and extract its components:

python

def handle_message(message):
    """Route a JSON-RPC message to the appropriate handler."""

    # Every JSON-RPC message must have jsonrpc version
    if message.get("jsonrpc") != "2.0":
        sys.stderr.write("Invalid JSON-RPC version\n")
        return

    # Check if it's a request (has method) or response (has result/error)
    method = message.get("method")
    msg_id = message.get("id")

    if method:
        # This is a request or notification
        if msg_id is not None:
            # Request - expects a response
            handle_request(msg_id, method, message.get("params", {}))
        else:
            # Notification - no response expected
            handle_notification(method, message.get("params", {}))
    else:
        # This is a response (we don't handle these as a server)
        sys.stderr.write(f"Received unexpected response: {message}\n")

Handling Specific Methods

Now you can implement handlers for specific MCP methods:

python

def handle_request(msg_id, method, params):
    """Handle a JSON-RPC request that expects a response."""

    if method == "initialize":
        # Handle MCP initialization
        result = {
            "protocolVersion": "2024-11-05",
            "capabilities": {
                "tools": {}
            },
            "serverInfo": {
                "name": "my-stdio-server",
                "version": "1.0.0"
            }
        }
        send_response(msg_id, result)

    elif method == "tools/list":
        # Return available tools
        result = {
            "tools": [
                {
                    "name": "echo",
                    "description": "Echo back a message",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "message": {"type": "string"}
                        },
                        "required": ["message"]
                    }
                }
            ]
        }
        send_response(msg_id, result)

    elif method == "tools/call":
        # Execute a tool
        tool_name = params.get("name", "")
        arguments = params.get("arguments", {})

        if tool_name == "echo":
            result = {
                "content": [
                    {
                        "type": "text",
                        "text": arguments.get("message", "")
                    }
                ]
            }
            send_response(msg_id, result)
        else:
            send_error(msg_id, -32601, f"Unknown tool: {tool_name}")

    else:
        # Method not found
        send_error(msg_id, -32601, f"Method not found: {method}")

def handle_notification(method, params):
    """Handle a JSON-RPC notification (no response expected)."""

    if method == "notifications/initialized":
        # Client confirmed initialization - we're ready!
        sys.stderr.write("Client initialized successfully\n")
    else:
        sys.stderr.write(f"Unknown notification: {method}\n")

Complete Reading Example

Here's a complete example that demonstrates proper stdin reading:

python

#!/usr/bin/env python3
"""MCP STDIO Server - stdin reading example."""

import sys
import json

def read_message():
    """
    Read a single JSON-RPC message from stdin.

    Returns:
        dict: Parsed JSON message, or None on EOF
    """
    try:
        line = sys.stdin.readline()

        # EOF check
        if not line:
            sys.stderr.write("EOF received, shutting down\n")
            return None

        # Strip whitespace and parse
        line = line.strip()
        if not line:
            return read_message()  # Skip empty lines

        message = json.loads(line)
        sys.stderr.write(f"Received: {message.get('method', 'response')}\n")
        return message

    except json.JSONDecodeError as e:
        sys.stderr.write(f"JSON parse error: {e}\n")
        sys.stderr.write(f"Raw line: {repr(line)}\n")
        return read_message()  # Try to read next message

def main():
    """Main entry point."""
    sys.stderr.write("STDIO server starting...\n")

    while True:
        message = read_message()
        if message is None:
            break

        # Process message here...
        handle_message(message)

    sys.stderr.write("STDIO server shutting down\n")

if __name__ == "__main__":
    main()

Testing Tip

You can test stdin reading by piping JSON to your script:

echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | python server.py

Common Pitfalls

Blocking reads: readline() blocks until a complete line is available
Buffer issues: Always read complete lines, never partial data
Encoding: MCP uses UTF-8; ensure your stdin is UTF-8 encoded
Error recovery: Don't crash on invalid JSON; log and continue