tornadomcp by henryxchen - MCP Server

Tornado-MCP: Model Context Protocol SDK for Python 3.7

A complete Model Context Protocol (MCP) server implementation built for Python 3.7.9 compatibility using the Tornado web framework and Pydantic V1.

Features

🚀 Full MCP Protocol Support: Complete implementation of MCP specification with tools, resources, prompts, and sampling
🔧 Multiple Transport Layers: STDIO, WebSocket, SSE, and Streamable HTTP transports
📊 Structured Output: Automatic JSON schema generation from Python type annotations using Pydantic V1
🤖 LLM Sampling: Built-in support for requesting text generation from connected LLMs
🎯 Python 3.7 Compatible: Designed specifically for older Python environments with full compatibility
⚡ High Performance: Built on Tornado's async architecture for excellent performance
🛠️ Developer Friendly: Decorator-based API similar to FastMCP with intuitive patterns

Quick Start

Installation

pip install -r requirements.txt

Basic Server Example

from mcp_tornado import TornadoMCP
from pydantic import BaseModel

# Create server instance
mcp = TornadoMCP("My MCP Server")

# Simple tool with structured output
class WeatherData(BaseModel):
    temperature: float
    conditions: str

@mcp.tool("get_weather")
def get_weather(city: str) -> WeatherData:
    """Get weather information for a city"""
    return WeatherData(temperature=22.5, conditions="sunny")

# Resource with template
@mcp.resource("file://{path}")
def read_file(path: str) -> str:
    """Read contents of a file"""
    with open(path, 'r') as f:
        return f.read()

# LLM sampling tool
@mcp.tool("analyze_text")
async def analyze_text(text: str, ctx) -> str:
    """Analyze text using connected LLM"""
    response = await ctx.create_message(
        f"Please analyze this text: {text}",
        max_tokens=500
    )
    return response.get('content', 'Analysis complete')

if __name__ == "__main__":
    # Run with STDIO transport (for Claude Desktop)
    mcp.run_stdio()
    
    # Or run with HTTP transports
    # mcp.run_server(host="localhost", port=8000)

Architecture

Transport Layers

STDIO: Perfect for Claude Desktop integration and CLI tools
WebSocket: Real-time bidirectional communication with persistent connections
SSE (Server-Sent Events): Browser-compatible streaming with HTTP POST
Streamable HTTP: Modern HTTP-based streaming for web deployments

Core Components

TornadoMCP: High-level server class with decorator-based API
Context System: Request-scoped data and session management
Manager Pattern: Organized tool, resource, and prompt management
Structured Output: Automatic JSON schema generation from type hints
Sampling API: Built-in LLM interaction capabilities

Advanced Features

Structured Output with Pydantic V1

from pydantic import BaseModel
from typing import List, Optional

class UserProfile(BaseModel):
    name: str
    email: str
    age: Optional[int] = None
    tags: List[str] = []

@mcp.tool("create_user")
def create_user(name: str, email: str) -> UserProfile:
    """Create a new user profile"""
    return UserProfile(name=name, email=email, tags=["new"])

LLM Sampling Integration

@mcp.tool("smart_summary")
async def smart_summary(text: str, ctx) -> str:
    """Generate an intelligent summary using LLM"""
    response = await ctx.create_message(
        messages=[
            {"role": "system", "content": "Summarize concisely"},
            {"role": "user", "content": "text"}
        ],
        max_tokens=200,
        temperature=0.3
    )
    return response['content']

Resource Templates with URI Patterns

@mcp.resource("github://repos/{owner}/{repo}/issues/{issue_id}")
def get_github_issue(owner: str, repo: str, issue_id: str) -> str:
    """Fetch GitHub issue details"""
    # URI template automatically extracts parameters
    return f"Issue #{issue_id} in {owner}/{repo}"

@mcp.resource("database://users/{user_id}")
def get_user_data(user_id: str) -> dict:
    """Get user data from database"""
    return {"id": user_id, "status": "active"}

Transport Usage

STDIO (Claude Desktop)

# Perfect for Claude Desktop integration
mcp.run_stdio()

HTTP Server

# Run HTTP server with multiple transports
mcp.run_server(
    host="localhost",
    port=8000,
    transports=["sse", "streamable_http", "websocket"]
)

WebSocket Only

# WebSocket-only server for real-time apps
mcp.run_websocket(host="localhost", port=8001)

Python 3.7 Compatibility

This implementation is specifically designed for Python 3.7.9 environments:

Pydantic V1: Full compatibility with older Pydantic versions
Typing Extensions: Uses typing_extensions for advanced type hints
Legacy Syntax: Compatible with Python 3.7 syntax requirements
Tornado 6.1: Leverages modern async/await patterns

Testing

# Run the test suite
python -m pytest tests/ -v

Documentation

For detailed technical documentation, see TECHNICAL_DOCUMENTATION.md which covers:

Complete architecture overview
Implementation details
Performance considerations
Advanced usage patterns
Troubleshooting guide

License

MIT License - see LICENSE file for details.