Meshtastic MCP Server

# MCP Best Practices Implementation This document explains how the Meshtastic MCP server follows Model Context Protocol best practices. ## Naming Conventions ### Server Name ✅ **Rule**: Python servers use `{service}_mcp` format **Implementation**: `meshtastic_mcp` **Why**: Standardized naming makes servers discoverable and predictable for clients. ### Tool Names ✅ **Rule**: Use `{service}_{action}_{resource}` with snake_case **Implementation**: All tools prefixed with `meshtastic_`: - `meshtastic_connect` - `meshtastic_send_message` - `meshtastic_get_nodes` - `meshtastic_scan_bluetooth` **Why**: Prevents naming conflicts when multiple MCP servers are used together. ## FastMCP Framework ✅ **Rule**: Use FastMCP for modern Python MCP servers **Implementation**: ```python from mcp.server.fastmcp import FastMCP mcp = FastMCP("meshtastic_mcp") @mcp.tool(name="meshtastic_connect", annotations={...}) async def meshtastic_connect(params: ConnectInput) -> str: ... ``` **Benefits**: - Automatic schema generation from Pydantic models - Decorator-based tool registration - Built-in type safety - Cleaner, more maintainable code ## Input Validation ✅ **Rule**: Use Pydantic models for all tool inputs **Implementation**: ```python class SendMessageInput(BaseModel): model_config = ConfigDict( str_strip_whitespace=True, validate_assignment=True, extra='forbid' ) text: str = Field( ..., description="Message content", min_length=1, max_length=500 ) ``` **Features**: - Type safety with Python type hints - Automatic validation of constraints - Clear error messages for invalid inputs - Whitespace stripping - Extra field prevention ## Response Formats ✅ **Rule**: Support both JSON and Markdown formats **Implementation**: All read operations accept `response_format` parameter: ```python class GetNodesInput(BaseModel): response_format: ResponseFormat = Field( default=ResponseFormat.MARKDOWN, description="Response format: 'json' or 'markdown'" ) ``` **JSON Format** - Machine-readable: ```json { "total_nodes": 5, "nodes": [...] } ``` **Markdown Format** - Human-readable: ```markdown ## Mesh Network Nodes (5 total) ### Base Station (BASE) - **Node ID**: `!a1b2c3d4` ``` **Why**: Different clients have different needs. JSON for programmatic processing, Markdown for human readability. ## Tool Annotations ✅ **Rule**: Provide behavioral hints to clients **Implementation**: ```python @mcp.tool( name="meshtastic_get_nodes", annotations={ "title": "List All Mesh Network Nodes", "readOnlyHint": True, # No modifications "destructiveHint": False, # No data deletion "idempotentHint": True, # Same result on repeat "openWorldHint": True # External interaction } ) ``` **Annotations Explained**: - `readOnlyHint`: Tool doesn't modify environment (queries only) - `destructiveHint`: Tool may delete or destructively update data - `idempotentHint`: Repeated calls with same args produce same effect - `openWorldHint`: Tool interacts with external entities (devices, networks) ## Error Handling ✅ **Rule**: Provide actionable error messages **Implementation**: ```python except FileNotFoundError as e: raise RuntimeError( f"Device not found: {device}\n\n" f"Troubleshooting:\n" f"- For serial: Check device path (Linux: /dev/ttyUSB0)\n" f"- Ensure device is plugged in\n" f"- Check permissions: sudo usermod -a -G dialout $USER" ) ``` **Features**: - Specific error types caught separately - Clear problem description - Step-by-step troubleshooting - Platform-specific guidance - No internal implementation details exposed ## Logging ✅ **Rule**: stdio servers must log to stderr **Implementation**: ```python logging.basicConfig( level=logging.INFO, stream=sys.stderr, # ← Critical for stdio transport format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) ``` **Why**: In stdio transport, stdout is used for MCP protocol communication. Logging to stdout would corrupt the protocol stream. ## Documentation ✅ **Rule**: Comprehensive docstrings for all tools **Implementation**: ```python async def meshtastic_send_message(params: SendMessageInput) -> str: """Send a text message to a node or broadcast to all nodes. Messages are transmitted over LoRa radio and can be received by any node in range. Keep messages under 237 bytes for best compatibility. Args: params (SendMessageInput): Message parameters containing: - text: Message content to send (1-500 characters) - destination: Node ID or '^all' for broadcast - response_format: Output format ('json' or 'markdown') Returns: str: Confirmation message in requested format Raises: RuntimeError: If not connected or send fails """ ``` **Features**: - Clear description of functionality - Parameter documentation - Return type specification - Exception documentation - Usage context and constraints ## State Management ✅ **Rule**: Clean, maintainable state handling **Implementation**: ```python class ServerState: """Maintains connection and message state.""" def __init__(self): self.interface: Optional[Any] = None self.connection_type: Optional[str] = None self.device_address: Optional[str] = None self.message_history: List[Dict[str, Any]] = [] state = ServerState() ``` **Benefits**: - Centralized state management - Type hints for clarity - Easy to test and maintain - Clear ownership and lifecycle ## Type Safety ✅ **Rule**: Use type hints throughout **Implementation**: ```python from typing import Optional, List, Dict, Any, Literal from enum import Enum class ConnectionType(str, Enum): SERIAL = "serial" TCP = "tcp" BLUETOOTH = "bluetooth" def get_interface() -> Any: """Get the current interface or raise error.""" if state.interface is None: raise RuntimeError("Not connected...") return state.interface ``` **Benefits**: - IDE autocomplete support - Early error detection - Self-documenting code - Better refactoring support ## Security ✅ ### Input Validation - Coordinate bounds checking: `-90 ≤ lat ≤ 90` - String length limits: `1 ≤ len ≤ 500` - Path validation (implicit via device path handling) - Node ID format validation ### Error Message Safety - No internal stack traces exposed - No sensitive data in error messages - Generic errors for security-relevant failures - Detailed logging server-side only ### Connection Security - Supports Meshtastic's built-in encryption - No credentials stored in code - Environment-based configuration recommended ## Testing ✅ ### Provided Test Utilities - `test_connection.py`: Pre-flight connectivity testing - Clear success/failure indicators - Platform-specific troubleshooting ### Manual Testing ```bash # Syntax check python -m py_compile server.py # MCP Inspector npx @modelcontextprotocol/inspector python server.py # Connection tests python test_connection.py --type serial python test_connection.py --type scan ``` ## Architecture Decisions ### Why FastMCP? - Modern, recommended framework - Automatic schema generation - Built-in validation - Less boilerplate code ### Why Pydantic v2? - Industry standard for Python validation - Excellent error messages - MCP SDK integration - Type safety guarantees ### Why Dual Response Formats? - Flexibility for different clients - Human-readable debugging - Machine-processable data - Better AI agent support ### Why State Class? - Encapsulation - Type safety - Easy testing - Clear lifecycle ## Compliance Checklist ### Naming ✅ - [x] Server name: `meshtastic_mcp` - [x] Tools prefixed: `meshtastic_*` - [x] snake_case throughout - [x] Action-oriented names ### Implementation ✅ - [x] FastMCP framework - [x] Pydantic models for inputs - [x] Async/await patterns - [x] Proper error handling - [x] Type hints throughout ### Features ✅ - [x] JSON and Markdown formats - [x] Tool annotations - [x] Input validation - [x] Actionable errors - [x] Comprehensive docs ### Quality ✅ - [x] No code duplication - [x] Consistent patterns - [x] Clear function/class names - [x] Proper logging - [x] Security considerations ## Performance Considerations ### Memory - Message history stored in memory - Grows unbounded (could add rotation) - Single connection per server instance ### I/O - All operations async/await - Non-blocking network calls - LoRa transmission inherently slow ### Scalability - Single device per server instance - Stateful connection management - Consider separate server per device for multi-device ## Future Improvements ### High Priority - [ ] Message history persistence (SQLite) - [ ] Connection retry logic - [ ] Telemetry monitoring tools - [ ] Admin message support ### Medium Priority - [ ] Multi-device support - [ ] Message filtering/search - [ ] Automated testing suite - [ ] Performance benchmarks ### Low Priority - [ ] Waypoint management - [ ] File transfer support - [ ] Custom emoji support - [ ] Voice message support ## References - [MCP Specification](https://modelcontextprotocol.io/specification) - [MCP Best Practices](https://modelcontextprotocol.io/docs/best-practices) - [FastMCP Documentation](https://github.com/modelcontextprotocol/python-sdk) - [Pydantic Documentation](https://docs.pydantic.dev/) - [Meshtastic Python API](https://meshtastic.org/docs/software/python/cli/)