mcesptool/MCP_LOGGER_INTEGRATION.md

🔗 ESPTool Custom Logger → MCP Integration

Overview

Bridge esptool's logging system with FastMCP's advanced logging, progress tracking, and user elicitation capabilities. This creates a seamless user experience where ESP operations provide real-time feedback and can request user input when needed.

🎯 Logger Mapping Strategy

ESPTool → MCP Method Mapping

# Direct logging method mappings
ESPTOOL_TO_MCP_MAPPING = {
    # Basic logging
    "print": "log.info",           # General information
    "note": "log.notice",          # Important notices
    "warning": "log.warning",      # Warning messages
    "error": "log.error",          # Error messages

    # Advanced MCP features
    "stage": "elicit_if_available",     # User interaction/confirmation
    "progress_bar": "progress",         # Progress tracking
    "set_verbosity": "configure_verbosity"  # Dynamic verbosity control
}

🏗️ Custom Logger Implementation

FastMCP-Integrated ESPTool Logger

# components/mcp_esptool_logger.py
from esptool.logger import TemplateLogger
from fastmcp import Context
from typing import Optional, Any
import asyncio
import sys

class MCPESPToolLogger(TemplateLogger):
    """Custom esptool logger that integrates with FastMCP capabilities"""

    def __init__(self, context: Context, operation_id: str):
        """
        Initialize MCP-integrated logger

        Args:
            context: FastMCP context for logging and elicitation
            operation_id: Unique identifier for this operation
        """
        self.context = context
        self.operation_id = operation_id
        self.verbosity_level = 1
        self.current_stage = None
        self.progress_total = 0
        self.progress_current = 0

        # Check MCP client capabilities
        self.supports_progress = self._check_progress_support()
        self.supports_elicitation = self._check_elicitation_support()

    def _check_progress_support(self) -> bool:
        """Check if client supports progress notifications"""
        try:
            return hasattr(self.context, 'progress') and callable(self.context.progress)
        except:
            return False

    def _check_elicitation_support(self) -> bool:
        """Check if client supports user elicitation"""
        try:
            return hasattr(self.context, 'request_user_input') and callable(self.context.request_user_input)
        except:
            return False

    def print(self, message: str = "", *args, **kwargs) -> None:
        """Map esptool print() to MCP info logging"""
        if self.verbosity_level >= 1:
            formatted_message = self._format_message(message, *args)

            # Use MCP context for structured logging
            asyncio.create_task(self._log_async("info", formatted_message))

    def note(self, message: str) -> None:
        """Map esptool note() to MCP notice logging"""
        formatted_message = f"📋 NOTE: {message}"
        asyncio.create_task(self._log_async("notice", formatted_message))

    def warning(self, message: str) -> None:
        """Map esptool warning() to MCP warning logging"""
        formatted_message = f"⚠️ WARNING: {message}"
        asyncio.create_task(self._log_async("warning", formatted_message))

    def error(self, message: str) -> None:
        """Map esptool error() to MCP error logging"""
        formatted_message = f"❌ ERROR: {message}"
        asyncio.create_task(self._log_async("error", formatted_message))

    def stage(self, message: str = "", finish: bool = False) -> None:
        """
        Map esptool stage() to MCP elicitation for user interaction

        This is where the magic happens - esptool "stages" become opportunities
        for user interaction and confirmation
        """
        if finish and self.current_stage:
            # End current stage
            asyncio.create_task(self._finish_stage())
            self.current_stage = None
        elif message and not finish:
            # Start new stage - potentially with user interaction
            self.current_stage = message
            asyncio.create_task(self._start_stage(message))

    def progress_bar(
        self,
        cur_iter: int,
        total_iters: int,
        prefix: str = "",
        suffix: str = "",
        bar_length: int = 30,
    ) -> None:
        """Map esptool progress_bar() to MCP progress notifications"""
        self.progress_current = cur_iter
        self.progress_total = total_iters

        if self.supports_progress:
            percentage = (cur_iter / total_iters) * 100 if total_iters > 0 else 0

            # Send MCP progress notification
            asyncio.create_task(self._update_progress(
                operation_id=self.operation_id,
                progress=percentage,
                total=total_iters,
                current=cur_iter,
                message=f"{prefix} {suffix}".strip()
            ))
        else:
            # Fallback to text-based progress
            percentage = f"{100 * (cur_iter / float(total_iters)):.1f}%"
            self.print(f"Progress: {percentage} {prefix} {suffix}")

    def set_verbosity(self, verbosity: int) -> None:
        """Dynamic verbosity control"""
        self.verbosity_level = verbosity
        asyncio.create_task(self._log_async("info", f"Verbosity set to level {verbosity}"))

    # MCP-specific async methods
    async def _log_async(self, level: str, message: str) -> None:
        """Async logging to MCP context"""
        try:
            if hasattr(self.context, 'log'):
                await self.context.log(level=level, message=message)
            else:
                # Fallback to console
                print(f"[{level.upper()}] {message}")
        except Exception as e:
            print(f"Logging error: {e}")

    async def _start_stage(self, stage_message: str) -> None:
        """Start interactive stage with potential user elicitation"""
        await self._log_async("info", f"🔄 Starting: {stage_message}")

        # Check if this stage might need user input
        if self._requires_user_interaction(stage_message):
            await self._elicit_user_confirmation(stage_message)

    async def _finish_stage(self) -> None:
        """Finish current stage"""
        if self.current_stage:
            await self._log_async("info", f"✅ Completed: {self.current_stage}")

    async def _update_progress(
        self,
        operation_id: str,
        progress: float,
        total: int,
        current: int,
        message: str
    ) -> None:
        """Send progress update via MCP"""
        try:
            if self.supports_progress:
                await self.context.progress(
                    operation_id=operation_id,
                    progress=progress,
                    total=total,
                    current=current,
                    message=message
                )
        except Exception as e:
            await self._log_async("warning", f"Progress update failed: {e}")

    async def _elicit_user_confirmation(self, stage_message: str) -> bool:
        """Elicit user confirmation for critical operations"""
        if not self.supports_elicitation:
            return True  # Proceed if no elicitation support

        try:
            # Determine confirmation message based on stage
            confirmation_message = self._generate_confirmation_message(stage_message)

            response = await self.context.request_user_input(
                prompt=confirmation_message,
                input_type="confirmation"
            )

            return response.get("confirmed", True)
        except Exception as e:
            await self._log_async("warning", f"User elicitation failed: {e}")
            return True  # Default to proceed

    def _requires_user_interaction(self, stage_message: str) -> bool:
        """Determine if stage requires user confirmation"""
        critical_operations = [
            "erasing flash",
            "burning efuses",
            "enabling secure boot",
            "enabling flash encryption",
            "factory reset"
        ]

        message_lower = stage_message.lower()
        return any(op in message_lower for op in critical_operations)

    def _generate_confirmation_message(self, stage_message: str) -> str:
        """Generate appropriate confirmation message"""
        confirmations = {
            "erasing flash": "⚠️ This will erase all data on the ESP flash memory. Continue?",
            "burning efuses": "🔥 eFuse burning is PERMANENT and cannot be undone. Continue?",
            "enabling secure boot": "🔐 Secure boot will permanently change chip configuration. Continue?",
            "enabling flash encryption": "🔒 Flash encryption is permanent and cannot be disabled. Continue?",
            "factory reset": "🏭 This will restore factory settings, erasing all user data. Continue?"
        }

        message_lower = stage_message.lower()
        for key, confirmation in confirmations.items():
            if key in message_lower:
                return confirmation

        return f"🤔 About to: {stage_message}. Continue?"

    def _format_message(self, message: str, *args) -> str:
        """Format message with optional arguments"""
        if args:
            try:
                return message % args
            except (TypeError, ValueError):
                return f"{message} {' '.join(map(str, args))}"
        return message

🔧 Integration with ESPTool Operations

Logger Factory and Context Management

# components/logger_factory.py
from contextlib import contextmanager
from esptool.logger import log

class MCPLoggerFactory:
    """Factory for creating MCP-integrated esptool loggers"""

    @staticmethod
    @contextmanager
    def create_mcp_logger(context: Context, operation_name: str):
        """
        Context manager for MCP logger integration

        Usage:
            with MCPLoggerFactory.create_mcp_logger(ctx, "flash_operation") as logger:
                # esptool operations here will use MCP logging
                pass
        """
        operation_id = f"{operation_name}_{int(time.time())}"
        mcp_logger = MCPESPToolLogger(context, operation_id)

        # Replace esptool's default logger
        original_logger = log.get_logger()
        log.set_logger(mcp_logger)

        try:
            yield mcp_logger
        finally:
            # Restore original logger
            log.set_logger(original_logger)

Enhanced Tool Implementation

# Example: Enhanced esp_flash_firmware with MCP logging
@app.tool("esp_flash_firmware_interactive")
async def flash_firmware_with_mcp_logging(
    context: Context,
    port: str,
    firmware_files: List[Dict],
    interactive: bool = True
) -> str:
    """Flash firmware with full MCP integration and user interaction"""

    with MCPLoggerFactory.create_mcp_logger(context, "flash_firmware") as mcp_logger:
        try:
            # esptool operations now use MCP logging automatically
            with detect_chip(port) as esp:
                # Logger will elicit user confirmation for critical operations
                esp = run_stub(esp)
                attach_flash(esp)

                # Progress updates go to MCP progress system
                write_flash(esp, [(item['address'], item['file']) for item in firmware_files])

                # User confirmation for reset
                if interactive:
                    await mcp_logger._elicit_user_confirmation("reset chip after flashing")

                reset_chip(esp, 'hard-reset')

                return "✅ Firmware flashed successfully with user interaction"

        except Exception as e:
            await mcp_logger._log_async("error", f"Flash operation failed: {e}")
            raise

🎨 Natural Language Workflow Enhancement

Interactive ESP Development Experience

User: "Flash my ESP32 with the new firmware, but ask me before doing anything destructive"

Claude: I'll flash your ESP32 with interactive confirmations for safety.

        🔄 Starting: Detecting ESP32 chip...
        ✅ Detected: ESP32-D0WD (revision 1)

        🔄 Starting: Loading flasher stub...
        ✅ Flasher stub loaded for faster operation

        🔄 Starting: Erasing flash memory...
        ⚠️ This will erase existing firmware on the ESP flash memory. Continue? [Y/n]

User: y

Claude: 📋 Proceeding with flash erase...
        Progress: ████████████████████████████████ 100%
        ✅ Flash erased successfully

        🔄 Starting: Writing firmware...
        Progress: ████████████████████████████████ 100%
        ✅ Firmware written and verified

        🤔 About to: reset chip after flashing. Continue? [Y/n]

User: y

Claude: ✅ ESP32 flashed successfully and reset
        Your device is ready to run the new firmware!

📊 Advanced MCP Features Integration

Progress Tracking for Long Operations

# Enhanced progress tracking
class ProgressAwareESPOperation:
    """Wrapper for ESP operations with detailed progress tracking"""

    async def flash_large_firmware(
        self,
        context: Context,
        port: str,
        firmware_path: str
    ) -> None:
        """Flash large firmware with detailed progress"""

        with MCPLoggerFactory.create_mcp_logger(context, "large_firmware_flash") as logger:
            # Multi-stage progress tracking
            stages = [
                ("Connecting to chip", 10),
                ("Loading flasher stub", 20),
                ("Erasing flash", 40),
                ("Writing firmware", 80),
                ("Verifying flash", 95),
                ("Resetting chip", 100)
            ]

            for stage_name, progress_percent in stages:
                await logger._update_progress(
                    operation_id=logger.operation_id,
                    progress=progress_percent,
                    total=100,
                    current=progress_percent,
                    message=stage_name
                )

                # Actual esptool operation here
                await self._execute_stage(stage_name, port, firmware_path)

Elicitation for Configuration Decisions

# Smart configuration elicitation
async def configure_esp_security_interactive(
    context: Context,
    port: str,
    security_level: str = "auto"
) -> Dict:
    """Configure ESP security with intelligent user prompts"""

    if security_level == "auto":
        # Use elicitation to determine security configuration
        security_config = await context.request_user_input(
            prompt="🔐 Security Configuration",
            input_type="selection",
            options=[
                {"id": "development", "label": "Development (no security)"},
                {"id": "testing", "label": "Testing (basic security)"},
                {"id": "production", "label": "Production (full security)"}
            ]
        )

        security_level = security_config.get("selected", "development")

    # Configure based on user choice
    return await apply_security_configuration(context, port, security_level)

This MCP logger integration transforms esptool from a silent command-line tool into an interactive, user-friendly system that provides real-time feedback and intelligently requests user input when needed.

★ Insight ───────────────────────────────────── Seamless User Experience: By mapping esptool's logging methods directly to MCP capabilities, we create a unified experience where ESP operations feel native to the Claude environment while preserving all of esptool's functionality.

Smart Interaction Points: The logger intelligently identifies critical operations (eFuse burning, secure boot) that require user confirmation, transforming potentially destructive operations into safe, interactive workflows.

Progressive Enhancement: The system gracefully degrades when MCP features aren't available, ensuring compatibility across different client implementations while providing enhanced experiences when possible. ─────────────────────────────────────────────────

<function_calls> [{"content": "Create FastMCP esptool server architecture documentation", "status": "completed", "activeForm": "Creating FastMCP esptool server architecture documentation"}, {"content": "Design esptool MCP server API structure", "status": "completed", "activeForm": "Designing esptool MCP server API structure"}, {"content": "Document integration patterns with existing Arduino MCP server", "status": "completed", "activeForm": "Documenting integration patterns with existing Arduino MCP server"}, {"content": "Create implementation roadmap and examples", "status": "completed", "activeForm": "Creating implementation roadmap and examples"}, {"content": "Design esptool custom logger for MCP integration", "status": "completed", "activeForm": "Designing esptool custom logger for MCP integration"}]