mirror of
https://github.com/espressif/esp-idf.git
synced 2026-09-22 13:01:16 +03:00
feat(tools): mcp-server accepts project_dir from chat
At the same time, it's possible to start idf.py mcp-server outside ESP-IDF project directory. Added ESP-IDF mcp server tests
This commit is contained in:
@@ -208,6 +208,7 @@ test_tools:
|
||||
- run_cmd pytest --noconftest test_idf_py.py --junitxml=${IDF_PATH}/XUNIT_IDF_PY.xml --ignore-result-files ${KNOWN_FAILURE_CASES_FILE_NAME} || stat=1
|
||||
- run_cmd pytest --noconftest test_hints.py --junitxml=${IDF_PATH}/XUNIT_HINTS.xml --ignore-result-files ${KNOWN_FAILURE_CASES_FILE_NAME} || stat=1
|
||||
- run_cmd pytest --noconftest test_idf_qemu.py --junitxml=${IDF_PATH}/XUNIT_IDF_PY_QEMU.xml --ignore-result-files ${KNOWN_FAILURE_CASES_FILE_NAME} || stat=1
|
||||
- run_cmd pytest --noconftest test_mcp_ext.py --junitxml=${IDF_PATH}/XUNIT_MCP_EXT.xml --ignore-result-files ${KNOWN_FAILURE_CASES_FILE_NAME} || stat=1
|
||||
- cd ${IDF_PATH}/tools/test_bsasm
|
||||
- run_cmd pytest --noconftest test_bsasm.py --junitxml=${IDF_PATH}/XUNIT_BSASM.xml --ignore-result-files ${KNOWN_FAILURE_CASES_FILE_NAME} || stat=1
|
||||
- cd ${IDF_PATH}/tools/test_mkdfu
|
||||
|
||||
@@ -32,7 +32,7 @@ except ImportError:
|
||||
MCP_AVAILABLE = False
|
||||
|
||||
|
||||
def is_valid_project_dir(directory: str) -> bool:
|
||||
def _is_valid_project_dir(directory: str) -> bool:
|
||||
"""
|
||||
Determine if the given directory is a valid ESP-IDF project directory.
|
||||
- Must be a directory.
|
||||
@@ -58,6 +58,45 @@ def is_valid_project_dir(directory: str) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def resolve_default_project_dir(launch_dir: str) -> str | None:
|
||||
"""
|
||||
Returns the first valid ESP-IDF project directory from the server's launch
|
||||
context, or None if none is found.
|
||||
|
||||
Priority: IDF_MCP_WORKSPACE_FOLDER env var > launch_dir
|
||||
|
||||
Use this from contexts without an explicit ``project_dir`` argument
|
||||
"""
|
||||
for candidate in [os.environ.get('IDF_MCP_WORKSPACE_FOLDER', ''), launch_dir]:
|
||||
if candidate and _is_valid_project_dir(candidate):
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def resolve_tool_project_dir(explicit_dir: str | None, launch_dir: str) -> tuple[str | None, str | None]:
|
||||
"""
|
||||
Resolves the effective project directory for an MCP tool call.
|
||||
|
||||
Returns ``(effective_dir, None)`` on success, or ``(None, error_message)``
|
||||
on failure. When ``explicit_dir`` is provided it is validated immediately —
|
||||
the fallback chain is never tried for an explicit but invalid path.
|
||||
|
||||
Use this from MCP tools that accept a ``project_dir`` argument
|
||||
"""
|
||||
if explicit_dir is not None:
|
||||
if not _is_valid_project_dir(explicit_dir):
|
||||
return None, f'"{explicit_dir}" is not a valid ESP-IDF project directory.'
|
||||
return explicit_dir, None
|
||||
effective = resolve_default_project_dir(launch_dir)
|
||||
if effective is not None:
|
||||
return effective, None
|
||||
return None, (
|
||||
'No valid ESP-IDF project directory found. '
|
||||
'Pass project_dir explicitly, set IDF_MCP_WORKSPACE_FOLDER, '
|
||||
'or restart with: idf.py -C <project_dir> mcp-server'
|
||||
)
|
||||
|
||||
|
||||
def action_extensions(base_actions: dict, project_path: str) -> dict:
|
||||
"""ESP-IDF MCP Server Extension"""
|
||||
|
||||
@@ -72,50 +111,59 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
|
||||
'or use "idf.py docs" and search for EIM configuration instructions.'
|
||||
)
|
||||
|
||||
# Verify that mcp-server was executed from a valid ESP-IDF project directory.
|
||||
# This is necessary to obtain the correct context such as args, project path, etc.
|
||||
if not is_valid_project_dir(project_path):
|
||||
current_project = None
|
||||
for candidate in [os.getcwd(), os.environ.get('IDF_MCP_WORKSPACE_FOLDER', '')]:
|
||||
if is_valid_project_dir(candidate):
|
||||
current_project = candidate
|
||||
break
|
||||
if not current_project:
|
||||
raise FatalError('Open the MCP server in a valid ESP-IDF project directory.')
|
||||
# Resolve the default project directory. Then derive the startup log line, and the
|
||||
# bound_hint that is appended to every tool's description so the LLM driving
|
||||
# the MCP client knows when (not) to pass project_dir.
|
||||
startup_default_dir = resolve_default_project_dir(project_path)
|
||||
if startup_default_dir is not None:
|
||||
print(f'INFO: Starting ESP-IDF MCP Server. Default project: {startup_default_dir}', file=sys.stderr)
|
||||
bound_hint = (
|
||||
f"This MCP server was launched with '{startup_default_dir}' as the default ESP-IDF "
|
||||
'project. Leave project_dir as None to operate on this project. Only set project_dir '
|
||||
'(absolute path to a directory containing a CMakeLists.txt with project()) when the '
|
||||
'user explicitly asks to operate on a different ESP-IDF project.'
|
||||
)
|
||||
else:
|
||||
print(
|
||||
'INFO: Starting ESP-IDF MCP Server. No project directory configured at startup. '
|
||||
'Pass project_dir in each tool call, or set IDF_MCP_WORKSPACE_FOLDER, '
|
||||
'or restart with: idf.py -C <project_dir> mcp-server',
|
||||
file=sys.stderr,
|
||||
)
|
||||
bound_hint = (
|
||||
'This MCP server was launched without a project context. You MUST pass project_dir '
|
||||
'(absolute path to a directory containing a CMakeLists.txt with project()) on every '
|
||||
'call, otherwise the call will fail.'
|
||||
)
|
||||
|
||||
# Initialize MCP server — project validity is checked per-tool call
|
||||
mcp = FastMCP('ESP-IDF')
|
||||
|
||||
# === TOOLS (Actions) ===
|
||||
@mcp.tool(description=f'Build the ESP-IDF project (runs `idf.py build`). {bound_hint}')
|
||||
def build_project(project_dir: str | None = None) -> str:
|
||||
"""Build the ESP-IDF project.
|
||||
|
||||
Args:
|
||||
project_dir: Optional absolute path to a valid ESP-IDF project directory.
|
||||
Leave as None to use the project this MCP server was launched with
|
||||
(or the IDF_MCP_WORKSPACE_FOLDER environment variable). Set this
|
||||
only to override that default with another project.
|
||||
"""
|
||||
effective_dir, error = resolve_tool_project_dir(project_dir, project_path)
|
||||
if error:
|
||||
return error
|
||||
assert effective_dir is not None # mypy narrowing
|
||||
try:
|
||||
cmd = [
|
||||
sys.executable,
|
||||
os.path.join(os.environ['IDF_PATH'], 'tools', 'idf.py'),
|
||||
'-C',
|
||||
current_project,
|
||||
'mcp-server',
|
||||
]
|
||||
print(
|
||||
f'Starting ESP-IDF MCP Server with command: {" ".join(cmd)} in project path: {current_project}',
|
||||
file=sys.stderr,
|
||||
)
|
||||
subprocess.run(cmd, cwd=current_project, check=True)
|
||||
return
|
||||
except Exception as e:
|
||||
print(f'ERROR: Failed to start ESP-IDF MCP Server: {str(e)}', file=sys.stderr)
|
||||
raise FatalError(f'Failed to start ESP-IDF MCP Server: {str(e)}') from e
|
||||
|
||||
# Initialize MCP server
|
||||
mcp = FastMCP('ESP-IDF')
|
||||
|
||||
# === TOOLS (Actions) ===
|
||||
@mcp.tool()
|
||||
def build_project() -> str:
|
||||
"""Build ESP-IDF project"""
|
||||
try:
|
||||
cmd = [
|
||||
sys.executable,
|
||||
os.path.join(os.environ['IDF_PATH'], 'tools', 'idf.py'),
|
||||
effective_dir,
|
||||
'build',
|
||||
]
|
||||
# Information logs are shown in some mcp clients using stderr
|
||||
print(f'INFO: Building project with command: {" ".join(cmd)} in path: {project_path}', file=sys.stderr)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=project_path)
|
||||
print(f'INFO: Building project with command: {" ".join(cmd)} in path: {effective_dir}', file=sys.stderr)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if result.returncode == 0:
|
||||
print('INFO: Build successful', file=sys.stderr)
|
||||
return 'Successfully built project'
|
||||
@@ -126,18 +174,37 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
|
||||
print(f'ERROR: Build failed: {str(e)}', file=sys.stderr)
|
||||
return f'Build failed: {str(e)}'
|
||||
|
||||
@mcp.tool()
|
||||
def set_target(target: str) -> str:
|
||||
"""Set the ESP-IDF target (esp32, esp32s3, esp32c6, etc.)"""
|
||||
@mcp.tool(
|
||||
description=(
|
||||
'Set the ESP-IDF target chip (esp32, esp32s3, esp32c6, etc.) for the project '
|
||||
f'(runs `idf.py set-target`). {bound_hint}'
|
||||
)
|
||||
)
|
||||
def set_target(target: str, project_dir: str | None = None) -> str:
|
||||
"""Set the ESP-IDF target for the project.
|
||||
|
||||
Args:
|
||||
target: Target chip identifier (e.g. esp32, esp32s3, esp32c6).
|
||||
project_dir: Optional absolute path to a valid ESP-IDF project directory.
|
||||
Leave as None to use the project this MCP server was launched with
|
||||
(or the IDF_MCP_WORKSPACE_FOLDER environment variable). Set this
|
||||
only to override that default with another project.
|
||||
"""
|
||||
effective_dir, error = resolve_tool_project_dir(project_dir, project_path)
|
||||
if error:
|
||||
return error
|
||||
assert effective_dir is not None # mypy narrowing
|
||||
try:
|
||||
cmd = [
|
||||
sys.executable,
|
||||
os.path.join(os.environ['IDF_PATH'], 'tools', 'idf.py'),
|
||||
'-C',
|
||||
effective_dir,
|
||||
'set-target',
|
||||
target,
|
||||
]
|
||||
print(f'INFO: Setting target with command: {" ".join(cmd)} in path: {project_path}', file=sys.stderr)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=project_path)
|
||||
print(f'INFO: Setting target with command: {" ".join(cmd)} in path: {effective_dir}', file=sys.stderr)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if result.returncode == 0:
|
||||
print(f'INFO: Target set to: {target}', file=sys.stderr)
|
||||
return f'Target set to: {target}'
|
||||
@@ -148,9 +215,24 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
|
||||
print(f'ERROR: Failed to set target: {str(e)}', file=sys.stderr)
|
||||
return f'Error setting target: {str(e)}'
|
||||
|
||||
@mcp.tool()
|
||||
def flash_project(port: str | None = None) -> str:
|
||||
"""Flash the built project to connected device"""
|
||||
@mcp.tool(
|
||||
description=(f'Flash the built ESP-IDF project to a connected device (runs `idf.py flash`). {bound_hint}')
|
||||
)
|
||||
def flash_project(port: str | None = None, project_dir: str | None = None) -> str:
|
||||
"""Flash the built ESP-IDF project to a connected device.
|
||||
|
||||
Args:
|
||||
port: Optional serial port to flash through (e.g. /dev/ttyUSB0, COM3).
|
||||
Leave as None to let idf.py auto-detect.
|
||||
project_dir: Optional absolute path to a valid ESP-IDF project directory.
|
||||
Leave as None to use the project this MCP server was launched with
|
||||
(or the IDF_MCP_WORKSPACE_FOLDER environment variable). Set this
|
||||
only to override that default with another project.
|
||||
"""
|
||||
effective_dir, error = resolve_tool_project_dir(project_dir, project_path)
|
||||
if error:
|
||||
return error
|
||||
assert effective_dir is not None # mypy narrowing
|
||||
try:
|
||||
flash_args = []
|
||||
if port:
|
||||
@@ -160,9 +242,11 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
|
||||
cmd = [
|
||||
sys.executable,
|
||||
os.path.join(os.environ['IDF_PATH'], 'tools', 'idf.py'),
|
||||
'-C',
|
||||
effective_dir,
|
||||
] + flash_args
|
||||
print(f'INFO: Flashing project with command: {" ".join(cmd)} in path: {project_path}', file=sys.stderr)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=project_path)
|
||||
print(f'INFO: Flashing project with command: {" ".join(cmd)} in path: {effective_dir}', file=sys.stderr)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
|
||||
if result.returncode == 0:
|
||||
print('INFO: Flash successful', file=sys.stderr)
|
||||
@@ -174,17 +258,30 @@ 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()
|
||||
def clean_project() -> str:
|
||||
"""Clean build artifacts"""
|
||||
@mcp.tool(description=f'Remove build artifacts from the ESP-IDF project (runs `idf.py clean`). {bound_hint}')
|
||||
def clean_project(project_dir: str | None = None) -> str:
|
||||
"""Remove build artifacts from the ESP-IDF project.
|
||||
|
||||
Args:
|
||||
project_dir: Optional absolute path to a valid ESP-IDF project directory.
|
||||
Leave as None to use the project this MCP server was launched with
|
||||
(or the IDF_MCP_WORKSPACE_FOLDER environment variable). Set this
|
||||
only to override that default with another project.
|
||||
"""
|
||||
effective_dir, error = resolve_tool_project_dir(project_dir, project_path)
|
||||
if error:
|
||||
return error
|
||||
assert effective_dir is not None # mypy narrowing
|
||||
try:
|
||||
cmd = [
|
||||
sys.executable,
|
||||
os.path.join(os.environ['IDF_PATH'], 'tools', 'idf.py'),
|
||||
'-C',
|
||||
effective_dir,
|
||||
'clean',
|
||||
]
|
||||
print(f'INFO: Cleaning project with command: {" ".join(cmd)} in path: {project_path}', file=sys.stderr)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=project_path)
|
||||
print(f'INFO: Cleaning project with command: {" ".join(cmd)} in path: {effective_dir}', file=sys.stderr)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if result.returncode == 0:
|
||||
print('INFO: Project cleaned successfully', file=sys.stderr)
|
||||
return 'Project cleaned successfully'
|
||||
@@ -199,15 +296,28 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
|
||||
@mcp.resource('project://config')
|
||||
def get_project_config() -> str:
|
||||
"""Get current project configuration"""
|
||||
build_dir = args.get('build_dir', '')
|
||||
effective_dir = resolve_default_project_dir(project_path)
|
||||
config: dict[str, Any] = {}
|
||||
|
||||
if effective_dir is None:
|
||||
config['error'] = (
|
||||
'No valid ESP-IDF project directory found. '
|
||||
'Set IDF_MCP_WORKSPACE_FOLDER or restart with: idf.py -C <project_dir> mcp-server'
|
||||
)
|
||||
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):
|
||||
config['build_dir_exists'] = False
|
||||
return json.dumps(config, indent=2)
|
||||
|
||||
config['build_dir'] = build_dir
|
||||
proj_desc_fn = f'{build_dir}/project_description.json'
|
||||
proj_desc_fn = os.path.join(build_dir, 'project_description.json')
|
||||
config['project_description'] = 'Project description does not exist'
|
||||
|
||||
try:
|
||||
@@ -221,15 +331,26 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
|
||||
@mcp.resource('project://status')
|
||||
def get_project_status() -> str:
|
||||
"""Get current project build status"""
|
||||
status: dict[str, Any] = {}
|
||||
try:
|
||||
status = {
|
||||
'project_path': project_path,
|
||||
'target': get_target(project_path),
|
||||
'idf_version': idf_version(),
|
||||
}
|
||||
effective_dir = resolve_default_project_dir(project_path)
|
||||
|
||||
# Check if built
|
||||
build_dir = args.build_dir
|
||||
if effective_dir is None:
|
||||
status['error'] = (
|
||||
'No valid ESP-IDF project directory found. '
|
||||
'Set IDF_MCP_WORKSPACE_FOLDER or restart with: idf.py -C <project_dir> mcp-server'
|
||||
)
|
||||
status['idf_version'] = idf_version()
|
||||
return json.dumps(status, indent=2)
|
||||
|
||||
status['project_path'] = effective_dir
|
||||
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')
|
||||
)
|
||||
if os.path.exists(build_dir):
|
||||
status['build_dir'] = build_dir
|
||||
artifacts = ['bootloader', 'partition_table', 'app-flash', 'flash_args']
|
||||
@@ -242,7 +363,8 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
|
||||
|
||||
return json.dumps(status, indent=2)
|
||||
except Exception as e:
|
||||
return f'Error getting status: {str(e)}'
|
||||
status['error'] = f'Error getting status: {str(e)}'
|
||||
return json.dumps(status, indent=2)
|
||||
|
||||
@mcp.resource('project://devices')
|
||||
def get_connected_devices() -> str:
|
||||
|
||||
504
tools/test_idf_py/test_mcp_ext.py
Normal file
504
tools/test_idf_py/test_mcp_ext.py
Normal file
@@ -0,0 +1,504 @@
|
||||
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
"""Unit tests for idf_py_actions/mcp_ext.py.
|
||||
|
||||
No full IDF environment is required — all external imports and subprocess
|
||||
calls are mocked.
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import types
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers for creating fake project directories
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
IDF_CMAKE_LINE = r'include($ENV{IDF_PATH}/tools/cmake/project.cmake)'
|
||||
|
||||
|
||||
def _make_valid_project(path: Path) -> Path:
|
||||
"""Write a minimal valid ESP-IDF CMakeLists.txt into *path*."""
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
(path / 'CMakeLists.txt').write_text(
|
||||
f'cmake_minimum_required(VERSION 3.16)\n{IDF_CMAKE_LINE}\nproject(hello_world)\n',
|
||||
encoding='utf-8',
|
||||
)
|
||||
return path
|
||||
|
||||
|
||||
def _make_invalid_project(path: Path) -> Path:
|
||||
"""Write a CMakeLists.txt that does NOT include the IDF line."""
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
(path / 'CMakeLists.txt').write_text(
|
||||
'cmake_minimum_required(VERSION 3.16)\nproject(plain_cmake)\n',
|
||||
encoding='utf-8',
|
||||
)
|
||||
return path
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixture: stub out all non-stdlib imports so the module can be loaded
|
||||
# without an IDF installation.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _MockFastMCP:
|
||||
"""Captures tool/resource registrations so tests can invoke them."""
|
||||
|
||||
def __init__(self, name: str) -> None:
|
||||
self.name = name
|
||||
self.tools: dict[str, Callable[..., Any]] = {}
|
||||
self.resources: dict[str, Callable[..., Any]] = {}
|
||||
|
||||
def tool(self, **kwargs: Any) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
|
||||
def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
|
||||
self.tools[fn.__name__] = fn
|
||||
return fn
|
||||
|
||||
return decorator
|
||||
|
||||
def resource(self, uri: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
|
||||
def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
|
||||
self.resources[uri] = fn
|
||||
return fn
|
||||
|
||||
return decorator
|
||||
|
||||
def run(self) -> None:
|
||||
pass # don't block in tests
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mcp_ext(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[types.ModuleType, _MockFastMCP]:
|
||||
"""
|
||||
Import (or reimport) mcp_ext with all external dependencies mocked.
|
||||
Returns the module object plus a _MockFastMCP instance that was used so
|
||||
that tests can inspect registered tools/resources.
|
||||
"""
|
||||
# Build a fresh _MockFastMCP for this test
|
||||
mock_mcp_instance = _MockFastMCP('ESP-IDF')
|
||||
|
||||
# Stub rich_click
|
||||
rich_click = types.ModuleType('rich_click')
|
||||
rich_click.Context = object # type: ignore[attr-defined]
|
||||
|
||||
# Stub idf_py_actions hierarchy
|
||||
idf_py_actions_pkg = types.ModuleType('idf_py_actions')
|
||||
errors_mod = types.ModuleType('idf_py_actions.errors')
|
||||
|
||||
class FatalError(Exception):
|
||||
pass
|
||||
|
||||
errors_mod.FatalError = FatalError # type: ignore[attr-defined]
|
||||
|
||||
tools_mod = types.ModuleType('idf_py_actions.tools')
|
||||
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]
|
||||
|
||||
idf_py_actions_pkg.errors = errors_mod # type: ignore[attr-defined]
|
||||
idf_py_actions_pkg.tools = tools_mod # type: ignore[attr-defined]
|
||||
|
||||
# Stub mcp.server.fastmcp — FastMCP constructor returns our mock
|
||||
mcp_pkg = types.ModuleType('mcp')
|
||||
mcp_server_pkg = types.ModuleType('mcp.server')
|
||||
fastmcp_mod = types.ModuleType('mcp.server.fastmcp')
|
||||
fastmcp_mod.FastMCP = lambda name: mock_mcp_instance # type: ignore[attr-defined]
|
||||
|
||||
stubs = {
|
||||
'rich_click': rich_click,
|
||||
'idf_py_actions': idf_py_actions_pkg,
|
||||
'idf_py_actions.errors': errors_mod,
|
||||
'idf_py_actions.tools': tools_mod,
|
||||
'mcp': mcp_pkg,
|
||||
'mcp.server': mcp_server_pkg,
|
||||
'mcp.server.fastmcp': fastmcp_mod,
|
||||
}
|
||||
for name, stub_mod in stubs.items():
|
||||
monkeypatch.setitem(sys.modules, name, stub_mod)
|
||||
|
||||
# Load mcp_ext directly from its file so that the stub 'idf_py_actions'
|
||||
# package (which is not a real package) doesn't prevent import.
|
||||
mcp_ext_path = Path(__file__).parent.parent / 'idf_py_actions' / 'mcp_ext.py'
|
||||
spec = importlib.util.spec_from_file_location('idf_py_actions.mcp_ext', mcp_ext_path)
|
||||
mod = importlib.util.module_from_spec(spec) # type: ignore[arg-type]
|
||||
# Register under both names so cross-references inside the module work
|
||||
monkeypatch.setitem(sys.modules, 'idf_py_actions.mcp_ext', mod)
|
||||
spec.loader.exec_module(mod) # type: ignore[union-attr]
|
||||
|
||||
return mod, mock_mcp_instance
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper: call start_mcp_server and return registered tools/resources
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _start_server(
|
||||
mcp_ext_module: tuple[types.ModuleType, _MockFastMCP],
|
||||
mock_mcp_instance: _MockFastMCP,
|
||||
project_path: str,
|
||||
) -> tuple[dict[str, Callable[..., Any]], dict[str, Callable[..., Any]]]:
|
||||
"""
|
||||
Call action_extensions / start_mcp_server with *project_path* so that
|
||||
tools and resources are registered on *mock_mcp_instance*.
|
||||
"""
|
||||
mod, _ = mcp_ext_module
|
||||
ext = mod.action_extensions({}, project_path)
|
||||
callback = ext['actions']['mcp-server']['callback']
|
||||
# ctx and args are only used inside resources; use simple stubs
|
||||
fake_args = {'build_dir': os.path.join(project_path, 'build')}
|
||||
callback('mcp-server', ctx=None, args=fake_args)
|
||||
return mock_mcp_instance.tools, mock_mcp_instance.resources
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: _is_valid_project_dir
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestIsValidProjectDir:
|
||||
def test_valid_project(self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP]) -> None:
|
||||
mod, _ = mcp_ext
|
||||
proj = _make_valid_project(tmp_path / 'my_proj')
|
||||
assert mod._is_valid_project_dir(str(proj)) is True
|
||||
|
||||
def test_missing_directory(self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP]) -> None:
|
||||
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, _MockFastMCP]) -> 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, _MockFastMCP]) -> None:
|
||||
mod, _ = mcp_ext
|
||||
proj = _make_invalid_project(tmp_path / 'plain')
|
||||
assert mod._is_valid_project_dir(str(proj)) is False
|
||||
|
||||
def test_empty_string(self, mcp_ext: tuple[types.ModuleType, _MockFastMCP]) -> None:
|
||||
mod, _ = mcp_ext
|
||||
assert mod._is_valid_project_dir('') is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: resolve_default_project_dir
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestResolveDefaultProjectDir:
|
||||
def test_env_var_takes_priority_over_default(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, _ = mcp_ext
|
||||
env_proj = _make_valid_project(tmp_path / 'env_proj')
|
||||
default = _make_valid_project(tmp_path / 'default')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', str(env_proj))
|
||||
result = mod.resolve_default_project_dir(str(default))
|
||||
assert result == str(env_proj)
|
||||
|
||||
def test_default_used_when_env_not_set(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, _ = mcp_ext
|
||||
default = _make_valid_project(tmp_path / 'default')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
result = mod.resolve_default_project_dir(str(default))
|
||||
assert result == str(default)
|
||||
|
||||
def test_returns_none_when_nothing_valid(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, _ = mcp_ext
|
||||
invalid = _make_invalid_project(tmp_path / 'bad')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
result = mod.resolve_default_project_dir(str(invalid))
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: tools
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildProject:
|
||||
def test_returns_error_when_no_valid_dir(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
invalid = _make_invalid_project(tmp_path / 'bad')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
tools, _ = _start_server(mcp_ext, mock_mcp, str(invalid))
|
||||
result = tools['build_project'](project_dir=None)
|
||||
assert 'No valid ESP-IDF project directory found' in result
|
||||
|
||||
def test_explicit_invalid_dir_returns_error_not_fallback(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
bad = _make_invalid_project(tmp_path / 'bad')
|
||||
good_env = _make_valid_project(tmp_path / 'good_env')
|
||||
# Even though IDF_MCP_WORKSPACE_FOLDER is a valid project, passing an
|
||||
# explicit but invalid project_dir must return an error, not silently
|
||||
# fall through to the env var project.
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', str(good_env))
|
||||
tools, _ = _start_server(mcp_ext, mock_mcp, str(bad))
|
||||
result = tools['build_project'](project_dir=str(bad))
|
||||
assert str(bad) in result
|
||||
assert 'not a valid' in result
|
||||
|
||||
def test_explicit_dir_builds_in_correct_location(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
proj = _make_valid_project(tmp_path / 'proj')
|
||||
invalid = _make_invalid_project(tmp_path / 'bad')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
monkeypatch.setenv('IDF_PATH', str(tmp_path))
|
||||
tools, _ = _start_server(mcp_ext, mock_mcp, str(invalid))
|
||||
|
||||
with mock.patch('subprocess.run') as mock_run:
|
||||
mock_run.return_value = mock.Mock(returncode=0, stderr='')
|
||||
result = tools['build_project'](project_dir=str(proj))
|
||||
|
||||
assert result == 'Successfully built project'
|
||||
call_args = mock_run.call_args
|
||||
cmd = call_args[0][0]
|
||||
assert '-C' in cmd
|
||||
assert str(proj) in cmd
|
||||
assert cmd[cmd.index('-C') + 1] == str(proj)
|
||||
# cwd is intentionally not passed — -C is authoritative for idf.py
|
||||
assert call_args[1].get('cwd') is None
|
||||
|
||||
def test_build_failure_returns_error(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
proj = _make_valid_project(tmp_path / 'proj')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
monkeypatch.setenv('IDF_PATH', str(tmp_path))
|
||||
tools, _ = _start_server(mcp_ext, mock_mcp, str(proj))
|
||||
|
||||
with mock.patch('subprocess.run') as mock_run:
|
||||
mock_run.return_value = mock.Mock(returncode=1, stderr='cmake error')
|
||||
result = tools['build_project']()
|
||||
|
||||
assert 'Build failed' in result
|
||||
assert 'cmake error' in result
|
||||
|
||||
|
||||
class TestSetTarget:
|
||||
def test_returns_error_when_no_valid_dir(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
invalid = _make_invalid_project(tmp_path / 'bad')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
tools, _ = _start_server(mcp_ext, mock_mcp, str(invalid))
|
||||
result = tools['set_target']('esp32c6')
|
||||
assert 'No valid ESP-IDF project directory found' in result
|
||||
|
||||
def test_explicit_dir_sets_target(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
proj = _make_valid_project(tmp_path / 'proj')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
monkeypatch.setenv('IDF_PATH', str(tmp_path))
|
||||
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, stderr='')
|
||||
result = tools['set_target']('esp32s3', project_dir=str(proj))
|
||||
|
||||
assert result == 'Target set to: esp32s3'
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert 'set-target' in cmd
|
||||
assert 'esp32s3' in cmd
|
||||
assert '-C' in cmd
|
||||
assert cmd[cmd.index('-C') + 1] == str(proj)
|
||||
|
||||
|
||||
class TestFlashProject:
|
||||
def test_returns_error_when_no_valid_dir(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
invalid = _make_invalid_project(tmp_path / 'bad')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
tools, _ = _start_server(mcp_ext, mock_mcp, str(invalid))
|
||||
result = tools['flash_project']()
|
||||
assert 'No valid ESP-IDF project directory found' in result
|
||||
|
||||
def test_port_and_dir_forwarded(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
proj = _make_valid_project(tmp_path / 'proj')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
monkeypatch.setenv('IDF_PATH', str(tmp_path))
|
||||
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, stderr='')
|
||||
result = tools['flash_project'](port='/dev/ttyUSB0', project_dir=str(proj))
|
||||
|
||||
assert 'Successfully flashed' in result
|
||||
assert '/dev/ttyUSB0' in result
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert '-p' in cmd
|
||||
assert '/dev/ttyUSB0' in cmd
|
||||
assert '-C' in cmd
|
||||
assert cmd[cmd.index('-C') + 1] == str(proj)
|
||||
|
||||
|
||||
class TestCleanProject:
|
||||
def test_returns_error_when_no_valid_dir(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
invalid = _make_invalid_project(tmp_path / 'bad')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
tools, _ = _start_server(mcp_ext, mock_mcp, str(invalid))
|
||||
result = tools['clean_project']()
|
||||
assert 'No valid ESP-IDF project directory found' in result
|
||||
|
||||
def test_explicit_dir_cleans(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
proj = _make_valid_project(tmp_path / 'proj')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
monkeypatch.setenv('IDF_PATH', str(tmp_path))
|
||||
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, stderr='')
|
||||
result = tools['clean_project'](project_dir=str(proj))
|
||||
|
||||
assert result == 'Project cleaned successfully'
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert 'clean' in cmd
|
||||
assert '-C' in cmd
|
||||
assert cmd[cmd.index('-C') + 1] == str(proj)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: server starts without error when project_path is not valid
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestServerStartsOutsideProject:
|
||||
def test_no_fatal_error_when_project_path_invalid(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""start_mcp_server must not raise when project_path is not a valid project."""
|
||||
mod, mock_mcp = mcp_ext
|
||||
invalid = _make_invalid_project(tmp_path / 'not_a_project')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
|
||||
ext = mod.action_extensions({}, str(invalid))
|
||||
callback = ext['actions']['mcp-server']['callback']
|
||||
fake_args = {'build_dir': str(invalid / 'build')}
|
||||
|
||||
# Should not raise
|
||||
callback('mcp-server', ctx=None, args=fake_args)
|
||||
|
||||
# FastMCP was still initialised
|
||||
assert mock_mcp.name == 'ESP-IDF'
|
||||
|
||||
def test_tools_registered_even_when_project_path_invalid(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
invalid = _make_invalid_project(tmp_path / 'not_a_project')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
|
||||
tools, _ = _start_server(mcp_ext, mock_mcp, str(invalid))
|
||||
assert 'build_project' in tools
|
||||
assert 'set_target' in tools
|
||||
assert 'flash_project' in tools
|
||||
assert 'clean_project' in tools
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: resources use resolve_default_project_dir
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetProjectStatus:
|
||||
def test_returns_error_json_when_no_valid_dir(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
invalid = _make_invalid_project(tmp_path / 'bad')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
_, resources = _start_server(mcp_ext, mock_mcp, str(invalid))
|
||||
|
||||
result = json.loads(resources['project://status']())
|
||||
assert 'error' in result
|
||||
assert 'idf_version' in result
|
||||
|
||||
def test_returns_status_when_valid_project(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
proj = _make_valid_project(tmp_path / 'proj')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
_, resources = _start_server(mcp_ext, mock_mcp, str(proj))
|
||||
|
||||
result = json.loads(resources['project://status']())
|
||||
assert result['project_path'] == str(proj)
|
||||
assert result['target'] == 'esp32'
|
||||
assert 'error' not in result
|
||||
|
||||
def test_uses_env_var_when_project_path_invalid(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
invalid = _make_invalid_project(tmp_path / 'bad')
|
||||
env_proj = _make_valid_project(tmp_path / 'env_proj')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', str(env_proj))
|
||||
_, resources = _start_server(mcp_ext, mock_mcp, str(invalid))
|
||||
|
||||
result = json.loads(resources['project://status']())
|
||||
assert result['project_path'] == str(env_proj)
|
||||
assert 'error' not in result
|
||||
|
||||
|
||||
class TestGetProjectConfig:
|
||||
def test_returns_error_json_when_no_valid_dir(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
invalid = _make_invalid_project(tmp_path / 'bad')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
_, resources = _start_server(mcp_ext, mock_mcp, str(invalid))
|
||||
|
||||
result = json.loads(resources['project://config']())
|
||||
assert 'error' in result
|
||||
|
||||
def test_returns_no_build_dir_when_build_missing(
|
||||
self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockFastMCP], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mod, mock_mcp = mcp_ext
|
||||
proj = _make_valid_project(tmp_path / 'proj')
|
||||
monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '')
|
||||
_, resources = _start_server(mcp_ext, mock_mcp, str(proj))
|
||||
|
||||
# build/ does not exist → build_dir_exists: False
|
||||
result = json.loads(resources['project://config']())
|
||||
assert result.get('build_dir_exists') is False
|
||||
Reference in New Issue
Block a user