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