mcesptool/MIDDLEWARE_ARCHITECTURE.md

🔗 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:

# 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:

# 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

# 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

# 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

# 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

# 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. ─────────────────────────────────────────────────