STDIO Client
Launch MCP servers as subprocesses and communicate via stdin/stdout.
The STDIO Transport
The STDIO transport is the simplest way to communicate with an MCP server. Your client launches the server as a subprocess and communicates through stdin (sending requests) and stdout (receiving responses).
┌──────────────────────────────────────────────────────────────┐
│ MCP Client │
│ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ subprocess.Popen() │ │
│ │ │ │
│ │ stdin ────────────────────────► Server stdin │ │
│ │ (JSON-RPC requests) │ │
│ │ │ │
│ │ stdout ◄──────────────────────── Server stdout │ │
│ │ (JSON-RPC responses) │ │
│ │ │ │
│ │ stderr ◄──────────────────────── Server stderr │ │
│ │ (Debug/error logs) │ │
│ └────────────────────────────────────────────────────┘ │
│ │
└──────────────────────────────────────────────────────────────┘Launching the Server
Use Python's subprocess module to spawn the MCP server:
import subprocess
import json
from typing import Optional, List
class StdioClient:
"""MCP client that communicates via STDIO."""
def __init__(self, command: List[str]):
"""
Initialize the STDIO client.
Args:
command: Command to launch the server, e.g.,
["python", "my_server.py"] or
["node", "server.js"]
"""
self.command = command
self.process: Optional[subprocess.Popen] = None
def connect(self):
"""Launch the MCP server as a subprocess."""
self.process = subprocess.Popen(
self.command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True, # Use text mode for easier JSON handling
bufsize=1, # Line buffered
)
print(f"Started server with PID {self.process.pid}")
def disconnect(self):
"""Terminate the server process."""
if self.process:
self.process.terminate()
self.process.wait(timeout=5)
print("Server terminated")
We use text=True for convenience, but be aware that this
can cause issues with non-ASCII characters. For production code,
consider using binary mode and explicit UTF-8 encoding.
Sending Requests
Each JSON-RPC message is sent as a single line to the server's stdin:
def send_request(self, request: dict) -> None:
"""Send a JSON-RPC request to the server."""
if not self.process or not self.process.stdin or self.process.poll() is not None:
raise RuntimeError("Server not running")
# Serialize to JSON and add newline
message = json.dumps(request) + "\n"
# Write to server's stdin
self.process.stdin.write(message)
self.process.stdin.flush()
print(f"Sent: {request.get('method', 'response')}")Reading Responses
Read one line at a time from stdout and parse as JSON:
def read_response(self) -> dict:
"""Read a JSON-RPC response from the server."""
if not self.process or not self.process.stdout or self.process.poll() is not None:
raise RuntimeError("Server not running")
# Read one line from stdout
line = self.process.stdout.readline()
if not line:
raise RuntimeError("Server closed connection")
# Parse JSON
response = json.loads(line.strip())
print(f"Received: id={response.get('id')}")
return responseComplete STDIO Client Class
Here's the complete basic client putting it all together.
Create: stdio_client.py
"""
stdio_client.py - Basic STDIO MCP client.
"""
import subprocess
import json
from dataclasses import dataclass
from typing import Optional, List, Any
class StdioClient:
"""MCP client that communicates via STDIO."""
def __init__(self, command: List[str]):
self.command = command
self.process: Optional[subprocess.Popen] = None
self._next_id = 1
def connect(self):
"""Launch the MCP server as a subprocess."""
self.process = subprocess.Popen(
self.command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
)
def disconnect(self):
"""Terminate the server process."""
if self.process:
self.process.terminate()
try:
self.process.wait(timeout=5)
except subprocess.TimeoutExpired:
self.process.kill()
self.process = None
def send_request(self, method: str, params: Optional[dict] = None) -> dict:
"""Send a request and wait for the response."""
if not self.process or not self.process.stdin or not self.process.stdout:
raise RuntimeError("Not connected")
# Create request
request: dict[str, Any] = {
"jsonrpc": "2.0",
"id": self._next_id,
"method": method,
}
if params:
request["params"] = params
self._next_id += 1
# Send request
message = json.dumps(request) + "\n"
self.process.stdin.write(message)
self.process.stdin.flush()
# Read response
line = self.process.stdout.readline()
if not line:
raise RuntimeError("Server closed connection")
return json.loads(line.strip())
def send_notification(self, method: str, params: Optional[dict] = None):
"""Send a notification (no response expected)."""
if not self.process or not self.process.stdin:
raise RuntimeError("Not connected")
notification: dict[str, Any] = {
"jsonrpc": "2.0",
"method": method,
}
if params:
notification["params"] = params
message = json.dumps(notification) + "\n"
self.process.stdin.write(message)
self.process.stdin.flush()
def __enter__(self):
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.disconnect()Usage Example
Here's how to use the client:
# Using context manager
with StdioClient(["python", "my_mcp_server.py"]) as client:
# Send a request
response = client.send_request("tools/list")
print(response)
# Or manually
client = StdioClient(["python", "my_mcp_server.py"])
client.connect()
try:
response = client.send_request("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "my-client", "version": "1.0.0"}
})
print(response)
finally:
client.disconnect()
The __enter__ and __exit__ methods let you
use the client with Python's with statement, ensuring
proper cleanup even if exceptions occur.