MCP Server Hero
A professional Model Context Protocol (MCP) server template and framework for building robust MCP servers in Python.
Features
- 🏗️ Modular Architecture: Clean separation of tools, resources, prompts, and configuration
- 🔧 Easy Registration: Simple APIs for registering tools, resources, and prompts
- 📝 Type Safety: Full type hints and validation throughout
- 🚀 Multiple Transports: Support for stdio and SSE (Server-Sent Events) transports
- 📊 Professional Logging: Built-in logging and debugging support
- 🧪 Testing Ready: Structured for easy testing with pytest
- 📖 Comprehensive Examples: Both basic and advanced usage examples
Enterprise Features
- ⚡ Middleware System: Request/response processing pipeline with validation, logging, timing, and rate limiting
- 🧩 Plugin System: Dynamic plugin loading with dependency management
- 🔐 Authentication & Authorization: Flexible auth providers with permission-based access control
- 💾 Caching System: Multi-layer caching with TTL support and LRU eviction
- 📊 Metrics & Monitoring: Comprehensive performance metrics and health checks
- 🛡️ Rate Limiting: Token bucket rate limiting with per-client support
Quick Start
Installation
# Install uv package manager
make install-uv
# Install all dependencies
make install
# Or install production only
make install-prod
Basic Usage
from mcp_server_hero import MCPServerHero
import anyio
from mcp.server.stdio import stdio_server
async def add_numbers(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
async def main():
# Create server instance
server = MCPServerHero(name="My Server")
# Register a tool
server.add_tool(
name="add",
tool_func=add_numbers,
description="Add two numbers",
schema={
"type": "object",
"properties": {
"a": {"type": "integer"},
"b": {"type": "integer"}
},
"required": ["a", "b"]
}
)
# Run the server
async with stdio_server() as streams:
await server.run(streams[0], streams[1], server.create_initialization_options())
if __name__ == "__main__":
anyio.run(main)
Enterprise Usage
from mcp_server_hero import MCPServerHero
from mcp_server_hero.middleware.rate_limit import RateLimitMiddleware
from mcp_server_hero.auth.base import SimpleAuthProvider
# Create enterprise server
server = MCPServerHero("Enterprise Server", debug=True)
# Add advanced features
server.add_middleware(RateLimitMiddleware(tool_limit=100))
server.enable_auth_provider(SimpleAuthProvider())
await server.load_plugins_from_directory("plugins/")
# Initialize and run
await server.initialize()
Running Examples
# Run different server examples
make run-basic # Basic function-based server
make run-advanced # Class-based server with custom components
make run-enterprise # Full-featured enterprise server
# Run with different transports
make run-server # SSE transport (web-based)
make run-stdio # Stdio transport (direct communication)
make run-debug # Debug mode with detailed logging
Development
# Code quality
make check # Run all quality checks (lint + format + typecheck)
make test # Run tests
make test-cov # Run tests with coverage
# Individual quality checks
make lint # Code linting
make format # Code formatting
make typecheck # Type checking
Architecture
src/mcp_server_hero/
├── core/ # Core server implementation
│ ├── cli.py # Command-line interface
│ ├── server.py # Main server class
│ └── version.py # Version info
├── tools/ # Tool management
├── resources/ # Resource management
├── prompts/ # Prompt management
├── middleware/ # Middleware system
│ ├── logging.py # Request/response logging
│ ├── timing.py # Performance monitoring
│ ├── validation.py # Input validation
│ └── rate_limit.py # Rate limiting
├── plugins/ # Plugin system
├── auth/ # Authentication framework
├── cache/ # Caching system
├── metrics/ # Metrics collection
├── config/ # Configuration
├── utils/ # Utilities
└── examples/ # Usage examples
Components
Tools perform actions and return results:
# Simple function tool
async def my_tool(param: str) -> str:
return f"Result: {param}"
server.add_tool("my_tool", my_tool, "Description")
# Class-based tool with advanced features
from mcp_server_hero.tools.base import BaseTool
class MyTool(BaseTool):
async def execute(self, arguments):
return [TextContent(type="text", text="Result")]
Resources
Resources provide read-only data:
# Simple function resource
async def get_data() -> str:
return "Resource data"
server.add_resource("data://example", get_data, "Example data")
Prompts
Prompts generate structured messages:
# Simple function prompt
async def create_prompt(topic: str) -> str:
return f"Please explain {topic}"
server.add_prompt("explain", create_prompt, "Explanation prompt")
Middleware
Process requests through a pipeline:
from mcp_server_hero.middleware import ValidationMiddleware, TimingMiddleware
server.add_middleware(ValidationMiddleware())
server.add_middleware(TimingMiddleware())
Plugins
Extend functionality dynamically:
from mcp_server_hero.plugins.base import BasePlugin
class MyPlugin(BasePlugin):
async def initialize(self, server):
server.add_tool("plugin_tool", self.my_tool, "Plugin tool")
Configuration
from mcp_server_hero.config import ServerSettings
settings = ServerSettings(
name="My Server",
debug=True,
log_level="DEBUG",
timeout=60.0
)
server = MCPServerHero(settings=settings)
Monitoring & Metrics
The server provides comprehensive monitoring:
# Get server statistics
stats = server.get_server_stats()
# Get performance metrics
metrics = await server.get_metrics()
# Health check
health = await server.health_check()
Project Commands
make help # Show all available commands
make examples # List example servers
make stats # Show project statistics
make clean # Clean build artifacts
Contributing
- Fork the repository
- Create a feature branch
- Make your changes
- Run
make check to verify code quality - Add tests if applicable
- Submit a pull request
License
MIT License - see LICENSE file for details.
MCP Protocol
This framework implements the Model Context Protocol (MCP), enabling seamless integration between AI models and external data sources and tools.
For more information about MCP, visit the official documentation.