# mcp-esp32

An MCP server that puts an ESP32 / MicroPython board behind tools an agent
can call: enumerate ports, flash firmware, run code over the raw REPL, move
files, capture serial output.

The problem this repository is actually about: flashing a board takes tens
of seconds, boards reset unexpectedly, the serial link drops bytes, and an
agent needs to be able to cancel a flash it started. None of that fits the
request/response shape most MCP servers use. This repo is a demonstration of
the parts of MCP built for exactly that -- progress notifications,
cancellation, and structured (not thrown-and-forgotten) error handling --
applied to a device that is genuinely slow, stateful, and failure-prone.

Result: eight tools, a from-scratch raw-REPL client, a flashing orchestrator
that reports progress at least every 5% and can be cancelled mid-write with
no orphaned process, and a from-scratch fault-injecting simulator so all of
that is verifiable without a board attached. 37 tests, all green, no
hardware required. See [Verification status](#verification-status) below
for exactly what that does and does not prove.

## Why a long-running hardware operation is a harder MCP shape than a REST wrapper

A REST wrapper around `esptool` is `POST /flash` and a 200 once it's done (or
a client-side timeout if it isn't). That throws away everything that
actually matters when the thing on the other end of a serial cable takes 30
seconds to flash and can silently vanish partway through:

- **Progress.** A flash with no feedback for 30 seconds is indistinguishable
  from a hang, to both a human and an agent deciding whether to keep waiting.
  `flash_firmware` reports progress via MCP's `notifications/progress` at
  least every 5%, not just at the end.
- **Cancellation.** An agent that started a flash against the wrong port
  needs to be able to stop it -- not just stop *waiting* for it, but actually
  stop the write and not leave a subprocess or a half-open serial port
  behind. `flash_firmware` handles `notifications/cancelled` by signalling
  the in-flight write to stop, waiting (inside a shielded scope, so the
  cleanup itself can't be cancelled) for it to actually stop, and only then
  letting the cancellation propagate.
- **Timeouts vs. failure.** A REST call that times out tells you nothing
  about whether the device is busy, dead, or reset. Every failure mode here
  is a distinct, structured exception (`DeviceResetError`,
  `FlashTimeoutError`, `ReplDesyncError`, ...) with a `kind` field an agent
  can branch on, not a generic timeout.
- **Recovery.** A dropped byte on a REPL exchange shouldn't corrupt the next
  one. `repl_exec` detects a desynchronised raw REPL and resynchronises
  before the next call, instead of returning garbage or hanging.
- **Statefulness under concurrency.** A board has one UART. Two tool calls
  racing against the same port would interleave bytes and corrupt both, so
  every tool call is serialised per-port (an `asyncio.Lock` per port name) --
  calls against different ports still run fully in parallel.

## Tools

| Tool | Description |
|---|---|
| `list_ports()` | Serial ports with VID/PID and a best-guess chip family, plus the simulator. |
| `board_info(port)` | Chip, flash size, MAC, MicroPython version if present. |
| `flash_firmware(port, firmware_path, erase=False, confirm_erase=False)` | Progress notifications every ≥5%, cancellable. `erase=True` requires `confirm_erase=True` as a separate argument. |
| `repl_exec(port, code, timeout_s=10)` | Raw-REPL execution; stdout, stderr, and exception come back as separate fields. |
| `fs_ls(port, path)` | Directory listing (name, size, is_dir). |
| `fs_get(port, path, max_bytes)` | Read a file (base64), capped and refused up front if it exceeds `max_bytes`. |
| `fs_put(port, path, content_base64, max_bytes)` | Write a file, chunked over several raw-REPL exchanges. |
| `tail_serial(port, seconds)` | Bounded (≤30s) raw serial capture, no REPL protocol involved. |

## Architecture

```
server.py        MCP tool registration; adapts Context.report_progress and
                  cancellation for flash_firmware. Everything else is a
                  direct pass-through to toolkit.py.
toolkit.py        Backend-agnostic async tool implementations. No mcp.Context
                  here -- every reliability property is tested by calling
                  these functions directly.
backends/base.py  Resolves a port string to a BoardHandle: transport
                  factory, flash-session factory, identify(). Ports named
                  "SIM*" route to the simulator; anything else routes to
                  the serial backend.
raw_repl.py       MicroPython raw-REPL client (transport-agnostic).
fsops.py          ls/get/put built on raw_repl.exec(), chunked, size-capped.
flasher.py        Flash orchestration: progress throttling, cancellation,
                  SimulatorFlashSession (talks to the simulator) and
                  SubprocessFlashSession (drives the real esptool CLI).
sim_flash_protocol.py
                  Wire format for the simulator's flash handshake.
backends/simulator.py
                  A pty-backed fake device: real raw-REPL protocol, real
                  code execution against an in-memory filesystem, and a
                  configurable fault injector.
backends/serial_backend.py
                  Real hardware via pyserial + the esptool CLI. Not
                  exercised by anything in this repository -- see below.
```

## The simulator, and exactly what it reproduces

The simulator (`backends/simulator.py`) is not a mock of the raw-REPL
client -- it's a second, independent implementation of the *device* side of
that protocol, running in a background thread on the other end of a real
pty. Code sent to it is genuinely executed (via a sandboxed `exec`, with
fake `os`/`machine`/`sys`/`ubinascii` modules backing an in-memory
filesystem), so `raw_repl.py` cannot tell it apart from a real board at the
protocol level.

Its flash handshake is a separate, deliberately simple framed protocol
(`sim_flash_protocol.py`) -- sync, begin, N acknowledged data blocks, end --
not a reimplementation of esptool's real SLIP/ROM-bootloader wire format.
Reproducing that byte-for-byte (chip-specific stub loaders, ROM quirks, SLIP
escaping) is a separate project from what this repository demonstrates. What
matters for exercising the MCP long-running-operation surface is the shape
of a flash: a handshake, many acknowledged writes, an end -- with the same
opportunities for a reset, a wedged link, or a cancellation. See
[Design notes](#design-notes) for why this was the simpler option.

A `FaultConfig` on the simulated board injects, on demand:

- **`FlashFault.RESET_MID_FLASH`** -- the device sends an explicit reset
  marker after N data blocks and then goes silent, standing in for a
  brown-out or watchdog reset partway through a write. Tested in
  `test_flasher_simulator.py::test_device_reset_mid_flash_is_a_structured_error_not_a_hang`.
- **`FlashFault.SILENT_TIMEOUT`** -- the device stops responding with no
  reset marker at all, standing in for a wedged link. Tested in
  `test_silent_timeout_surfaces_as_flash_timeout_error`.
- **`ReplFault.GARBAGE`** -- one raw-REPL response is replaced with bytes
  that don't match any expected marker, standing in for a dropped/corrupted
  byte. Tested in
  `test_raw_repl.py::test_garbage_on_the_line_is_detected_and_resynced_transparently`.
- **`block_delay_s`** -- adds latency per flash block, used to make
  cancellation-mid-flash deterministically testable without a real multi-
  second flash.

Cancellation itself doesn't need fault injection -- it's tested by cancelling
a real (simulated) flash in progress and asserting partial progress was
reported and, for the subprocess path, that the child process was actually
reaped (`test_flasher_subprocess.py::test_cancellation_leaves_no_orphan_process`).

Per-port locking is tested by timing two concurrent calls against the same
simulated port (they serialise) against two calls on different ports (they
don't) -- `test_toolkit.py::test_concurrent_calls_on_same_port_are_serialised`.

## Verification status

Everything in this repository was built and tested against the simulator
described above. **No physical ESP32 or other hardware was available or used
at any point.** `backends/serial_backend.py` (real pyserial + the `esptool`
CLI) is written to the same `BoardHandle` contract the simulator satisfies
and its subprocess-lifecycle logic (progress parsing, cancel-and-reap,
reset-string detection) is tested against a stand-in script
(`tests/fixtures/fake_esptool.py`) that mimics esptool's stdout shape --
but the module itself has never been run against a real board or even a
real copy of `esptool`. Treat it as "should work", not "verified".

## Installation

```bash
python -m venv .venv
.venv/bin/pip install -e ".[dev]"        # simulator only
.venv/bin/pip install -e ".[serial]"     # adds pyserial + esptool for real hardware
```

## Running it

```bash
mcp-esp32                 # starts the MCP server on stdio
mcp-esp32 --demo          # narrated walkthrough of every tool against the simulator, no MCP client needed
```

`--demo` calls `toolkit.py` directly (the same functions the tests call) and
prints each result, including a deliberate reset-mid-flash fault, so the
whole tool surface is visible without wiring up an MCP client.

## Testing

```bash
pytest
```

37 tests, no hardware, no network access, no secrets. CI
(`.github/workflows/ci.yml`) runs the suite plus `--demo` on Python 3.11 and
3.12.

## Design notes

Decisions made without a way to check them against real hardware; simpler
option chosen in each case.

- **Simulated flash protocol instead of real SLIP/ROM-bootloader bytes.**
  Reimplementing esptool's actual wire protocol (chip-specific stub loaders,
  ROM quirks) would be a second project and wouldn't change what's being
  demonstrated -- the MCP-level orchestration around a slow, faulty write.
  The simulator's protocol has the same shape (sync/begin/data×N/end,
  acknowledged blocks, an explicit reset signal) and is documented as such
  in `sim_flash_protocol.py`.
- **`fs_get`/`fs_put` transfer content as hex-encoded lines over repeated
  raw-REPL `exec()` calls**, not a dedicated binary protocol. MicroPython's
  raw REPL keeps globals alive across `exec()` calls (it's a persistent
  interpreter, not a fresh one per call), so `fs_put` opens a file handle in
  one call and writes to it in several more -- this works identically
  against real hardware and the simulator, and needed no new wire protocol.
- **Base64 in the MCP tool signatures, hex on the wire.** `fs_get`/`fs_put`
  take/return `content_base64` because raw bytes aren't JSON-safe; internally
  `fsops.py` uses `ubinascii.hexlify`/`unhexlify` since that's what
  MicroPython actually has available on-device.
  MAX_TRANSFER_BYTES (512 KiB) is checked before a transfer starts, not
  discovered partway through.
- **Ports named `SIM*` always route to the simulator** (`backends/base.py`),
  auto-created on first use. No environment variable or config flag needed
  to use the simulator -- call any tool with a port starting with `SIM` and
  it exists.
- **Errors are structured return values, not raised exceptions, for every
  expected failure mode** (reset, timeout, desync, confirmation-required).
  A tool call that hits one of these returns `{"error": {"kind": ..., ...}}`
  rather than an MCP tool error, so an agent can branch on `kind` without
  parsing a message string. Actual bugs still raise and surface as normal
  tool errors.
- **Cancellation cleanup runs inside an `anyio.CancelScope(shield=True)`.**
  Without shielding, the first checkpoint after catching the cancellation
  would immediately re-raise (cancel scopes are level-triggered), which
  would skip waiting for the flash thread to actually stop before the
  request is torn down.
- **esptool is shelled out to as a CLI subprocess, not used as a library.**
  Its Python API is not considered stable across versions; the CLI's stdout
  format (`(NN %)` progress lines, specific fatal-error strings) is what
  every existing esptool-wrapping tool already depends on, and it's the
  surface `tests/fixtures/fake_esptool.py` can stand in for without needing
  esptool internals to match.

## Requirements

Python 3.11+. `mcp` for the server itself; `pyserial` and `esptool` are
optional (only needed for the serial backend against real hardware) and
their absence never breaks import -- see `tests/test_optional_deps.py`.