feat(tools): mcp-server added monitor-device tool

Add a `monitor_device` MCP tool that lets an AI agent run a scripted,
non-interactive `esp-idf-monitor` session against a flashed device and
get back a short status plus a log file path, instead of raw serial
output inline.

Under the hood:

- The agent supplies a plain-text command body (expect/send/sleep/reset/
  exit/comments). `assemble_monitor_script_from_agent_commands()` frames
  it into a script the monitor's non-interactive command mode can
  consume via stdin: it appends `exit` if the agent didn't already end
  with one, and rewrites every bare `expect <regex>` into `expect
  --timeout <timeout_sec> <regex>` via `_monitor_normalize_expect_line()`
  (an already-bounded `expect --timeout ...` line is left untouched so
  the monitor itself reports a bad value). A leading `reset` is not
  prepended - the monitor already resets the chip when it opens the
  port - and any `reset` the agent wrote is left in place.
  `_monitor_parse_sleep_duration()` extracts each `sleep <n>` duration.
  The effective timeout is the sum of every bounded expect duration
  plus every sleep duration. Scripts whose sum exceeds
  `MONITOR_MAX_SCRIPT_SEC` are rejected. If the script has neither
  expect nor sleep (for example only `send`), `timeout_sec` is used so
  the process still has a kill bound.

- `monitor_device()` runs `python -m esp_idf_monitor` via
  `subprocess.run(..., input=script, timeout=2 * effective_timeout)`.
  `no_reset` is forwarded as `--no-reset` so the connection reset can be
  skipped; an explicit `-p` is forwarded when a port is given. Extra
  arguments match `idf.py monitor` where a build exists: baud (`baud`
  tool arg, else `monitor_baud` from `project_description.json`),
  toolchain prefix, `--target`/`--revision`, coredump/panic decode, and
  ELF files with the app ELF first. The 2x hard timeout is a safety net
  independent of the script's own `expect --timeout`/`exit` logic; on
  `TimeoutExpired` the process is killed but any output already captured
  is preserved and logged. `decode_stream()` normalizes that captured
  output, which can be `bytes` on the timeout path even though the
  process otherwise runs in text mode.

- The monitor's exit code drives the reported status via
  `_monitor_status()`, using `EXIT_EXPECT_TIMEOUT` and
  `EXIT_SCRIPT_ERROR` from `esp_idf_monitor.base.constants`: 0 is
  success, 110 means an `expect` pattern never showed up before its
  `--timeout` elapsed, 2 means the monitor rejected the script (bad
  syntax/timeout/regex), anything else is reported generically.

- Serial output and the monitor's own messages share one pipe
  (`stderr=STDOUT`) so decoded panic backtraces stay next to the lines
  that triggered them. `_save_monitor_output()` writes the full merge to
  `<tempdir>/esp_idf_mcp_log/action_monitor/monitor_<timestamp>.log` and
  reports a dedicated `Log file:` line. On non-zero exit or process
  kill, a short tail of that same merge is also returned inline so the
  agent has some failure context without a second file read. If the log
  file can't be written, it falls back to inlining a truncated tail.

Closes https://github.com/espressif/esp-idf/issues/18757
Closes https://github.com/espressif/esp-idf/pull/18385

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Marek Fiala
2026-08-13 10:34:13 +02:00
parent 7e5235a7a7
commit 3f005c8a4d
4 changed files with 824 additions and 23 deletions

View File

@@ -337,8 +337,9 @@ The MCP server provides the following tools:
- ``flash project``: Flash the built project to a connected device. Specify it by port name
- ``clean project``: Clean build artifacts
- ``create project``: Create a new ESP-IDF project from the sample template. Can be used before any project exists
- ``monitor device``: Run a scripted serial monitor session against a flashed device and wait for expected output. See :ref:`mcp-monitor-device`
All tools accept an optional ``project_dir`` argument. When omitted, the tool operates on the directory configured at startup (``-C`` flag or ``IDF_MCP_WORKSPACE_FOLDER``). You can instruct the AI model to use a specific project directory explicitly, for example when working with multiple projects or when no default project was configured at startup.
The project tools accept an optional ``project_dir`` argument. When omitted, the tool operates on the directory configured at startup (``-C`` flag or ``IDF_MCP_WORKSPACE_FOLDER``). You can instruct the AI model to use a specific project directory explicitly, for example when working with multiple projects or when no default project was configured at startup. The ``create project`` and ``monitor device`` tools are the exception: the former takes the parent ``path`` for the new project, and the latter talks to a device rather than to a project directory.
The MCP server also provides these resources:
@@ -346,6 +347,15 @@ The MCP server also provides these resources:
- ``project://status``: Get current project build status and artifacts
- ``project://devices``: Get list of connected devices
.. _mcp-monitor-device:
Monitoring a Device
^^^^^^^^^^^^^^^^^^^
The ``monitor device`` tool lets an AI assistant observe what a flashed device prints. Ask in ordinary language, for example "flash it and check that the device prints ``Minimum free heap size`` within 30 seconds". The assistant waits for that output and then tells you whether it appeared.
The full serial log is kept in a file. Ask the assistant to search that log if you want more detail than the short status it reports.
Adding ESP-IDF MCP Server to IDEs and AI agents
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

View File

@@ -337,8 +337,9 @@ MCP 服务器提供以下工具:
- ``flash project``:将已构建的项目烧录到已连接的设备,通过端口名称进行指定
- ``clean project``:清理构建产物
- ``create project``:基于示例模板创建新的 ESP-IDF 项目,可在尚无项目时使用
- ``monitor device``:在已烧录的设备上运行一次脚本化的串口监视会话,并等待预期的输出。参见 :ref:`mcp-monitor-device`
所有工具接受可选的 ``project_dir`` 参数。当省略该参数时,工具将默认使用启动时配置的目录(该目录可通过 ``-C`` 参数或 ``IDF_MCP_WORKSPACE_FOLDER`` 环境变量指定)。你可以要求 AI 模型明确指定某个项目目录,例如当同时处理多个项目,或启动时未配置默认项目的情况下。
项目工具接受可选的 ``project_dir`` 参数。当省略该参数时,工具将默认使用启动时配置的目录(该目录可通过 ``-C`` 参数或 ``IDF_MCP_WORKSPACE_FOLDER`` 环境变量指定)。在同时处理多个项目,或启动时未配置默认项目的情况下,你可以明确指示 AI 模型使用某个特定的项目目录。``create project````monitor device`` 工具是例外:前者使用新项目的上级 ``path``;后者与设备交互,而不是针对项目目录。
同时提供以下资源:
@@ -346,6 +347,15 @@ MCP 服务器提供以下工具:
- ``project://status``:获取当前项目的构建状态和构建产物
- ``project://devices``:获取已连接的设备列表
.. _mcp-monitor-device:
监视设备
^^^^^^^^
AI 助手可以通过 ``monitor device`` 工具观察已烧录设备的打印输出,只需用自然语言提问即可。例如“烧录并检查设备是否在 30 秒内打印 ``Minimum free heap size``”。AI 助手会等待该输出,然后告知你结果。
完整的串口日志会保存在一个文件中。如果你需要了解详细信息,可以让 AI 助手搜索日志文件。
将 ESP-IDF MCP 服务器添加至 IDE 和 AI 智能体
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

View File

@@ -2,16 +2,23 @@
# SPDX-License-Identifier: Apache-2.0
import json
import math
import os
import subprocess
import sys
import tempfile
import time
from pathlib import Path
from typing import Annotated
from typing import Any
from esp_idf_monitor.base.constants import EXIT_EXPECT_TIMEOUT
from esp_idf_monitor.base.constants import EXIT_SCRIPT_ERROR
from rich_click import Context
from idf_py_actions.errors import FatalError
from idf_py_actions.tools import PropertyDict
from idf_py_actions.tools import get_sdkconfig_value
from idf_py_actions.tools import get_target
from idf_py_actions.tools import idf_version
@@ -39,6 +46,37 @@ except ImportError:
MCP_AVAILABLE = False
try:
# Field() feeds per-argument descriptions into a tool's inputSchema (see the
# Annotated[...] argument hints on monitor_device below).
# Annotated evaluates Field(...) at import time, and this whole module is imported
# unconditionally by idf.py on every run, regardless of the command invoked - so without
# this fallback, idf.py would break entirely on installs that lack the optional
# "mcp" feature. pydantic ships as a dependency of mcp, so this import only
# fails when mcp itself is not installed either (see MCP_AVAILABLE above).
from pydantic import Field
except ImportError:
def Field(**kwargs: Any) -> Any: # type: ignore[misc] # noqa: N802
return None
# Root folder (inside the OS temp directory) that ESP-IDF MCP tool logs are
# written under, one subfolder per tool.
MCP_LOG_ROOT_DIR_NAME = 'esp_idf_mcp_log'
# Bounded wait injected into 'expect' lines that arrive without --timeout.
DEFAULT_MONITOR_TIMEOUT_SEC = 20.0
# Whole-script cap: sum of expect and sleep durations.
MONITOR_MAX_SCRIPT_SEC = 600.0
# Commands the MCP monitor script may contain. Everything else (flash, log, ...)
# is rejected so this tool cannot trigger a rebuild or toggle monitor UI state.
MONITOR_SCRIPT_COMMANDS = frozenset({'expect', 'send', 'sleep', 'reset', 'bootloader', 'exit'})
# Size limit for text embedded directly in a tool result, in characters. Serial
# output belongs in the log file, not in the agent's context.
MONITOR_TAIL_CHARS = 1500
def _is_valid_project_dir(directory: str) -> bool:
"""
Determine if the given directory is a valid ESP-IDF project directory.
@@ -71,6 +109,160 @@ def _is_valid_project_dir(directory: str) -> bool:
return False
def _derive_build_dir(effective_dir: str, launch_dir: str, args: PropertyDict) -> str:
"""Use idf.py's build_dir when it belongs to this project; otherwise ``<project>/build``."""
if launch_dir == effective_dir:
build_dir = args.get('build_dir', '')
if build_dir:
return str(build_dir)
return os.path.join(effective_dir, 'build')
def _load_project_description(
launch_dir: str, args: PropertyDict
) -> tuple[str | None, str | None, dict[str, Any] | None]:
"""Resolve project dir, build dir, and ``project_description.json`` without failing.
Returns ``(project_dir, build_dir, description)``. Any piece that cannot be
resolved is ``None``. Missing project, build directory, or description is not
an error.
"""
project_dir = resolve_default_project_dir(launch_dir)
if project_dir is None:
return None, None, None
build_dir = _derive_build_dir(project_dir, launch_dir, args)
if not os.path.exists(build_dir):
return project_dir, None, None
desc_path = os.path.join(build_dir, 'project_description.json')
try:
with open(desc_path, encoding='utf-8') as f:
description = json.load(f)
except (OSError, ValueError):
return project_dir, build_dir, None
if not isinstance(description, dict):
return project_dir, build_dir, None
return project_dir, build_dir, description
def _monitor_normalize_expect_line(line: str, default_expect_duration: float) -> tuple[str, float | None]:
"""Inject '--timeout default_expect_duration' into a bare 'expect' line.
Returns (line, expect_duration): expect_duration is the bound this line
ends up with, or None if the line already carries a --timeout the monitor
will reject - left untouched so the monitor reports the script error
itself. Timeout syntax is parsed the same way CommandReader parses it.
"""
command, _, argument = line.partition(' ')
argument = argument.strip()
if argument.startswith('--timeout'):
parts = argument.split(None, 2)
try:
expect_duration = float(parts[1])
except (IndexError, ValueError):
return line, None
if not math.isfinite(expect_duration) or expect_duration <= 0:
return line, None
return line, expect_duration
return (
' '.join(filter(None, [command, '--timeout', f'{default_expect_duration:g}', argument])),
default_expect_duration,
)
def _monitor_parse_sleep_duration(line: str) -> float | None:
"""Return the duration of a 'sleep' line, or None if the monitor would
reject it (such a line is skipped, so it costs no time)."""
argument = line.partition(' ')[2].strip()
try:
sleep_duration = float(argument)
except ValueError:
return None
if not math.isfinite(sleep_duration) or sleep_duration <= 0:
return None
return sleep_duration
def _save_monitor_output(output: str) -> str:
"""Save the monitor's output to a log file and describe where it went, or
inline a tail on failure - keeps the full serial dump out of the agent's context.
*output* is the monitor's merged stdout and stderr: the chip prints serial
data on stdout while the monitor's own '--- ' lines (decoded panic
backtraces among them) go to stderr, so both streams are captured through
one pipe to keep a decoded backtrace next to the raw one that triggered it.
"""
log_dir = Path(tempfile.gettempdir()) / MCP_LOG_ROOT_DIR_NAME / 'action_monitor'
log_path = log_dir / f'monitor_{time.strftime("%Y%m%d_%H%M%S")}.log'
try:
log_dir.mkdir(parents=True, exist_ok=True)
log_path.write_text(output, encoding='utf-8', errors='replace')
# Path on its own line so an agent cannot paraphrase the character
# count and drop the location.
return f'Log file: {log_path}\nMonitor output: {len(output)} characters. Read or grep that file.'
except OSError as e:
return f'Log file could not be written ({e}), so here is its tail instead:\n{tail(output, MONITOR_TAIL_CHARS)}'
def _monitor_status(returncode: int) -> str:
"""Explain a monitor exit code, so the agent does not have to scrape stderr."""
if returncode == 0:
return 'Monitor script completed successfully (exit code 0).'
if returncode == EXIT_EXPECT_TIMEOUT:
return (
f'An expect pattern was not seen before its --timeout elapsed (exit code {returncode}). '
'The monitor aborted the rest of the script.'
)
if returncode == EXIT_SCRIPT_ERROR:
return (
f'The monitor rejected the script (exit code {returncode}): invalid expect syntax, '
'timeout value or regular expression.'
)
return f'The monitor exited with code {returncode}.'
def _monitor_project_args(baud: str | int | None, launch_dir: str, args: PropertyDict) -> list[str]:
"""Baud, optional decode metadata, and ELF files derived from the project build."""
extra: list[str] = []
_, build_dir, description = _load_project_description(launch_dir, args)
# baud: explicit argument, else monitor_baud from the description
if baud is not None:
extra += ['-b', str(baud)]
elif description is not None and description.get('monitor_baud') is not None:
extra += ['-b', str(description['monitor_baud'])]
if description is None or not build_dir:
return extra
# ELF files (app ELF first, same order as idf.py monitor)
elf_file = os.path.join(build_dir, description.get('app_elf', '') or '')
elf_list = [str(elf) for elf in Path(build_dir).rglob('*.elf')]
if elf_file and elf_file in elf_list:
elf_list.insert(0, elf_list.pop(elf_list.index(elf_file)))
# without ELF files, metadata has no additional value to monitor
if not elf_list:
return extra
# toolchain / target / decode flags (only useful with ELFs)
toolchain_prefix = description.get('monitor_toolprefix')
if toolchain_prefix:
extra += ['--toolchain-prefix', str(toolchain_prefix)]
config_file = str(description.get('config_file') or '')
coredump_decode = get_sdkconfig_value(config_file, 'CONFIG_ESP_COREDUMP_DECODE')
if coredump_decode is not None:
extra += ['--decode-coredumps', coredump_decode]
target = description.get('target')
if target:
extra += ['--target', str(target)]
revision = description.get('min_rev')
if revision:
extra += ['--revision', str(revision)]
if get_sdkconfig_value(config_file, 'CONFIG_IDF_TARGET_ARCH_RISCV'):
extra += ['--decode-panic', 'backtrace']
extra += elf_list
return extra
def resolve_default_project_dir(launch_dir: str) -> str | None:
"""
Returns the first valid ESP-IDF project directory from the server's launch
@@ -110,6 +302,88 @@ def resolve_tool_project_dir(explicit_dir: str | None, launch_dir: str) -> tuple
)
def assemble_monitor_script_from_agent_commands(
commands: str,
timeout_sec: float = DEFAULT_MONITOR_TIMEOUT_SEC,
) -> tuple[str, float]:
"""Frame the agent's command body into a runnable monitor script.
Bounds every 'expect' with --timeout (default *timeout_sec*) and appends
'exit' if missing, so the monitor always terminates on its own. A leading
'reset' is not added: the monitor already resets the chip when it opens the
port. Agent-supplied 'reset' lines are left in place.
Returns (script, effective_timeout). Raises ValueError if *commands* has no
command at all, if a line is not one of MONITOR_SCRIPT_COMMANDS, or if the
script would run longer than MONITOR_MAX_SCRIPT_SEC.
effective_timeout is the sum of every bounded expect duration plus every
sleep duration. If the script has neither (for example only 'send'),
*timeout_sec* is used instead so the monitor process still has a kill bound.
"""
lines: list[str] = []
effective_timeout = 0.0
has_command = False
allowed = ', '.join(sorted(MONITOR_SCRIPT_COMMANDS))
for raw_line in commands.splitlines():
line = raw_line.strip()
if not line:
continue
if line.startswith('#'):
lines.append(line)
continue
command = line.partition(' ')[0].lower()
if command not in MONITOR_SCRIPT_COMMANDS:
raise ValueError(f'Unsupported monitor command {command!r}. Allowed commands: {allowed}.')
has_command = True
if command == 'expect':
line, expect_duration = _monitor_normalize_expect_line(line, timeout_sec)
if expect_duration is not None:
effective_timeout += expect_duration
elif command == 'sleep':
sleep_duration = _monitor_parse_sleep_duration(line)
if sleep_duration is not None:
effective_timeout += sleep_duration
lines.append(line)
if not has_command:
raise ValueError(
'No monitor commands given. Pass a commands body with at least one command, '
'for example: expect --timeout 20 Hello world!'
)
if not lines or lines[-1].lower() != 'exit':
lines.append('exit')
if effective_timeout <= 0:
effective_timeout = timeout_sec
if effective_timeout > MONITOR_MAX_SCRIPT_SEC:
raise ValueError(f'This script would run for {effective_timeout:g}s. The limit is {MONITOR_MAX_SCRIPT_SEC:g}s.')
return '\n'.join(lines) + '\n', effective_timeout
def decode_stream(stream: Any) -> str:
"""Decode a subprocess stream that may be bytes even in text mode.
TimeoutExpired carries the output collected so far as bytes on POSIX, while
a completed run in text mode yields str.
"""
if stream is None:
return ''
if isinstance(stream, bytes):
return stream.decode('utf-8', errors='replace')
return str(stream)
def tail(text: str, limit: int) -> str:
"""Return at most the last *limit* characters of *text*, marking a cut."""
text = text.strip()
if len(text) <= limit:
return text
return f'[...truncated...]\n{text[-limit:]}'
def action_extensions(base_actions: dict, project_path: str) -> dict:
"""ESP-IDF MCP Server Extension"""
@@ -274,6 +548,126 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
print(f'ERROR: Flash failed: {str(e)}', file=sys.stderr)
return f'Error flashing: {str(e)}'
@mcp.tool(
description=(
'Run a serial monitor session against a flashed device, driven by a script '
'(runs `python -m esp_idf_monitor` in non-interactive command mode). '
'Use this to observe what a device prints and to wait for specific output. '
'Write the commands body from the project context or from the exact instructions '
'of the user; allowed commands are expect, send, sleep, reset, bootloader and exit. '
'The monitor already resets the chip when it opens the port, so do not start the '
'script with reset unless you need an extra reset later. Set no_reset to skip that '
'connection reset. ALWAYS end the script with an explicit exit as its last line. '
'Every expect should be bounded: "expect --timeout <seconds> <regex>". '
f'The whole script may take at most {MONITOR_MAX_SCRIPT_SEC:g} seconds. '
'Exit code 0 means the script finished, 110 means an expect pattern never appeared, '
'2 means the script itself was invalid. '
'Returns a short status plus a Log file: line with the absolute path of the monitor '
'log, which holds the serial output together with the monitor messages. '
'Always quote that path when answering the user; read or '
'grep the file instead of expecting the whole log inline. '
'The port is optional and is autodetected when omitted; see project://devices '
'when several devices are connected.'
)
)
def monitor_device(
commands: Annotated[
str,
Field(
description=(
'Monitor script body, one command per line (expect, send, sleep, reset, bootloader, exit). '
'Bound every wait as "expect --timeout <seconds> <regex>" (default 20 seconds). '
f'The whole script may take at most {MONITOR_MAX_SCRIPT_SEC:g} seconds. '
'A reset at the start is unnecessary: the monitor resets on connect unless no_reset. '
'Leave any later reset in the script. End with exit; the server appends one if missing. '
'Lines starting with # are comments.'
)
),
],
timeout_sec: Annotated[
float,
Field(
description=(
'Wait, in seconds, injected into expect lines that omit --timeout. '
f'The whole script may take at most {MONITOR_MAX_SCRIPT_SEC:g} seconds, '
'the sum of expect and sleep durations. The monitor process is killed after '
'twice that time. If the script has no expect or sleep, twice timeout_sec is used.'
)
),
] = DEFAULT_MONITOR_TIMEOUT_SEC,
port: str | None = None,
baud: str | int | None = None,
no_reset: bool = False,
) -> str:
"""Run a scripted esp-idf-monitor session and return a status plus a log path.
Args:
port: Optional serial port such as /dev/ttyUSB0 or COM3. Leave as
None to let the monitor autodetect it.
baud: Optional monitor baud rate. When omitted, ``monitor_baud``
from ``project_description.json`` is used if that file exists.
no_reset: Pass --no-reset so the monitor does not reset the chip
when it opens the port. Agent-supplied reset commands in the
script are left unchanged.
"""
if not math.isfinite(timeout_sec) or timeout_sec <= 0:
return f'timeout_sec must be a finite number greater than 0, got {timeout_sec}.'
try:
script, effective_timeout = assemble_monitor_script_from_agent_commands(
commands, timeout_sec=timeout_sec
)
except ValueError as e:
return str(e)
hard_timeout = 2 * effective_timeout
cmd = [sys.executable, '-m', 'esp_idf_monitor']
if port:
cmd.extend(['-p', port])
cmd.extend(_monitor_project_args(baud, project_path, args))
if no_reset:
cmd.append('--no-reset')
print(f'INFO: Running monitor: {" ".join(cmd)} (hard timeout {hard_timeout:g}s)', file=sys.stderr)
print(f'INFO: Monitor script:\n{script}', file=sys.stderr)
try:
result = subprocess.run(
cmd,
input=script,
stdout=subprocess.PIPE,
# One pipe for both streams keeps the monitor's decoded backtraces
# in order with the serial lines they belong to.
stderr=subprocess.STDOUT,
text=True,
timeout=hard_timeout,
)
except subprocess.TimeoutExpired as e:
print(f'ERROR: Monitor killed after {hard_timeout:g}s', file=sys.stderr)
output = decode_stream(e.output)
parts = [
f'The monitor did not exit on its own and was killed after {hard_timeout:g} seconds '
'(twice the time the script should have taken). '
'The serial output captured until then was kept.',
_save_monitor_output(output),
]
output_tail = tail(output, MONITOR_TAIL_CHARS)
if output_tail:
parts.append(f'Last output before the kill:\n{output_tail}')
return '\n'.join(parts)
except Exception as e:
print(f'ERROR: Monitor failed to run: {e}', file=sys.stderr)
return f'Failed to run the monitor: {e}'
print(f'INFO: Monitor exited with code {result.returncode}', file=sys.stderr)
output = result.stdout or ''
parts = [_monitor_status(result.returncode), _save_monitor_output(output)]
if result.returncode != 0:
output_tail = tail(output, MONITOR_TAIL_CHARS)
if output_tail:
parts.append(f'Last output before the exit:\n{output_tail}')
return '\n'.join(parts)
@mcp.tool(
description=(
'Create a new ESP-IDF project from the sample template (runs `idf.py create-project`). '
@@ -354,7 +748,7 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
@mcp.resource('project://config')
def get_project_config() -> str:
"""Get current project configuration"""
effective_dir = resolve_default_project_dir(project_path)
effective_dir, build_dir, description = _load_project_description(project_path, args)
config: dict[str, Any] = {}
if effective_dir is None:
@@ -365,24 +759,15 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
return json.dumps(config, indent=2)
config['project_path'] = effective_dir
# Use the build_dir from idf.py args when the project matches; otherwise derive it.
build_dir = (
args.get('build_dir', '') if project_path == effective_dir else os.path.join(effective_dir, 'build')
)
if not os.path.exists(build_dir):
if build_dir is None:
config['build_dir_exists'] = False
return json.dumps(config, indent=2)
config['build_dir'] = build_dir
proj_desc_fn = os.path.join(build_dir, 'project_description.json')
config['project_description'] = 'Project description does not exist'
try:
with open(proj_desc_fn, encoding='utf-8') as f:
config['project_description'] = json.load(f)
except (OSError, ValueError):
pass
if description is not None:
config['project_description'] = description
else:
config['project_description'] = 'Project description does not exist'
return json.dumps(config, indent=2)
@@ -405,10 +790,7 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
status['target'] = get_target(effective_dir)
status['idf_version'] = idf_version()
# Use the build_dir from idf.py args when the project matches; otherwise derive it.
build_dir = (
args.get('build_dir', '') if project_path == effective_dir else os.path.join(effective_dir, 'build')
)
build_dir = _derive_build_dir(effective_dir, project_path, args)
if os.path.exists(build_dir):
status['build_dir'] = build_dir
artifacts = ['bootloader', 'partition_table', 'app-flash', 'flash_args']

View File

@@ -11,7 +11,9 @@ import importlib
import importlib.util
import json
import os
import subprocess
import sys
import tempfile
import types
from collections.abc import Callable
from pathlib import Path
@@ -106,6 +108,7 @@ def mcp_ext(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[types.Modu
tools_mod.PropertyDict = dict # type: ignore[attr-defined]
tools_mod.get_target = mock.Mock(return_value='esp32') # type: ignore[attr-defined]
tools_mod.idf_version = mock.Mock(return_value='5.4.0') # type: ignore[attr-defined]
tools_mod.get_sdkconfig_value = mock.Mock(return_value=None) # type: ignore[attr-defined]
idf_py_actions_pkg.errors = errors_mod # type: ignore[attr-defined]
idf_py_actions_pkg.tools = tools_mod # type: ignore[attr-defined]
@@ -115,6 +118,13 @@ def mcp_ext(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[types.Modu
mcp_server_pkg = types.ModuleType('mcp.server')
mcp_server_pkg.MCPServer = lambda name: mock_mcp_instance # type: ignore[attr-defined]
# Stub monitor exit codes (same values as esp_idf_monitor.base.constants).
esp_idf_monitor_pkg = types.ModuleType('esp_idf_monitor')
esp_idf_monitor_base = types.ModuleType('esp_idf_monitor.base')
esp_idf_monitor_constants = types.ModuleType('esp_idf_monitor.base.constants')
esp_idf_monitor_constants.EXIT_EXPECT_TIMEOUT = 110 # type: ignore[attr-defined]
esp_idf_monitor_constants.EXIT_SCRIPT_ERROR = 2 # type: ignore[attr-defined]
stubs = {
'rich_click': rich_click,
'idf_py_actions': idf_py_actions_pkg,
@@ -122,6 +132,9 @@ def mcp_ext(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[types.Modu
'idf_py_actions.tools': tools_mod,
'mcp': mcp_pkg,
'mcp.server': mcp_server_pkg,
'esp_idf_monitor': esp_idf_monitor_pkg,
'esp_idf_monitor.base': esp_idf_monitor_base,
'esp_idf_monitor.base.constants': esp_idf_monitor_constants,
}
for name, stub_mod in stubs.items():
monkeypatch.setitem(sys.modules, name, stub_mod)
@@ -176,13 +189,17 @@ class TestIsValidProjectDir:
mod, _ = mcp_ext
assert mod._is_valid_project_dir(str(tmp_path / 'nonexistent')) is False
def test_directory_without_cmakelists(self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockMCPServer]) -> None:
def test_directory_without_cmakelists(
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockMCPServer]
) -> None:
mod, _ = mcp_ext
d = tmp_path / 'no_cmake'
d.mkdir()
assert mod._is_valid_project_dir(str(d)) is False
def test_cmakelists_without_idf_line(self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockMCPServer]) -> None:
def test_cmakelists_without_idf_line(
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockMCPServer]
) -> None:
mod, _ = mcp_ext
proj = _make_invalid_project(tmp_path / 'plain')
assert mod._is_valid_project_dir(str(proj)) is False
@@ -481,6 +498,387 @@ class TestCleanProject:
assert cmd[cmd.index('-C') + 1] == str(proj)
# ---------------------------------------------------------------------------
# Tests: monitor script building
# ---------------------------------------------------------------------------
class TestBuildMonitorScript:
"""Covers only what the MCP server itself is responsible for: framing the
agent's command body (exit/comments) and computing the effective
timeout. Validity of esp-idf-monitor's own 'expect --timeout' syntax is
esp-idf-monitor's job, not duplicated here."""
def test_commands_without_an_actual_command_are_rejected(
self, mcp_ext: tuple[types.ModuleType, _MockMCPServer]
) -> None:
mod, _ = mcp_ext
with pytest.raises(ValueError, match='No monitor commands given'):
mod.assemble_monitor_script_from_agent_commands(' \n# just a note\n')
def test_bare_expect_gets_default_timeout_and_framing(
self, mcp_ext: tuple[types.ModuleType, _MockMCPServer]
) -> None:
mod, _ = mcp_ext
script, effective = mod.assemble_monitor_script_from_agent_commands('expect Hello world!')
assert script == 'expect --timeout 20 Hello world!\nexit\n'
assert effective == 20.0
@pytest.mark.parametrize(
'commands, timeout_sec, expected_effective, expected_snippet',
[
('expect --timeout 45 ALL TESTS PASSED', 20.0, 45.0, 'expect --timeout 45 ALL TESTS PASSED'),
(
'expect --timeout 5 first\nexpect --timeout 90 second\nexpect third',
30.0,
125.0,
'expect --timeout 30 third',
),
('sleep 120\nexpect ready\nsleep 30', 20.0, 170.0, 'expect --timeout 20 ready'),
('expect --timeout abc pattern', 20.0, 20.0, 'expect --timeout abc pattern'),
('send hello', 20.0, 20.0, 'send hello'),
],
ids=[
'keeps_explicit_timeout',
'expect_timeouts_are_summed_and_default_is_injected',
'sleeps_are_added_to_the_sum',
'malformed_timeout_left_to_the_monitor',
'send_only_uses_timeout_sec_as_kill_bound',
],
)
def test_timeout_handling(
self,
mcp_ext: tuple[types.ModuleType, _MockMCPServer],
commands: str,
timeout_sec: float,
expected_effective: float,
expected_snippet: str,
) -> None:
mod, _ = mcp_ext
script, effective = mod.assemble_monitor_script_from_agent_commands(commands, timeout_sec=timeout_sec)
assert expected_snippet in script
assert effective == expected_effective
def test_script_at_max_duration_is_accepted(self, mcp_ext: tuple[types.ModuleType, _MockMCPServer]) -> None:
mod, _ = mcp_ext
_, effective = mod.assemble_monitor_script_from_agent_commands(
'sleep 580\nexpect --timeout 20 ready', timeout_sec=20.0
)
assert effective == 600.0
@pytest.mark.parametrize(
'commands, timeout_sec',
[
('sleep 99999\nexpect ready', 20.0),
('expect --timeout 601 ready', 20.0),
('expect --timeout 400 a\nexpect --timeout 250 b', 20.0),
],
ids=['long_sleep', 'long_expect', 'sum_of_expects'],
)
def test_script_over_max_duration_is_rejected(
self, mcp_ext: tuple[types.ModuleType, _MockMCPServer], commands: str, timeout_sec: float
) -> None:
mod, _ = mcp_ext
with pytest.raises(ValueError, match='The limit is'):
mod.assemble_monitor_script_from_agent_commands(commands, timeout_sec=timeout_sec)
@pytest.mark.parametrize(
'commands, expected_script',
[
('expect uptime', 'expect --timeout 20 uptime\nexit\n'),
('reset\nexpect done\nexit\n', 'reset\nexpect --timeout 20 done\nexit\n'),
(
'expect first\nreset\nexpect second',
'expect --timeout 20 first\nreset\nexpect --timeout 20 second\nexit\n',
),
],
ids=['exit_appended_and_reset_not_prepended', 'agent_reset_and_exit_kept', 'mid_script_reset_kept'],
)
def test_script_framing(
self, mcp_ext: tuple[types.ModuleType, _MockMCPServer], commands: str, expected_script: str
) -> None:
mod, _ = mcp_ext
script, _ = mod.assemble_monitor_script_from_agent_commands(commands)
assert script == expected_script
@pytest.mark.parametrize('command', ['flash', 'app-flash', 'log', 'output'])
def test_disallowed_commands_are_rejected(
self, mcp_ext: tuple[types.ModuleType, _MockMCPServer], command: str
) -> None:
mod, _ = mcp_ext
with pytest.raises(ValueError, match=rf'Unsupported monitor command {command!r}'):
mod.assemble_monitor_script_from_agent_commands(f'{command}\nexpect ready')
# ---------------------------------------------------------------------------
# Tests: monitor_device tool
# ---------------------------------------------------------------------------
@pytest.fixture()
def monitor_tools(
mcp_ext: tuple[types.ModuleType, _MockMCPServer],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> dict[str, Callable[..., Any]]:
"""Registered tools with monitor logs redirected into tmp_path."""
_, mock_mcp = mcp_ext
monkeypatch.setattr(tempfile, 'tempdir', str(tmp_path))
tools, _ = _start_server(mcp_ext, mock_mcp, str(tmp_path))
return tools
def _log_path_from(result: str) -> Path:
"""Extract the log file path the tool reported."""
for line in result.splitlines():
if line.startswith('Log file: '):
return Path(line.split(': ', 1)[1])
raise AssertionError(f'no serial log path in result:\n{result}')
class TestMonitorDevice:
"""Covers what the MCP tool itself is responsible for: turning a script and
exit code into a status message, wiring subprocess/log-file plumbing, and
forwarding its own arguments. Behaviour of esp-idf-monitor's non-interactive
command mode (e.g. what makes an 'expect --timeout' line valid) is exercised
by esp-idf-monitor's own tests, not duplicated here."""
@pytest.mark.parametrize(
'commands, timeout_sec, expected_snippet',
[
(' ', 20.0, 'No monitor commands given'),
('expect ready', 0, 'timeout_sec must be a finite number greater than 0'),
('flash\nexpect ready', 20.0, "Unsupported monitor command 'flash'"),
],
ids=['empty_commands', 'invalid_timeout_sec', 'disallowed_command'],
)
def test_guard_clauses_return_error_without_running(
self, monitor_tools: dict[str, Callable[..., Any]], commands: str, timeout_sec: float, expected_snippet: str
) -> None:
with mock.patch('subprocess.run') as mock_run:
result = monitor_tools['monitor_device'](commands=commands, timeout_sec=timeout_sec)
assert expected_snippet in result
mock_run.assert_not_called()
@pytest.mark.parametrize(
'returncode, output, expected_status',
[
(0, 'Hello world!\n', 'completed successfully (exit code 0)'),
(110, "boot\nExpect pattern 'nope' timed out after 20.0s", 'not seen before its --timeout elapsed'),
(2, 'Invalid expect timeout value: (must be a finite number > 0)', 'rejected the script'),
(1, 'could not open port', 'exited with code 1'),
],
ids=['success', 'expect_timeout', 'script_error', 'unexpected_code'],
)
def test_status_message_and_log_reflect_exit_code(
self,
monitor_tools: dict[str, Callable[..., Any]],
returncode: int,
output: str,
expected_status: str,
) -> None:
with mock.patch('subprocess.run') as mock_run:
mock_run.return_value = mock.Mock(returncode=returncode, stdout=output, stderr=None)
result = monitor_tools['monitor_device'](commands='expect ready')
assert expected_status in result
if returncode != 0:
# Why the monitor stopped is its last output, so the tail carries it.
assert output in result
log_file = _log_path_from(result)
assert log_file.is_absolute()
assert log_file.read_text(encoding='utf-8') == output
@pytest.mark.parametrize(
'call_kwargs, expected_in_cmd, expected_not_in_cmd',
[
({'port': '/dev/ttyUSB0'}, ['-p', '/dev/ttyUSB0'], []),
({'no_reset': True}, ['--no-reset'], []),
],
ids=['port_forwarded', 'no_reset_passed_to_the_monitor'],
)
def test_cli_arguments_are_forwarded(
self,
monitor_tools: dict[str, Callable[..., Any]],
call_kwargs: dict[str, Any],
expected_in_cmd: list[str],
expected_not_in_cmd: list[str],
) -> None:
with mock.patch('subprocess.run') as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr='')
monitor_tools['monitor_device'](commands='expect ready', **call_kwargs)
cmd = mock_run.call_args[0][0]
assert cmd[:3] == [sys.executable, '-m', 'esp_idf_monitor']
for item in expected_in_cmd:
assert item in cmd
for item in expected_not_in_cmd:
assert item not in cmd
def test_hard_timeout_is_twice_the_effective_timeout(self, monitor_tools: dict[str, Callable[..., Any]]) -> None:
"""The effective-timeout math itself (sum of waits plus sleeps) is covered
by TestBuildMonitorScript; this only checks the tool wires 2x it in."""
with mock.patch('subprocess.run') as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr='')
monitor_tools['monitor_device'](commands='sleep 60\nexpect --timeout 20 ready')
assert mock_run.call_args[1]['timeout'] == 2 * 80.0
def test_process_timeout_keeps_partial_output(self, monitor_tools: dict[str, Callable[..., Any]]) -> None:
with mock.patch('subprocess.run') as mock_run:
mock_run.side_effect = subprocess.TimeoutExpired(
cmd=['python', '-m', 'esp_idf_monitor'],
timeout=60.0,
output=b'partial serial output\nExpect still waiting\n',
)
result = monitor_tools['monitor_device'](commands='expect ready')
assert 'was killed after 40 seconds' in result
assert 'Expect still waiting' in result
logged = _log_path_from(result).read_text(encoding='utf-8')
assert logged == 'partial serial output\nExpect still waiting\n'
def test_launch_failure_is_reported(self, monitor_tools: dict[str, Callable[..., Any]]) -> None:
with mock.patch('subprocess.run') as mock_run:
mock_run.side_effect = OSError('no such interpreter')
result = monitor_tools['monitor_device'](commands='expect ready')
assert 'Failed to run the monitor' in result
assert 'no such interpreter' in result
def test_log_write_failure_falls_back_to_inline_tail(
self,
monitor_tools: dict[str, Callable[..., Any]],
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(Path, 'mkdir', mock.Mock(side_effect=OSError('disk full')))
with mock.patch('subprocess.run') as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout='important tail\n', stderr='')
result = monitor_tools['monitor_device'](commands='expect ready')
assert 'Log file could not be written (disk full)' in result
assert 'important tail' in result
def test_long_output_is_not_inlined(self, monitor_tools: dict[str, Callable[..., Any]]) -> None:
"""The whole point of the log file: a huge serial dump must not come back
through the tool result."""
big_output = 'x' * 200000
with mock.patch('subprocess.run') as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout=big_output, stderr='')
result = monitor_tools['monitor_device'](commands='expect ready')
assert len(result) < 500
assert _log_path_from(result).read_text(encoding='utf-8') == big_output
def test_both_streams_are_captured_through_one_pipe(self, monitor_tools: dict[str, Callable[..., Any]]) -> None:
"""esp-idf-monitor prints decoded panic backtraces on stderr, so a log
built from stdout alone would lose exactly what the ELF files enable."""
with mock.patch('subprocess.run') as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr=None)
monitor_tools['monitor_device'](commands='expect ready')
assert mock_run.call_args[1]['stderr'] == subprocess.STDOUT
def test_no_baud_elf_or_metadata_when_description_absent(
self, monitor_tools: dict[str, Callable[..., Any]]
) -> None:
with mock.patch('subprocess.run') as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr='')
monitor_tools['monitor_device'](commands='expect ready')
cmd = mock_run.call_args[0][0]
assert '-b' not in cmd
assert '--toolchain-prefix' not in cmd
assert '--target' not in cmd
assert '--revision' not in cmd
assert '--decode-coredumps' not in cmd
assert '--decode-panic' not in cmd
assert not any(str(arg).endswith('.elf') for arg in cmd)
def test_forwards_baud_elf_order_and_metadata_when_description_has_elfs(
self,
mcp_ext: tuple[types.ModuleType, _MockMCPServer],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
mod, mock_mcp = mcp_ext
monkeypatch.setattr(tempfile, 'tempdir', str(tmp_path))
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
proj = _make_valid_project(tmp_path / 'proj')
build_dir = proj / 'build'
boot_dir = build_dir / 'bootloader'
boot_dir.mkdir(parents=True)
app_elf = build_dir / 'hello_world.elf'
boot_elf = boot_dir / 'bootloader.elf'
app_elf.write_bytes(b'')
boot_elf.write_bytes(b'')
(build_dir / 'project_description.json').write_text(
json.dumps(
{
'app_elf': 'hello_world.elf',
'monitor_baud': '115200',
'monitor_toolprefix': 'xtensa-esp32-elf-',
'target': 'esp32',
'min_rev': '3',
'config_file': str(proj / 'sdkconfig'),
}
),
encoding='utf-8',
)
monkeypatch.setattr(
mod,
'get_sdkconfig_value',
lambda _path, key: {
'CONFIG_ESP_COREDUMP_DECODE': 'info',
'CONFIG_IDF_TARGET_ARCH_RISCV': 'y',
}.get(key),
)
tools, _ = _start_server(mcp_ext, mock_mcp, str(proj))
with mock.patch('subprocess.run') as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr='')
tools['monitor_device'](commands='expect ready')
cmd = mock_run.call_args[0][0]
assert cmd[cmd.index('-b') + 1] == '115200'
elf_args = [arg for arg in cmd if str(arg).endswith('.elf')]
assert elf_args[0] == str(app_elf)
assert str(boot_elf) in elf_args
assert cmd[cmd.index('--toolchain-prefix') + 1] == 'xtensa-esp32-elf-'
assert cmd[cmd.index('--target') + 1] == 'esp32'
assert cmd[cmd.index('--revision') + 1] == '3'
assert cmd[cmd.index('--decode-coredumps') + 1] == 'info'
assert cmd[cmd.index('--decode-panic') + 1] == 'backtrace'
def test_explicit_baud_overrides_description_monitor_baud(
self,
mcp_ext: tuple[types.ModuleType, _MockMCPServer],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
_, mock_mcp = mcp_ext
monkeypatch.setattr(tempfile, 'tempdir', str(tmp_path))
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
proj = _make_valid_project(tmp_path / 'proj')
build_dir = proj / 'build'
build_dir.mkdir()
(build_dir / 'project_description.json').write_text(
json.dumps({'monitor_baud': '115200', 'app_elf': 'hello_world.elf'}),
encoding='utf-8',
)
tools, _ = _start_server(mcp_ext, mock_mcp, str(proj))
with mock.patch('subprocess.run') as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr='')
tools['monitor_device'](commands='expect ready', baud='9600')
cmd = mock_run.call_args[0][0]
assert cmd[cmd.index('-b') + 1] == '9600'
assert '115200' not in cmd
# ---------------------------------------------------------------------------
# Test: server starts without error when project_path is not valid
# ---------------------------------------------------------------------------
@@ -517,6 +915,7 @@ class TestServerStartsOutsideProject:
assert 'set_target' in tools
assert 'flash_project' in tools
assert 'clean_project' in tools
assert 'monitor_device' in tools
# ---------------------------------------------------------------------------