# mcpbytes-lambda-stdio
[](https://www.python.org/downloads/)
[](https://spec.modelcontextprotocol.io/)
[](https://opensource.org/licenses/Apache-2.0)
Ultra-minimal stdio transport adapter for [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) servers.
Enables local MCP server deployment with line-delimited JSON-RPC over stdin/stdout for Claude Desktop, VS Code MCP, and other MCP clients.
## 🚀 **Quick Start**
```python
import sys
from mcpbytes_lambda.core import MCPServer
from mcpbytes_lambda.stdio import StdioAdapter
# Create your MCP server with tools
mcp = MCPServer(name="local-server", version="1.0.0")
@mcp.tool(name="hello", description="Say hello")
def hello(name: str) -> str:
return f"Hello, {name}!"
def main():
"""Local MCP server mode - stdio transport."""
adapter = StdioAdapter()
try:
while True:
# Read JSON-RPC line from stdin
line = sys.stdin.readline()
if not line:
break
# Process: stdin → JSON-RPC → tool execution → JSON-RPC → stdout
response_json = mcp.handle(line, adapter)
# Write response to stdout
sys.stdout.write(response_json)
sys.stdout.flush()
except (KeyboardInterrupt, EOFError):
sys.exit(0)
if __name__ == "__main__":
main()
```
**Configure in Claude Desktop:**
```json
{
"mcpServers": {
"local-server": {
"command": "python3",
"args": ["/path/to/your/server.py"]
}
}
}
```
## ✨ **Features**
- **📺 Line-Delimited JSON-RPC** - Standard MCP stdio transport protocol
- **🏠 Local Development** - Perfect for Claude Desktop and local MCP clients
- **⚡ Zero Overhead** - Minimal transport layer with direct JSON passthrough
- **🔄 Dual-Mode Support** - Same tools work in both Lambda and local environments
- **🧪 Easy Testing** - Simple command-line testing with JSON inputs
- **📊 Structured Output** - Full MCP 2025-06-18 compliance
- **🔧 Developer-Friendly** - Seamless local development workflow
## 📋 **Requirements**
- Python 3.12+
- `mcpbytes-lambda-core` package
- Standard input/output streams (terminal, process pipes)
## 🏗️ **Architecture**
```
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ MCP Client │───▶│ Stdio Adapter │───▶│ MCP Core │
│ (Claude Desktop)│ │ (stdin/stdout) │ │ JSON-RPC │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Line-Delimited │ │
│ │ JSON-RPC │ │
│ │ .strip() only │ │
│ └─────────────────┘ │
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ MCP Client │◀─────────────────────────────│ Tool Results │
│ Integration │ │ JSON-RPC │
└─────────────────┘ └─────────────────┘
```
## 🛠️ **Usage Examples**
### **Basic Stdio Server**
```python
import sys
from mcpbytes_lambda.core import MCPServer, ToolResult, TextContent
from mcpbytes_lambda.stdio import StdioAdapter
mcp = MCPServer(name="my-tools", version="1.0.0")
@mcp.tool(name="file.read")
def read_file(path: str) -> ToolResult:
"""Read a text file."""
try:
with open(path, 'r') as f:
content = f.read()
return ToolResult(
content=[TextContent(text=content)],
isError=False
)
except Exception as e:
return ToolResult(
content=[TextContent(text=f"Error reading file: {e}")],
isError=True
)
def main():
adapter = StdioAdapter()
while True:
try:
line = sys.stdin.readline()
if not line:
break
response = mcp.handle(line, adapter)
sys.stdout.write(response)
sys.stdout.flush()
except (KeyboardInterrupt, EOFError):
break
if __name__ == "__main__":
main()
```
### **Dual-Mode Server (Lambda + Stdio)**
```python
import sys
from typing import Dict, Any
from mcpbytes_lambda.core import MCPServer
from mcpbytes_lambda.stdio import StdioAdapter
from mcpbytes_lambda.apigw import ApiGatewayAdapter
# Shared MCP server instance
mcp = MCPServer(name="dual-mode-server", version="1.0.0")
@mcp.tool(name="calculate")
def calculate(expression: str) -> str:
"""Safely evaluate mathematical expressions."""
try:
# Basic calculator logic here
result = eval(expression.replace('^', '**')) # Simple example
return f"{expression} = {result}"
except Exception as e:
return f"Error: {e}"
def main():
"""Local stdio mode."""
print("Starting MCP Server (stdio mode)", file=sys.stderr)
adapter = StdioAdapter()
try:
while True:
line = sys.stdin.readline()
if not line:
break
response = mcp.handle(line, adapter)
sys.stdout.write(response)
sys.stdout.flush()
except (KeyboardInterrupt, EOFError):
print("Shutting down MCP Server", file=sys.stderr)
def lambda_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
"""AWS Lambda mode."""
adapter = ApiGatewayAdapter()
if adapter.need_preflight(event):
return adapter.preflight_response(event.get("headers"))
response = mcp.handle(event, adapter, headers=event.get("headers"))
return adapter.add_cors(response, event.get("headers"))
if __name__ == "__main__":
main()
```
### **With Logging and Error Handling**
```python
import sys
import logging
from mcpbytes_lambda.core import MCPServer
from mcpbytes_lambda.stdio import StdioAdapter
# Configure logging to stderr (stdout reserved for MCP protocol)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
stream=sys.stderr
)
logger = logging.getLogger(__name__)
mcp = MCPServer(name="robust-server", version="1.0.0")
@mcp.tool(name="system.info")
def system_info() -> Dict[str, str]:
"""Get system information."""
import platform
import os
return {
"platform": platform.system(),
"python_version": platform.python_version(),
"cwd": os.getcwd(),
"pid": os.getpid()
}
def main():
logger.info("Starting MCP Server (stdio mode)")
adapter = StdioAdapter()
try:
while True:
line = sys.stdin.readline()
if not line:
logger.info("EOF received, shutting down")
break
line = line.strip()
if not line:
continue
logger.debug(f"Processing request: {line[:100]}...")
try:
response = mcp.handle(line, adapter)
sys.stdout.write(response)
sys.stdout.flush()
logger.debug("Response sent successfully")
except Exception as e:
logger.error(f"Request processing failed: {e}", exc_info=True)
# Send error response
error_response = {
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32603, "message": "Internal error"}
}
sys.stdout.write(json.dumps(error_response) + '\n')
sys.stdout.flush()
except (KeyboardInterrupt, EOFError):
logger.info("Shutting down MCP Server")
except Exception as e:
logger.error(f"Fatal error: {e}", exc_info=True)
sys.exit(1)
if __name__ == "__main__":
main()
```
## 🖥️ **Client Configuration**
### **Claude Desktop**
Add to `~/.config/claude_desktop_config.json` (macOS/Linux) or `%AppData%\Claude\claude_desktop_config.json` (Windows):
```json
{
"mcpServers": {
"my-mcp-server": {
"command": "python3",
"args": ["/full/path/to/your/server.py"],
"env": {
"PYTHONPATH": "/path/to/your/packages"
}
}
}
}
```
**With virtual environment:**
```json
{
"mcpServers": {
"my-server": {
"command": "/path/to/venv/bin/python",
"args": ["/path/to/server.py"]
}
}
}
```
**Using uv:**
```json
{
"mcpServers": {
"my-server": {
"command": "uv",
"args": ["run", "--directory", "/path/to/project", "python", "server.py"]
}
}
}
```
### **VS Code MCP Extension**
Add to `.vscode/mcp.json`:
```json
{
"servers": {
"my-mcp-server": {
"command": "python3",
"args": ["/path/to/server.py"],
"cwd": "/path/to/project"
}
}
}
```
### **Custom MCP Client**
```python
import subprocess
import json
# Start MCP server process
process = subprocess.Popen(
["python3", "server.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# Send initialization
request = {
"jsonrpc": "2.0",
"method": "initialize",
"id": "init-1"
}
process.stdin.write(json.dumps(request) + '\n')
process.stdin.flush()
# Read response
response = process.stdout.readline()
result = json.loads(response)
print("Initialize result:", result)
# List tools
request = {
"jsonrpc": "2.0",
"method": "tools/list",
"id": "list-1"
}
process.stdin.write(json.dumps(request) + '\n')
process.stdin.flush()
response = process.stdout.readline()
tools = json.loads(response)
print("Available tools:", tools)
# Clean shutdown
process.stdin.close()
process.wait()
```
## 🧪 **Testing**
### **Command Line Testing**
Create `test_requests.jsonl`:
```json
{"jsonrpc": "2.0", "method": "initialize", "id": "1"}
{"jsonrpc": "2.0", "method": "tools/list", "id": "2"}
{"jsonrpc": "2.0", "method": "tools/call", "params": {"name": "hello", "arguments": {"name": "World"}}, "id": "3"}
```
Test your server:
```bash
python3 server.py < test_requests.jsonl
```
### **Interactive Testing**
```bash
# Start server
python3 server.py
# Type JSON-RPC requests (press Enter after each):
{"jsonrpc": "2.0", "method": "initialize", "id": "1"}
{"jsonrpc": "2.0", "method": "tools/list", "id": "2"}
{"jsonrpc": "2.0", "method": "tools/call", "params": {"name": "my.tool", "arguments": {"param": "value"}}, "id": "3"}
# Press Ctrl+C or Ctrl+D to exit
```
### **Unit Tests**
```python
import io
import sys
from contextlib import redirect_stdout, redirect_stdin
def test_stdio_adapter():
from mcpbytes_lambda.stdio import StdioAdapter
adapter = StdioAdapter()
# Test line processing
line = '{"jsonrpc": "2.0", "method": "initialize", "id": "1"}\n'
core_request = adapter.to_core_request(line)
assert core_request == '{"jsonrpc": "2.0", "method": "initialize", "id": "1"}'
# Test response formatting
response = {"jsonrpc": "2.0", "id": "1", "result": {"test": True}}
output = adapter.from_core_response(response)
assert output.endswith('\n')
assert '"test": true' in output.lower()
def test_server_integration():
# Mock stdin/stdout for testing
input_data = '{"jsonrpc": "2.0", "method": "initialize", "id": "test"}\n'
with redirect_stdin(io.StringIO(input_data)):
with redirect_stdout(io.StringIO()) as output:
# Run one iteration of server loop
try:
line = sys.stdin.readline()
if line:
response = mcp.handle(line, StdioAdapter())
sys.stdout.write(response)
except EOFError:
pass
result = output.getvalue()
assert '"jsonrpc": "2.0"' in result
assert '"id": "test"' in result
```
### **Integration Testing with Process**
```python
import subprocess
import json
import time
def test_mcp_server_process():
# Start server process
process = subprocess.Popen(
["python3", "server.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
# Test initialize
init_request = {
"jsonrpc": "2.0",
"method": "initialize",
"id": "test-init"
}
process.stdin.write(json.dumps(init_request) + '\n')
process.stdin.flush()
# Read response with timeout
response_line = process.stdout.readline()
assert response_line, "No response received"
response = json.loads(response_line)
assert response["jsonrpc"] == "2.0"
assert response["id"] == "test-init"
assert "result" in response
finally:
process.terminate()
process.wait(timeout=5)
```
## 🚀 **Development Workflow**
### **Project Structure**
```
my-mcp-server/
├── server.py # Main server file
├── requirements.txt # Dependencies
├── test_requests.jsonl # Test cases
├── .env # Environment variables
├── README.md # Documentation
└── tests/
├── __init__.py
├── test_tools.py
└── test_server.py
```
### **Development Script**
```python
#!/usr/bin/env python3
"""Development helper for MCP server."""
import sys
import json
import argparse
from pathlib import Path
def create_test_request(method: str, params: dict = None, id_val: str = "1") -> str:
"""Create a JSON-RPC test request."""
request = {
"jsonrpc": "2.0",
"method": method,
"id": id_val
}
if params:
request["params"] = params
return json.dumps(request)
def main():
parser = argparse.ArgumentParser(description="MCP Server Development Helper")
parser.add_argument("--test", action="store_true", help="Run test requests")
parser.add_argument("--method", help="Test specific method")
parser.add_argument("--params", help="JSON parameters for method")
args = parser.parse_args()
if args.test:
# Generate common test requests
requests = [
create_test_request("initialize", id_val="init-1"),
create_test_request("tools/list", id_val="list-1"),
]
if args.method:
params = json.loads(args.params) if args.params else {}
requests.append(create_test_request(args.method, params, "custom-1"))
for req in requests:
print(req)
else:
# Run the actual server
from server import main as server_main
server_main()
if __name__ == "__main__":
main()
```
### **Debugging Tips**
1. **Use stderr for logging** (stdout is reserved for MCP protocol)
2. **Test with simple JSON files** before integrating with clients
3. **Validate JSON-RPC format** - clients are strict about compliance
4. **Handle EOF gracefully** - clients may close stdin unexpectedly
5. **Flush stdout** after each response to ensure delivery
## 📊 **Performance Considerations**
### **Memory Usage**
- **Minimal overhead** - No HTTP parsing or transport layers
- **Streaming processing** - Handle one request at a time
- **No buffering** - Direct stdin/stdout interaction
### **Latency**
- **Near-zero transport overhead** - Simple string operations
- **Direct JSON processing** - No intermediate representations
- **Immediate response** - No batching or queuing
### **Reliability**
- **Process isolation** - Each client gets its own server process
- **Automatic cleanup** - Process termination cleans up resources
- **Simple error handling** - Failures are contained to single requests
## 🤝 **Contributing**
1. Fork the repository
2. Create a feature branch: `git checkout -b feature-name`
3. Make your changes and add tests
4. Run tests: `python -m pytest`
5. Test with actual MCP clients (Claude Desktop, etc.)
6. Submit a pull request
## 📄 **License**
Apache 2.0 License - see [LICENSE](../../LICENSE) for details.
## 🔗 **Related Packages**
- [`mcpbytes-lambda-core`](../core/) - Transport-agnostic MCP server core
- [`mcpbytes-lambda-apigw`](../apigw/) - API Gateway transport adapter
- [`mcpbytes-lambda-invoke`](../invoke/) - Direct Lambda invocation adapter
## 📚 **Documentation**
- [MCP Specification](https://spec.modelcontextprotocol.io/)
- [MCP Stdio Transport](https://spec.modelcontextprotocol.io/specification/basic/transports/#stdio)
- [Claude Desktop MCP Guide](https://docs.anthropic.com/claude/docs/mcp)
- [Project Examples](../../examples/)
---
Built with ❤️ for the MCP ecosystem
Raw data
{
"_id": null,
"home_page": null,
"name": "mcpbytes-lambda-stdio",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": "mcp, model-context-protocol, stdio, transport",
"author": null,
"author_email": "\"MCPBytes.com\" <hello@mcpbytes.com>",
"download_url": "https://files.pythonhosted.org/packages/70/58/d86d2156f08ec66be7cb4b110111ad5f63343f83ac6fcf1588d26af123ac/mcpbytes_lambda_stdio-0.1.3.tar.gz",
"platform": null,
"description": "# mcpbytes-lambda-stdio\n\n[](https://www.python.org/downloads/)\n[](https://spec.modelcontextprotocol.io/)\n[](https://opensource.org/licenses/Apache-2.0)\n\nUltra-minimal stdio transport adapter for [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) servers. \nEnables local MCP server deployment with line-delimited JSON-RPC over stdin/stdout for Claude Desktop, VS Code MCP, and other MCP clients.\n\n## \ud83d\ude80 **Quick Start**\n\n```python\nimport sys\nfrom mcpbytes_lambda.core import MCPServer\nfrom mcpbytes_lambda.stdio import StdioAdapter\n\n# Create your MCP server with tools\nmcp = MCPServer(name=\"local-server\", version=\"1.0.0\")\n\n@mcp.tool(name=\"hello\", description=\"Say hello\")\ndef hello(name: str) -> str:\n return f\"Hello, {name}!\"\n\ndef main():\n \"\"\"Local MCP server mode - stdio transport.\"\"\"\n adapter = StdioAdapter()\n \n try:\n while True:\n # Read JSON-RPC line from stdin\n line = sys.stdin.readline()\n if not line:\n break\n \n # Process: stdin \u2192 JSON-RPC \u2192 tool execution \u2192 JSON-RPC \u2192 stdout\n response_json = mcp.handle(line, adapter)\n \n # Write response to stdout\n sys.stdout.write(response_json)\n sys.stdout.flush()\n \n except (KeyboardInterrupt, EOFError):\n sys.exit(0)\n\nif __name__ == \"__main__\":\n main()\n```\n\n**Configure in Claude Desktop:**\n```json\n{\n \"mcpServers\": {\n \"local-server\": {\n \"command\": \"python3\",\n \"args\": [\"/path/to/your/server.py\"]\n }\n }\n}\n```\n\n## \u2728 **Features**\n\n- **\ud83d\udcfa Line-Delimited JSON-RPC** - Standard MCP stdio transport protocol\n- **\ud83c\udfe0 Local Development** - Perfect for Claude Desktop and local MCP clients\n- **\u26a1 Zero Overhead** - Minimal transport layer with direct JSON passthrough\n- **\ud83d\udd04 Dual-Mode Support** - Same tools work in both Lambda and local environments\n- **\ud83e\uddea Easy Testing** - Simple command-line testing with JSON inputs\n- **\ud83d\udcca Structured Output** - Full MCP 2025-06-18 compliance\n- **\ud83d\udd27 Developer-Friendly** - Seamless local development workflow\n\n## \ud83d\udccb **Requirements**\n\n- Python 3.12+\n- `mcpbytes-lambda-core` package\n- Standard input/output streams (terminal, process pipes)\n\n## \ud83c\udfd7\ufe0f **Architecture**\n\n```\n\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n\u2502 MCP Client \u2502\u2500\u2500\u2500\u25b6\u2502 Stdio Adapter \u2502\u2500\u2500\u2500\u25b6\u2502 MCP Core \u2502\n\u2502 (Claude Desktop)\u2502 \u2502 (stdin/stdout) \u2502 \u2502 JSON-RPC \u2502\n\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n \u2502 \u2502 \u2502\n \u2502 \u25bc \u2502\n \u2502 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u2502\n \u2502 \u2502 Line-Delimited \u2502 \u2502\n \u2502 \u2502 JSON-RPC \u2502 \u2502\n \u2502 \u2502 .strip() only \u2502 \u2502\n \u2502 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2502\n \u2502 \u2502\n \u25bc \u25bc\n\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n\u2502 MCP Client \u2502\u25c0\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2502 Tool Results \u2502\n\u2502 Integration \u2502 \u2502 JSON-RPC \u2502\n\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n```\n\n## \ud83d\udee0\ufe0f **Usage Examples**\n\n### **Basic Stdio Server**\n\n```python\nimport sys\nfrom mcpbytes_lambda.core import MCPServer, ToolResult, TextContent\nfrom mcpbytes_lambda.stdio import StdioAdapter\n\nmcp = MCPServer(name=\"my-tools\", version=\"1.0.0\")\n\n@mcp.tool(name=\"file.read\")\ndef read_file(path: str) -> ToolResult:\n \"\"\"Read a text file.\"\"\"\n try:\n with open(path, 'r') as f:\n content = f.read()\n return ToolResult(\n content=[TextContent(text=content)],\n isError=False\n )\n except Exception as e:\n return ToolResult(\n content=[TextContent(text=f\"Error reading file: {e}\")],\n isError=True\n )\n\ndef main():\n adapter = StdioAdapter()\n \n while True:\n try:\n line = sys.stdin.readline()\n if not line:\n break\n \n response = mcp.handle(line, adapter)\n sys.stdout.write(response)\n sys.stdout.flush()\n \n except (KeyboardInterrupt, EOFError):\n break\n\nif __name__ == \"__main__\":\n main()\n```\n\n### **Dual-Mode Server (Lambda + Stdio)**\n\n```python\nimport sys\nfrom typing import Dict, Any\nfrom mcpbytes_lambda.core import MCPServer\nfrom mcpbytes_lambda.stdio import StdioAdapter\nfrom mcpbytes_lambda.apigw import ApiGatewayAdapter\n\n# Shared MCP server instance\nmcp = MCPServer(name=\"dual-mode-server\", version=\"1.0.0\")\n\n@mcp.tool(name=\"calculate\")\ndef calculate(expression: str) -> str:\n \"\"\"Safely evaluate mathematical expressions.\"\"\"\n try:\n # Basic calculator logic here\n result = eval(expression.replace('^', '**')) # Simple example\n return f\"{expression} = {result}\"\n except Exception as e:\n return f\"Error: {e}\"\n\ndef main():\n \"\"\"Local stdio mode.\"\"\"\n print(\"Starting MCP Server (stdio mode)\", file=sys.stderr)\n adapter = StdioAdapter()\n \n try:\n while True:\n line = sys.stdin.readline()\n if not line:\n break\n \n response = mcp.handle(line, adapter)\n sys.stdout.write(response)\n sys.stdout.flush()\n \n except (KeyboardInterrupt, EOFError):\n print(\"Shutting down MCP Server\", file=sys.stderr)\n\ndef lambda_handler(event: Dict[str, Any], context) -> Dict[str, Any]:\n \"\"\"AWS Lambda mode.\"\"\"\n adapter = ApiGatewayAdapter()\n \n if adapter.need_preflight(event):\n return adapter.preflight_response(event.get(\"headers\"))\n \n response = mcp.handle(event, adapter, headers=event.get(\"headers\"))\n return adapter.add_cors(response, event.get(\"headers\"))\n\nif __name__ == \"__main__\":\n main()\n```\n\n### **With Logging and Error Handling**\n\n```python\nimport sys\nimport logging\nfrom mcpbytes_lambda.core import MCPServer\nfrom mcpbytes_lambda.stdio import StdioAdapter\n\n# Configure logging to stderr (stdout reserved for MCP protocol)\nlogging.basicConfig(\n level=logging.INFO,\n format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',\n stream=sys.stderr\n)\nlogger = logging.getLogger(__name__)\n\nmcp = MCPServer(name=\"robust-server\", version=\"1.0.0\")\n\n@mcp.tool(name=\"system.info\")\ndef system_info() -> Dict[str, str]:\n \"\"\"Get system information.\"\"\"\n import platform\n import os\n \n return {\n \"platform\": platform.system(),\n \"python_version\": platform.python_version(),\n \"cwd\": os.getcwd(),\n \"pid\": os.getpid()\n }\n\ndef main():\n logger.info(\"Starting MCP Server (stdio mode)\")\n adapter = StdioAdapter()\n \n try:\n while True:\n line = sys.stdin.readline()\n if not line:\n logger.info(\"EOF received, shutting down\")\n break\n \n line = line.strip()\n if not line:\n continue\n \n logger.debug(f\"Processing request: {line[:100]}...\")\n \n try:\n response = mcp.handle(line, adapter)\n sys.stdout.write(response)\n sys.stdout.flush()\n logger.debug(\"Response sent successfully\")\n \n except Exception as e:\n logger.error(f\"Request processing failed: {e}\", exc_info=True)\n # Send error response\n error_response = {\n \"jsonrpc\": \"2.0\",\n \"id\": None,\n \"error\": {\"code\": -32603, \"message\": \"Internal error\"}\n }\n sys.stdout.write(json.dumps(error_response) + '\\n')\n sys.stdout.flush()\n \n except (KeyboardInterrupt, EOFError):\n logger.info(\"Shutting down MCP Server\")\n except Exception as e:\n logger.error(f\"Fatal error: {e}\", exc_info=True)\n sys.exit(1)\n\nif __name__ == \"__main__\":\n main()\n```\n\n## \ud83d\udda5\ufe0f **Client Configuration**\n\n### **Claude Desktop**\n\nAdd to `~/.config/claude_desktop_config.json` (macOS/Linux) or `%AppData%\\Claude\\claude_desktop_config.json` (Windows):\n\n```json\n{\n \"mcpServers\": {\n \"my-mcp-server\": {\n \"command\": \"python3\",\n \"args\": [\"/full/path/to/your/server.py\"],\n \"env\": {\n \"PYTHONPATH\": \"/path/to/your/packages\"\n }\n }\n }\n}\n```\n\n**With virtual environment:**\n```json\n{\n \"mcpServers\": {\n \"my-server\": {\n \"command\": \"/path/to/venv/bin/python\",\n \"args\": [\"/path/to/server.py\"]\n }\n }\n}\n```\n\n**Using uv:**\n```json\n{\n \"mcpServers\": {\n \"my-server\": {\n \"command\": \"uv\",\n \"args\": [\"run\", \"--directory\", \"/path/to/project\", \"python\", \"server.py\"]\n }\n }\n}\n```\n\n### **VS Code MCP Extension**\n\nAdd to `.vscode/mcp.json`:\n\n```json\n{\n \"servers\": {\n \"my-mcp-server\": {\n \"command\": \"python3\",\n \"args\": [\"/path/to/server.py\"],\n \"cwd\": \"/path/to/project\"\n }\n }\n}\n```\n\n### **Custom MCP Client**\n\n```python\nimport subprocess\nimport json\n\n# Start MCP server process\nprocess = subprocess.Popen(\n [\"python3\", \"server.py\"],\n stdin=subprocess.PIPE,\n stdout=subprocess.PIPE,\n stderr=subprocess.PIPE,\n text=True\n)\n\n# Send initialization\nrequest = {\n \"jsonrpc\": \"2.0\",\n \"method\": \"initialize\",\n \"id\": \"init-1\"\n}\n\nprocess.stdin.write(json.dumps(request) + '\\n')\nprocess.stdin.flush()\n\n# Read response\nresponse = process.stdout.readline()\nresult = json.loads(response)\nprint(\"Initialize result:\", result)\n\n# List tools\nrequest = {\n \"jsonrpc\": \"2.0\",\n \"method\": \"tools/list\",\n \"id\": \"list-1\"\n}\n\nprocess.stdin.write(json.dumps(request) + '\\n')\nprocess.stdin.flush()\n\nresponse = process.stdout.readline()\ntools = json.loads(response)\nprint(\"Available tools:\", tools)\n\n# Clean shutdown\nprocess.stdin.close()\nprocess.wait()\n```\n\n## \ud83e\uddea **Testing**\n\n### **Command Line Testing**\n\nCreate `test_requests.jsonl`:\n```json\n{\"jsonrpc\": \"2.0\", \"method\": \"initialize\", \"id\": \"1\"}\n{\"jsonrpc\": \"2.0\", \"method\": \"tools/list\", \"id\": \"2\"}\n{\"jsonrpc\": \"2.0\", \"method\": \"tools/call\", \"params\": {\"name\": \"hello\", \"arguments\": {\"name\": \"World\"}}, \"id\": \"3\"}\n```\n\nTest your server:\n```bash\npython3 server.py < test_requests.jsonl\n```\n\n### **Interactive Testing**\n\n```bash\n# Start server\npython3 server.py\n\n# Type JSON-RPC requests (press Enter after each):\n{\"jsonrpc\": \"2.0\", \"method\": \"initialize\", \"id\": \"1\"}\n{\"jsonrpc\": \"2.0\", \"method\": \"tools/list\", \"id\": \"2\"}\n{\"jsonrpc\": \"2.0\", \"method\": \"tools/call\", \"params\": {\"name\": \"my.tool\", \"arguments\": {\"param\": \"value\"}}, \"id\": \"3\"}\n\n# Press Ctrl+C or Ctrl+D to exit\n```\n\n### **Unit Tests**\n\n```python\nimport io\nimport sys\nfrom contextlib import redirect_stdout, redirect_stdin\n\ndef test_stdio_adapter():\n from mcpbytes_lambda.stdio import StdioAdapter\n \n adapter = StdioAdapter()\n \n # Test line processing\n line = '{\"jsonrpc\": \"2.0\", \"method\": \"initialize\", \"id\": \"1\"}\\n'\n core_request = adapter.to_core_request(line)\n assert core_request == '{\"jsonrpc\": \"2.0\", \"method\": \"initialize\", \"id\": \"1\"}'\n \n # Test response formatting\n response = {\"jsonrpc\": \"2.0\", \"id\": \"1\", \"result\": {\"test\": True}}\n output = adapter.from_core_response(response)\n assert output.endswith('\\n')\n assert '\"test\": true' in output.lower()\n\ndef test_server_integration():\n # Mock stdin/stdout for testing\n input_data = '{\"jsonrpc\": \"2.0\", \"method\": \"initialize\", \"id\": \"test\"}\\n'\n \n with redirect_stdin(io.StringIO(input_data)):\n with redirect_stdout(io.StringIO()) as output:\n # Run one iteration of server loop\n try:\n line = sys.stdin.readline()\n if line:\n response = mcp.handle(line, StdioAdapter())\n sys.stdout.write(response)\n except EOFError:\n pass\n \n result = output.getvalue()\n assert '\"jsonrpc\": \"2.0\"' in result\n assert '\"id\": \"test\"' in result\n```\n\n### **Integration Testing with Process**\n\n```python\nimport subprocess\nimport json\nimport time\n\ndef test_mcp_server_process():\n # Start server process\n process = subprocess.Popen(\n [\"python3\", \"server.py\"],\n stdin=subprocess.PIPE,\n stdout=subprocess.PIPE,\n stderr=subprocess.PIPE,\n text=True\n )\n \n try:\n # Test initialize\n init_request = {\n \"jsonrpc\": \"2.0\",\n \"method\": \"initialize\",\n \"id\": \"test-init\"\n }\n \n process.stdin.write(json.dumps(init_request) + '\\n')\n process.stdin.flush()\n \n # Read response with timeout\n response_line = process.stdout.readline()\n assert response_line, \"No response received\"\n \n response = json.loads(response_line)\n assert response[\"jsonrpc\"] == \"2.0\"\n assert response[\"id\"] == \"test-init\"\n assert \"result\" in response\n \n finally:\n process.terminate()\n process.wait(timeout=5)\n```\n\n## \ud83d\ude80 **Development Workflow**\n\n### **Project Structure**\n\n```\nmy-mcp-server/\n\u251c\u2500\u2500 server.py # Main server file\n\u251c\u2500\u2500 requirements.txt # Dependencies\n\u251c\u2500\u2500 test_requests.jsonl # Test cases\n\u251c\u2500\u2500 .env # Environment variables\n\u251c\u2500\u2500 README.md # Documentation\n\u2514\u2500\u2500 tests/\n \u251c\u2500\u2500 __init__.py\n \u251c\u2500\u2500 test_tools.py\n \u2514\u2500\u2500 test_server.py\n```\n\n### **Development Script**\n\n```python\n#!/usr/bin/env python3\n\"\"\"Development helper for MCP server.\"\"\"\n\nimport sys\nimport json\nimport argparse\nfrom pathlib import Path\n\ndef create_test_request(method: str, params: dict = None, id_val: str = \"1\") -> str:\n \"\"\"Create a JSON-RPC test request.\"\"\"\n request = {\n \"jsonrpc\": \"2.0\",\n \"method\": method,\n \"id\": id_val\n }\n if params:\n request[\"params\"] = params\n return json.dumps(request)\n\ndef main():\n parser = argparse.ArgumentParser(description=\"MCP Server Development Helper\")\n parser.add_argument(\"--test\", action=\"store_true\", help=\"Run test requests\")\n parser.add_argument(\"--method\", help=\"Test specific method\")\n parser.add_argument(\"--params\", help=\"JSON parameters for method\")\n \n args = parser.parse_args()\n \n if args.test:\n # Generate common test requests\n requests = [\n create_test_request(\"initialize\", id_val=\"init-1\"),\n create_test_request(\"tools/list\", id_val=\"list-1\"),\n ]\n \n if args.method:\n params = json.loads(args.params) if args.params else {}\n requests.append(create_test_request(args.method, params, \"custom-1\"))\n \n for req in requests:\n print(req)\n else:\n # Run the actual server\n from server import main as server_main\n server_main()\n\nif __name__ == \"__main__\":\n main()\n```\n\n### **Debugging Tips**\n\n1. **Use stderr for logging** (stdout is reserved for MCP protocol)\n2. **Test with simple JSON files** before integrating with clients\n3. **Validate JSON-RPC format** - clients are strict about compliance\n4. **Handle EOF gracefully** - clients may close stdin unexpectedly\n5. **Flush stdout** after each response to ensure delivery\n\n## \ud83d\udcca **Performance Considerations**\n\n### **Memory Usage**\n- **Minimal overhead** - No HTTP parsing or transport layers\n- **Streaming processing** - Handle one request at a time\n- **No buffering** - Direct stdin/stdout interaction\n\n### **Latency**\n- **Near-zero transport overhead** - Simple string operations\n- **Direct JSON processing** - No intermediate representations\n- **Immediate response** - No batching or queuing\n\n### **Reliability**\n- **Process isolation** - Each client gets its own server process\n- **Automatic cleanup** - Process termination cleans up resources\n- **Simple error handling** - Failures are contained to single requests\n\n## \ud83e\udd1d **Contributing**\n\n1. Fork the repository\n2. Create a feature branch: `git checkout -b feature-name`\n3. Make your changes and add tests\n4. Run tests: `python -m pytest`\n5. Test with actual MCP clients (Claude Desktop, etc.)\n6. Submit a pull request\n\n## \ud83d\udcc4 **License**\n\nApache 2.0 License - see [LICENSE](../../LICENSE) for details.\n\n## \ud83d\udd17 **Related Packages**\n\n- [`mcpbytes-lambda-core`](../core/) - Transport-agnostic MCP server core\n- [`mcpbytes-lambda-apigw`](../apigw/) - API Gateway transport adapter\n- [`mcpbytes-lambda-invoke`](../invoke/) - Direct Lambda invocation adapter\n\n## \ud83d\udcda **Documentation**\n\n- [MCP Specification](https://spec.modelcontextprotocol.io/)\n- [MCP Stdio Transport](https://spec.modelcontextprotocol.io/specification/basic/transports/#stdio)\n- [Claude Desktop MCP Guide](https://docs.anthropic.com/claude/docs/mcp)\n- [Project Examples](../../examples/)\n\n---\n\nBuilt with \u2764\ufe0f for the MCP ecosystem\n",
"bugtrack_url": null,
"license": null,
"summary": "Stdio transport adapter for mcpbytes-lambda MCP servers",
"version": "0.1.3",
"project_urls": {
"Documentation": "https://github.com/MCPBytes/mcpbytes-lambda#readme",
"Homepage": "https://mcpbytes.com/",
"Issues": "https://github.com/MCPBytes/mcpbytes-lambda/issues",
"Repository": "https://github.com/MCPBytes/mcpbytes-lambda"
},
"split_keywords": [
"mcp",
" model-context-protocol",
" stdio",
" transport"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "034405eb7f76932dd98b4739aee1d088d54b15f9c821991d0120daa172f683c0",
"md5": "248ea0b61ef40a593ab729abf95d5b33",
"sha256": "72521febb096efaa5d0b85ed82f09844b9cbc911c6ae0b819bbdcb78d00cd988"
},
"downloads": -1,
"filename": "mcpbytes_lambda_stdio-0.1.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "248ea0b61ef40a593ab729abf95d5b33",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 10122,
"upload_time": "2025-09-01T20:12:11",
"upload_time_iso_8601": "2025-09-01T20:12:11.492758Z",
"url": "https://files.pythonhosted.org/packages/03/44/05eb7f76932dd98b4739aee1d088d54b15f9c821991d0120daa172f683c0/mcpbytes_lambda_stdio-0.1.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "7058d86d2156f08ec66be7cb4b110111ad5f63343f83ac6fcf1588d26af123ac",
"md5": "acaaa1003df99cdfe6a286a5fb68a26c",
"sha256": "0cff2d88ecf63372bfdc162cc48040f1af2ecf2a9f5b77a0a6ac048bf7ed5f96"
},
"downloads": -1,
"filename": "mcpbytes_lambda_stdio-0.1.3.tar.gz",
"has_sig": false,
"md5_digest": "acaaa1003df99cdfe6a286a5fb68a26c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 9210,
"upload_time": "2025-09-01T20:12:13",
"upload_time_iso_8601": "2025-09-01T20:12:13.250676Z",
"url": "https://files.pythonhosted.org/packages/70/58/d86d2156f08ec66be7cb4b110111ad5f63343f83ac6fcf1588d26af123ac/mcpbytes_lambda_stdio-0.1.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-01 20:12:13",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "MCPBytes",
"github_project": "mcpbytes-lambda#readme",
"github_not_found": true,
"lcname": "mcpbytes-lambda-stdio"
}
JSON
