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
444 lines
17 KiB
Markdown
444 lines
17 KiB
Markdown
# 🔗 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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>
|
|
<invoke name="TodoWrite">
|
|
<parameter name="todos">[{"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"}] |