Ryan Malloy
64c1505a00
Integrate Espressif's QEMU fork for virtual ESP device management: - QemuManager component with 5 MCP tools (start/stop/list/status/flash) - Config auto-detects QEMU binaries from ~/.espressif/tools/ - Supports esp32, esp32s2, esp32s3, esp32c3 chip emulation - Virtual serial over TCP (socket://localhost:PORT) transparent to esptool - Scan integration: QEMU instances appear in esp_scan_ports results - Blank flash images initialized to 0xFF (erased NOR flash state) - 38 unit tests covering lifecycle, port allocation, flash writes
410 lines
17 KiB
Markdown
410 lines
17 KiB
Markdown
# 🔗 MCP Middleware Architecture Pattern
|
|
|
|
## Overview
|
|
|
|
The MCP Logger Middleware represents a novel architectural pattern for integrating existing command-line tools with Model Context Protocol servers. This middleware acts as a bidirectional translation layer that transforms traditional CLI tools into AI-native, interactive systems without modifying the original tool's codebase.
|
|
|
|
## 🏗️ Architectural Principles
|
|
|
|
### Core Middleware Concept
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ MCP Middleware Layer │
|
|
├─────────────────────────────────────────────────────────────────┤
|
|
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
|
|
│ │ Legacy │───▶│ Translation │───▶│ MCP │ │
|
|
│ │ Tool │ │ Engine │ │ Context │ │
|
|
│ │ API │◀───│ │◀───│ API │ │
|
|
│ └─────────────┘ └─────────────┘ └─────────────┘ │
|
|
│ │ │
|
|
│ ▼ │
|
|
│ ┌─────────────────┐ │
|
|
│ │ Enhancement │ │
|
|
│ │ Services │ │
|
|
│ │ │ │
|
|
│ │ • Progress │ │
|
|
│ │ • Elicitation │ │
|
|
│ │ • Context │ │
|
|
│ │ • Validation │ │
|
|
│ └─────────────────┘ │
|
|
└─────────────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
### Design Philosophy
|
|
|
|
1. **Non-Invasive Integration**: Zero modifications to existing tools
|
|
2. **Protocol Translation**: Seamless sync/async bridging
|
|
3. **Progressive Enhancement**: Graceful degradation across client capabilities
|
|
4. **Bidirectional Communication**: Interactive workflows with user feedback
|
|
5. **Context Awareness**: Rich integration with MCP ecosystem
|
|
|
|
## 🔧 Middleware Implementation Patterns
|
|
|
|
### 1. **Logger Interception Pattern**
|
|
|
|
The most elegant approach for tools with pluggable logging systems:
|
|
|
|
```python
|
|
# middleware/logger_interceptor.py
|
|
from abc import ABC, abstractmethod
|
|
from typing import Protocol, Any, Dict, Optional
|
|
from fastmcp import Context
|
|
|
|
class LoggerInterceptor(ABC):
|
|
"""Abstract base for logger interception middleware"""
|
|
|
|
def __init__(self, context: Context, operation_id: str):
|
|
self.context = context
|
|
self.operation_id = operation_id
|
|
self.capabilities = self._detect_capabilities()
|
|
|
|
@abstractmethod
|
|
def intercept_logging_calls(self) -> None:
|
|
"""Intercept and redirect tool's logging calls"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def translate_to_mcp(self, call_type: str, *args, **kwargs) -> None:
|
|
"""Translate tool calls to MCP context methods"""
|
|
pass
|
|
|
|
def _detect_capabilities(self) -> Dict[str, bool]:
|
|
"""Detect available MCP client capabilities"""
|
|
return {
|
|
"progress": hasattr(self.context, 'progress'),
|
|
"elicitation": hasattr(self.context, 'request_user_input'),
|
|
"logging": hasattr(self.context, 'log'),
|
|
"sampling": hasattr(self.context, 'sample')
|
|
}
|
|
```
|
|
|
|
### 2. **Context Manager Pattern**
|
|
|
|
Ensures proper resource management and cleanup:
|
|
|
|
```python
|
|
# middleware/context_manager.py
|
|
from contextlib import contextmanager
|
|
from typing import Generator, TypeVar, Generic
|
|
|
|
T = TypeVar('T')
|
|
|
|
class MiddlewareContextManager(Generic[T]):
|
|
"""Context manager for middleware lifecycle"""
|
|
|
|
def __init__(self, tool_instance: T, middleware_class):
|
|
self.tool_instance = tool_instance
|
|
self.middleware_class = middleware_class
|
|
self.original_state = None
|
|
|
|
@contextmanager
|
|
def activate(self, context: Context, operation_id: str) -> Generator[T, None, None]:
|
|
"""Activate middleware for the duration of operations"""
|
|
|
|
# Store original state
|
|
self.original_state = self._capture_original_state()
|
|
|
|
# Create and inject middleware
|
|
middleware = self.middleware_class(context, operation_id)
|
|
self._inject_middleware(middleware)
|
|
|
|
try:
|
|
yield self.tool_instance
|
|
finally:
|
|
# Restore original state
|
|
self._restore_original_state()
|
|
|
|
def _capture_original_state(self) -> Dict[str, Any]:
|
|
"""Capture tool's original configuration"""
|
|
# Implementation specific to each tool
|
|
pass
|
|
|
|
def _inject_middleware(self, middleware) -> None:
|
|
"""Inject middleware into tool's execution path"""
|
|
# Implementation specific to each tool
|
|
pass
|
|
|
|
def _restore_original_state(self) -> None:
|
|
"""Restore tool to original state"""
|
|
# Implementation specific to each tool
|
|
pass
|
|
```
|
|
|
|
### 3. **Factory Pattern for Tool-Specific Middleware**
|
|
|
|
```python
|
|
# middleware/factory.py
|
|
from typing import Type, Dict, Any
|
|
from .esptool_middleware import ESPToolMiddleware
|
|
from .platformio_middleware import PlatformIOMiddleware
|
|
from .idf_middleware import IDFMiddleware
|
|
|
|
class MiddlewareFactory:
|
|
"""Factory for creating tool-specific middleware instances"""
|
|
|
|
_middleware_registry: Dict[str, Type] = {
|
|
'esptool': ESPToolMiddleware,
|
|
'platformio': PlatformIOMiddleware,
|
|
'esp-idf': IDFMiddleware,
|
|
# Add more tools as needed
|
|
}
|
|
|
|
@classmethod
|
|
def create_middleware(
|
|
cls,
|
|
tool_name: str,
|
|
context: Context,
|
|
operation_id: str,
|
|
**kwargs
|
|
) -> LoggerInterceptor:
|
|
"""Create appropriate middleware for the specified tool"""
|
|
|
|
if tool_name not in cls._middleware_registry:
|
|
raise ValueError(f"No middleware available for tool: {tool_name}")
|
|
|
|
middleware_class = cls._middleware_registry[tool_name]
|
|
return middleware_class(context, operation_id, **kwargs)
|
|
|
|
@classmethod
|
|
def register_middleware(cls, tool_name: str, middleware_class: Type) -> None:
|
|
"""Register new middleware for a tool"""
|
|
cls._middleware_registry[tool_name] = middleware_class
|
|
|
|
@classmethod
|
|
def list_supported_tools(cls) -> List[str]:
|
|
"""List all tools with available middleware"""
|
|
return list(cls._middleware_registry.keys())
|
|
```
|
|
|
|
## 🎯 ESPTool Implementation Case Study
|
|
|
|
### Complete ESPTool Middleware Implementation
|
|
|
|
```python
|
|
# middleware/esptool_middleware.py
|
|
from esptool.logger import log, TemplateLogger
|
|
from .logger_interceptor import LoggerInterceptor
|
|
import asyncio
|
|
from typing import Dict, Any, Optional
|
|
|
|
class ESPToolMiddleware(LoggerInterceptor):
|
|
"""ESPTool-specific middleware implementation"""
|
|
|
|
def __init__(self, context: Context, operation_id: str):
|
|
super().__init__(context, operation_id)
|
|
self.original_logger = None
|
|
self.active_stages = []
|
|
|
|
def intercept_logging_calls(self) -> None:
|
|
"""Replace esptool's logger with MCP-aware version"""
|
|
self.original_logger = log.get_logger()
|
|
mcp_logger = MCPESPToolLogger(self.context, self.operation_id)
|
|
log.set_logger(mcp_logger)
|
|
|
|
def restore_original_logger(self) -> None:
|
|
"""Restore esptool's original logger"""
|
|
if self.original_logger:
|
|
log.set_logger(self.original_logger)
|
|
|
|
async def translate_to_mcp(self, call_type: str, *args, **kwargs) -> None:
|
|
"""Translate esptool calls to MCP context methods"""
|
|
|
|
translation_map = {
|
|
'print': self._handle_print,
|
|
'note': self._handle_note,
|
|
'warning': self._handle_warning,
|
|
'error': self._handle_error,
|
|
'stage': self._handle_stage,
|
|
'progress_bar': self._handle_progress,
|
|
'set_verbosity': self._handle_verbosity
|
|
}
|
|
|
|
handler = translation_map.get(call_type)
|
|
if handler:
|
|
await handler(*args, **kwargs)
|
|
|
|
async def _handle_print(self, message: str, *args) -> None:
|
|
"""Handle general print messages"""
|
|
if self.capabilities['logging']:
|
|
await self.context.log(level='info', message=self._format_message(message, *args))
|
|
|
|
async def _handle_note(self, message: str) -> None:
|
|
"""Handle note messages with special formatting"""
|
|
formatted = f"📋 {message}"
|
|
if self.capabilities['logging']:
|
|
await self.context.log(level='notice', message=formatted)
|
|
|
|
async def _handle_warning(self, message: str) -> None:
|
|
"""Handle warning messages"""
|
|
formatted = f"⚠️ {message}"
|
|
if self.capabilities['logging']:
|
|
await self.context.log(level='warning', message=formatted)
|
|
|
|
async def _handle_error(self, message: str) -> None:
|
|
"""Handle error messages"""
|
|
formatted = f"❌ {message}"
|
|
if self.capabilities['logging']:
|
|
await self.context.log(level='error', message=formatted)
|
|
|
|
async def _handle_stage(self, message: str = "", finish: bool = False) -> None:
|
|
"""Handle stage transitions with potential user interaction"""
|
|
if finish and self.active_stages:
|
|
stage = self.active_stages.pop()
|
|
await self._finish_stage(stage)
|
|
elif message:
|
|
self.active_stages.append(message)
|
|
await self._start_stage(message)
|
|
|
|
async def _handle_progress(self, cur_iter: int, total_iters: int, **kwargs) -> None:
|
|
"""Handle progress updates"""
|
|
if self.capabilities['progress']:
|
|
percentage = (cur_iter / total_iters) * 100 if total_iters > 0 else 0
|
|
await self.context.progress(
|
|
operation_id=self.operation_id,
|
|
progress=percentage,
|
|
total=total_iters,
|
|
current=cur_iter,
|
|
message=kwargs.get('prefix', '') + ' ' + kwargs.get('suffix', '')
|
|
)
|
|
|
|
async def _start_stage(self, stage_message: str) -> None:
|
|
"""Start new stage with potential user interaction"""
|
|
await self.context.log(level='info', message=f"🔄 Starting: {stage_message}")
|
|
|
|
# Check if stage requires user confirmation
|
|
if self._requires_confirmation(stage_message) and self.capabilities['elicitation']:
|
|
await self._request_user_confirmation(stage_message)
|
|
|
|
async def _finish_stage(self, stage_message: str) -> None:
|
|
"""Finish stage with completion notification"""
|
|
await self.context.log(level='info', message=f"✅ Completed: {stage_message}")
|
|
|
|
def _requires_confirmation(self, stage_message: str) -> bool:
|
|
"""Determine if stage requires user confirmation"""
|
|
critical_keywords = ['erase', 'burn', 'encrypt', 'secure', 'factory']
|
|
return any(keyword in stage_message.lower() for keyword in critical_keywords)
|
|
|
|
async def _request_user_confirmation(self, stage_message: str) -> bool:
|
|
"""Request user confirmation for critical operations"""
|
|
try:
|
|
response = await self.context.request_user_input(
|
|
prompt=f"🤔 About to: {stage_message}. Continue?",
|
|
input_type="confirmation"
|
|
)
|
|
return response.get('confirmed', True)
|
|
except Exception:
|
|
return True # Default to proceed if elicitation fails
|
|
|
|
def _format_message(self, message: str, *args) -> str:
|
|
"""Format message with arguments"""
|
|
try:
|
|
return message % args if args else message
|
|
except (TypeError, ValueError):
|
|
return f"{message} {' '.join(map(str, args))}" if args else message
|
|
```
|
|
|
|
## 🔄 Middleware Usage Patterns
|
|
|
|
### Simple Tool Wrapping
|
|
|
|
```python
|
|
# Usage example with context manager
|
|
@app.tool("esp_flash_with_middleware")
|
|
async def flash_with_middleware(
|
|
context: Context,
|
|
port: str,
|
|
firmware_path: str
|
|
) -> str:
|
|
"""Flash ESP32 with full middleware integration"""
|
|
|
|
middleware = MiddlewareFactory.create_middleware('esptool', context, 'flash_operation')
|
|
|
|
with MiddlewareContextManager(middleware).activate():
|
|
# All esptool operations now use MCP integration
|
|
with detect_chip(port) as esp:
|
|
esp = run_stub(esp)
|
|
attach_flash(esp)
|
|
write_flash(esp, [(0x1000, firmware_path)])
|
|
reset_chip(esp, 'hard-reset')
|
|
|
|
return "✅ Flashing completed with user interaction"
|
|
```
|
|
|
|
### Advanced Middleware Configuration
|
|
|
|
```python
|
|
# Advanced middleware with custom configuration
|
|
@app.tool("esp_configure_advanced_middleware")
|
|
async def configure_advanced_middleware(
|
|
context: Context,
|
|
operation_config: Dict[str, Any]
|
|
) -> str:
|
|
"""Configure middleware with advanced options"""
|
|
|
|
middleware_config = {
|
|
'enable_progress': operation_config.get('show_progress', True),
|
|
'require_confirmations': operation_config.get('interactive', True),
|
|
'verbosity_level': operation_config.get('verbosity', 1),
|
|
'custom_translations': operation_config.get('message_translations', {})
|
|
}
|
|
|
|
middleware = MiddlewareFactory.create_middleware(
|
|
'esptool',
|
|
context,
|
|
'advanced_operation',
|
|
**middleware_config
|
|
)
|
|
|
|
# Use configured middleware for operations
|
|
return await execute_with_middleware(middleware, operation_config)
|
|
```
|
|
|
|
## 🌟 Benefits and Advantages
|
|
|
|
### For Tool Integration
|
|
|
|
1. **Zero Code Changes**: Original tools remain completely unmodified
|
|
2. **Preservation of Functionality**: All original features remain available
|
|
3. **Enhanced User Experience**: Interactive workflows with progress tracking
|
|
4. **Error Recovery**: Better error handling and user guidance
|
|
5. **Context Awareness**: Operations become aware of broader development context
|
|
|
|
### For MCP Server Development
|
|
|
|
1. **Rapid Integration**: Quick integration of existing CLI tools
|
|
2. **Consistent Patterns**: Reusable middleware architecture
|
|
3. **Extensibility**: Easy to add new tools and capabilities
|
|
4. **Maintainability**: Clear separation of concerns
|
|
5. **Testing**: Isolated middleware can be unit tested independently
|
|
|
|
### For End Users
|
|
|
|
1. **Natural Language Interface**: CLI tools become conversational
|
|
2. **Safety Features**: Interactive confirmations for destructive operations
|
|
3. **Progress Visibility**: Real-time feedback on long operations
|
|
4. **Error Guidance**: Helpful error messages and recovery suggestions
|
|
5. **Context Integration**: Tools work seamlessly with AI assistants
|
|
|
|
## 🚀 Broader Applications
|
|
|
|
This middleware pattern extends far beyond esptool to any CLI tool with pluggable interfaces:
|
|
|
|
- **PlatformIO**: Embedded development framework
|
|
- **ESP-IDF**: Espressif's official development framework
|
|
- **Arduino CLI**: Arduino command-line interface
|
|
- **OpenOCD**: On-chip debugging tool
|
|
- **GDB**: GNU Debugger
|
|
- **Make/CMake**: Build systems
|
|
- **Git**: Version control operations
|
|
- **Docker**: Container operations
|
|
|
|
The pattern creates a pathway for transforming the entire embedded development toolchain into AI-native, interactive systems while preserving their original capabilities and maintaining backward compatibility.
|
|
|
|
`★ Insight ─────────────────────────────────────`
|
|
**Universal Integration Pattern**: This middleware architecture represents a universal solution for modernizing CLI tools with AI interfaces. It demonstrates how existing software ecosystems can be enhanced without disruption.
|
|
|
|
**Bidirectional Translation**: The middleware doesn't just capture output - it enables bidirectional communication, allowing AI systems to interact with tools in real-time, creating truly collaborative development experiences.
|
|
|
|
**Emergent Intelligence**: By providing tools with context awareness and user interaction capabilities, the middleware enables emergent intelligent behaviors that weren't possible with traditional CLI interfaces.
|
|
`─────────────────────────────────────────────────` |