Production-ready Model Context Protocol (MCP) server with file, shell, and search tools.
MCP enables AI assistants to interact with external tools and data sources through a standardized interface. This server provides a complete implementation with built-in tools for common operations.
pip install -r requirements.txtimport asyncio
from mcp_server import MCPServer, ServerConfig, create_server
# Create server with default tools
server = create_server()
# Or customize configuration
config = ServerConfig(
name="my-mcp-server",
allowed_paths=["/home/user/projects"],
enable_shell_tools=True,
)
server = create_server(config=config)
# Run the server
asyncio.run(server.run())Read file contents with path restrictions and size limits.
{
"name": "file_read",
"arguments": {
"path": "/path/to/file.txt",
"encoding": "utf-8"
}
}Write or append content to files.
{
"name": "file_write",
"arguments": {
"path": "/path/to/file.txt",
"content": "Hello, world!",
"mode": "write"
}
}Execute shell commands with security controls.
{
"name": "shell",
"arguments": {
"command": "ls -la",
"timeout": 30
}
}Search for patterns in files (grep-like).
{
"name": "search",
"arguments": {
"pattern": "TODO",
"path": "/path/to/project",
"include": "*.py",
"ignore_case": true
}
}Find files matching a glob pattern.
{
"name": "glob",
"arguments": {
"pattern": "**/*.py",
"path": "/path/to/project"
}
}Create your own tools by extending BaseTool:
from mcp_server import BaseTool, ToolParameter, ToolResult
class WeatherTool(BaseTool):
@property
def name(self) -> str:
return "weather"
@property
def description(self) -> str:
return "Get current weather for a location"
@property
def parameters(self):
return [
ToolParameter("location", "string", "City name", required=True),
]
def execute(self, location: str) -> ToolResult:
# Your implementation here
return ToolResult(
success=True,
content=f"Weather for {location}: Sunny, 72F",
)
# Register with server
server.register_tool(WeatherTool())Communication via stdin/stdout with JSON-RPC framing:
from mcp_server import StdioTransport
transport = StdioTransport()
await server.run(transport)For web-based integrations:
from mcp_server import WebSocketTransport
# Requires a WebSocket connection (e.g., from websockets library)
transport = WebSocketTransport(websocket)
await server.run(transport)Limit file operations to specific directories:
config = ServerConfig(
allowed_paths=["/home/user/projects", "/tmp"],
)Block dangerous shell commands:
from mcp_server import ShellTool
tool = ShellTool(
blocked_commands=["rm -rf /", "mkfs"],
allowed_commands=["ls", "cat", "grep"], # Whitelist mode
timeout=30,
)Control concurrent requests and timeouts:
config = ServerConfig(
max_concurrent_requests=10,
request_timeout=60.0,
)Implements MCP with JSON-RPC 2.0:
// Request
{
"jsonrpc": "2.0",
"id": "1",
"method": "tools/call",
"params": {
"name": "file_read",
"arguments": {"path": "/etc/hostname"}
}
}
// Response
{
"jsonrpc": "2.0",
"id": "1",
"result": {
"content": [{"type": "text", "text": "myhost\n"}]
}
}| Method | Description |
|---|---|
initialize |
Initialize connection, exchange capabilities |
initialized |
Client notification after init complete |
tools/list |
List available tools |
tools/call |
Execute a tool |
shutdown |
Request server shutdown |
ping |
Health check |
pytest tests/ -v112 tests covering:
- Protocol message parsing and serialization
- Tool execution and validation
- Transport layer (stdio, websocket)
- Server lifecycle and request handling
MIT