MCP/mcesptool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Espressif product catalog integration

Browse Source 
5 new MCP tools for chip/module discovery, comparison, and procurement
planning using Espressif's public product catalog API. Data is lazily
fetched and cached in-memory with a 24h TTL.

Tools: esp_product_search, esp_product_info, esp_chip_compare,
esp_product_recommend, esp_product_availability
Resource: esp://products/catalog
This commit is contained in:
Ryan Malloy 2026-02-24 13:33:34 -07:00
parent 58f9b3a4df
commit dab9c6848e
6 changed files with 1151 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pyproject.toml
View File

@ -35,6 +35,7 @@ dependencies = [
"pyserial>=3.5", # Serial communication
"pyserial-asyncio>=0.6", # Async serial support
"thefuzz[speedup]>=0.22.1", # Fuzzy string matching
"httpx>=0.28", # Async HTTP client (product catalog)
"pydantic>=2.11.7", # Data validation
"click>=8.1.0", # CLI framework
"rich>=13.9.4", # Rich console output

3
src/mcesptool/components/__init__.py
View File

@ -11,6 +11,7 @@ from .firmware_builder import FirmwareBuilder
from .flash_manager import FlashManager
from .ota_manager import OTAManager
from .partition_manager import PartitionManager
from .product_catalog import ProductCatalog
from .production_tools import ProductionTools
from .qemu_manager import QemuManager
from .security_manager import SecurityManager
@ -26,6 +27,7 @@ COMPONENT_REGISTRY = {
"production_tools": ProductionTools,
"diagnostics": Diagnostics,
"qemu_manager": QemuManager,
"product_catalog": ProductCatalog,
}
__all__ = [
@ -38,5 +40,6 @@ __all__ = [
"ProductionTools",
"Diagnostics",
"QemuManager",
"ProductCatalog",
"COMPONENT_REGISTRY",
]

619
src/mcesptool/components/product_catalog.py Normal file
View File

@ -0,0 +1,619 @@
"""
Product Catalog Component
Provides access to Espressif's public product catalog API for chip/module
discovery, comparison, and procurement planning. Data is lazily fetched
and cached in-memory with a 24-hour TTL.
"""
import json
import logging
import re
import time
from typing import Any
import httpx
from fastmcp import Context, FastMCP
from ..config import ESPToolServerConfig
logger = logging.getLogger(__name__)
CATALOG_URL = "https://products.espressif.com/api/user/products?language=en"
def _parse_memory_field(value: str | int | None) -> tuple[int, str]:
"""Parse compound memory strings like '4, Quad' into (size_mb, interface).
Returns (0, '') for empty/unparsable values. Handles bare ints from the API.
"""
if isinstance(value, int):
return value, ""
if not value or value in ("N/A", "NA", "-", "—"):
return 0, ""
parts = [p.strip() for p in value.split(",")]
try:
size = int(parts[0])
except (ValueError, IndexError):
return 0, ""
interface = parts[1] if len(parts) > 1 else ""
return size, interface
def _parse_sram_kb(value: str | int | None) -> int:
"""Parse SRAM field — may be plain int or compound string."""
if isinstance(value, int):
return value
if not value or value in ("N/A", "NA", "-", "—"):
return 0
parts = value.split(",")
try:
return int(parts[0].strip())
except (ValueError, IndexError):
return 0
def _has_capability(value: str) -> bool:
"""Check if a feature field indicates availability."""
if not value:
return False
v = value.strip().lower()
return v not in ("", "n/a", "na", "-", "—", "no", "0")
def _normalize_status(value: str) -> str:
"""Normalize status strings for consistent filtering."""
mapping = {
"mass production": "Mass Production",
"nrnd": "NRND",
"eol": "EOL",
"replaced": "Replaced",
"sample": "Sample",
}
return mapping.get(value.strip().lower(), value.strip())
def _product_summary(product: dict) -> dict[str, Any]:
"""Extract a concise summary of a product for search results."""
flash_mb, flash_type = _parse_memory_field(product.get("flash", ""))
psram_mb, psram_type = _parse_memory_field(product.get("psram", ""))
return {
"name": product.get("name", ""),
"type": product.get("type", ""),
"series": product.get("seriesName", ""),
"status": _normalize_status(product.get("status", "")),
"wifi": _has_capability(product.get("wifi", "")),
"bluetooth": _has_capability(product.get("bluetooth", "")),
"thread_zigbee": _has_capability(product.get("threadZigbee", "")),
"flash_mb": flash_mb,
"psram_mb": psram_mb,
"sram_kb": _parse_sram_kb(product.get("sram", "")),
"gpio": int(product.get("gpio", 0) or 0),
"antenna": product.get("antenna", ""),
}
def _product_detail(product: dict) -> dict[str, Any]:
"""Full formatted product record."""
flash_mb, flash_type = _parse_memory_field(product.get("flash", ""))
psram_mb, psram_type = _parse_memory_field(product.get("psram", ""))
return {
"name": product.get("name", ""),
"name_new": product.get("nameNew", ""),
"type": product.get("type", ""),
"series": product.get("seriesName", ""),
"status": _normalize_status(product.get("status", "")),
"mpn": product.get("mpn", ""),
"dimensions": product.get("dimensions", ""),
"wifi": product.get("wifi", ""),
"wifi6": product.get("wifi6", ""),
"bluetooth": product.get("bluetooth", ""),
"thread_zigbee": product.get("threadZigbee", ""),
"pins": product.get("pins", ""),
"frequency_mhz": product.get("freq", ""),
"sram_kb": _parse_sram_kb(product.get("sram", "")),
"rom_kb": product.get("rom", ""),
"flash_mb": flash_mb,
"flash_type": flash_type,
"psram_mb": psram_mb,
"psram_type": psram_type,
"gpio": int(product.get("gpio", 0) or 0),
"operating_temp": product.get("operatingTemp", ""),
"voltage_range": product.get("voltageRange", ""),
"antenna": product.get("antenna", ""),
"size_type": product.get("sizeType", ""),
"release_time": product.get("releaseTime", ""),
"idf_supports": product.get("idfSupports", ""),
"spq": product.get("spq", ""),
"moq": product.get("moq", ""),
"lead_time": product.get("leadTime", ""),
"eccn_code": product.get("eccnCode", ""),
"ccatsz_code": product.get("ccatszCode", ""),
"hs_code": product.get("hsCode", ""),
"pre_firmware": product.get("preFirmware", ""),
"replaced_by": product.get("replacedByName", ""),
}
def _parse_lead_weeks(lead_time: str) -> int | None:
"""Extract numeric weeks from lead time strings like '8-10 weeks'."""
if not lead_time:
return None
m = re.search(r"(\d+)", lead_time)
return int(m.group(1)) if m else None
def _parse_temp_max(temp_str: str) -> int | None:
"""Extract max operating temperature from strings like '-40~85C'."""
if not temp_str:
return None
m = re.search(r"~\s*(\d+)", temp_str)
if m:
return int(m.group(1))
m = re.search(r"(\d+)\s*[Cc]", temp_str)
return int(m.group(1)) if m else None
class ProductCatalog:
"""Espressif product catalog — chip/module discovery, comparison, and procurement"""
def __init__(self, app: FastMCP, config: ESPToolServerConfig) -> None:
self.app = app
self.config = config
self._products: list[dict] = []
self._last_fetch: float = 0
self._cache_ttl: int = 86400 # 24 hours
self._register_tools()
self._register_resources()
async def _ensure_data(self, force_refresh: bool = False) -> list[dict]:
"""Lazy-load and cache the product catalog from Espressif's API."""
now = time.time()
if self._products and not force_refresh and (now - self._last_fetch) < self._cache_ttl:
return self._products
logger.info("Fetching Espressif product catalog...")
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.get(CATALOG_URL)
resp.raise_for_status()
data = resp.json()
self._products = data.get("results", [])
self._last_fetch = time.time()
logger.info("Cached %d products from Espressif catalog", len(self._products))
return self._products
def _fuzzy_find(self, products: list[dict], query: str, threshold: int = 60) -> list[dict]:
"""Fuzzy-match products by name or MPN."""
from thefuzz import fuzz
scored = []
q = query.lower()
for p in products:
name = (p.get("name") or "").lower()
mpn = (p.get("mpn") or "").lower()
if name == q or mpn == q:
scored.append((300, p))
elif q in name or q in mpn:
scored.append((200, p))
elif name in q or mpn in q:
scored.append((150, p))
else:
score = max(fuzz.partial_ratio(q, name), fuzz.partial_ratio(q, mpn))
if score >= threshold:
scored.append((score, p))
scored.sort(key=lambda x: x[0], reverse=True)
return [p for _, p in scored]
def _register_resources(self) -> None:
"""Register MCP resources."""
@self.app.resource("esp://products/catalog")
async def product_catalog_resource() -> str:
"""Complete Espressif product catalog — SoCs and modules with full specs"""
products = await self._ensure_data()
return json.dumps([_product_summary(p) for p in products])
def _register_tools(self) -> None:
"""Register product catalog tools with FastMCP."""
@self.app.tool("esp_product_search")
async def product_search(
context: Context,
series: str | None = None,
product_type: str | None = None,
wifi: bool | None = None,
bluetooth: bool | None = None,
thread_zigbee: bool | None = None,
min_flash_mb: int | None = None,
min_psram_mb: int | None = None,
min_sram_kb: int | None = None,
min_gpio: int | None = None,
max_temp_c: int | None = None,
status: str | None = None,
antenna: str | None = None,
keyword: str | None = None,
) -> dict[str, Any]:
"""Search the Espressif product catalog by specs, capabilities, or keyword.
Filter by chip family, connectivity, memory, GPIO count, temperature
range, production status, antenna type, or free-text keyword. All
filters are optional and combine with AND logic. Returns matching
products with key specs.
Args:
series: Chip family filter, e.g. "ESP32-S3", "ESP32-C6"
product_type: "SoC" or "Module"
wifi: Filter to products with Wi-Fi
bluetooth: Filter to products with Bluetooth
thread_zigbee: Filter to products with Thread/Zigbee (802.15.4)
min_flash_mb: Minimum flash size in MB
min_psram_mb: Minimum PSRAM size in MB
min_sram_kb: Minimum SRAM size in KB
min_gpio: Minimum GPIO pin count
max_temp_c: Required upper operating temperature (e.g. 105 for industrial)
status: Production status — "Mass Production", "NRND", "EOL", "Sample"
antenna: Antenna type — "PCB", "IPEX", etc.
keyword: Fuzzy search on product name or MPN
"""
return await self._product_search_impl(
context,
series=series,
product_type=product_type,
wifi=wifi,
bluetooth=bluetooth,
thread_zigbee=thread_zigbee,
min_flash_mb=min_flash_mb,
min_psram_mb=min_psram_mb,
min_sram_kb=min_sram_kb,
min_gpio=min_gpio,
max_temp_c=max_temp_c,
status=status,
antenna=antenna,
keyword=keyword,
)
@self.app.tool("esp_product_info")
async def product_info(
context: Context,
name: str,
) -> dict[str, Any]:
"""Get detailed specifications for a specific Espressif product.
Looks up a product by name or MPN (fuzzy-matched) and returns
the full spec sheet: memory, connectivity, GPIO, temperature
range, procurement details, ECCN/HS codes, and more.
Args:
name: Product name or MPN, e.g. "ESP32-S3-WROOM-1" (fuzzy-matched)
"""
return await self._product_info_impl(context, name)
@self.app.tool("esp_chip_compare")
async def chip_compare(
context: Context,
products: list[str],
) -> dict[str, Any]:
"""Compare 2-4 Espressif products side-by-side.
Produces a comparison table highlighting differences in connectivity,
memory, GPIO, temperature range, and procurement status. Useful for
deciding between chip families or module variants.
Args:
products: List of 2-4 product names to compare (fuzzy-matched)
"""
return await self._chip_compare_impl(context, products)
@self.app.tool("esp_product_recommend")
async def product_recommend(
context: Context,
use_case: str,
constraints: str | None = None,
prefer_module: bool = True,
) -> dict[str, Any]:
"""Get chip/module recommendations for a use case.
Describe what you're building and any constraints, and this tool
filters the Espressif catalog to matching products ranked by fit.
Only returns products in Mass Production by default.
Args:
use_case: What you're building, e.g. "battery-powered BLE sensor"
constraints: Technical constraints, e.g. "needs PSRAM, temp to 105C"
prefer_module: Prefer modules over bare SoCs (default: true)
"""
return await self._product_recommend_impl(context, use_case, constraints, prefer_module)
@self.app.tool("esp_product_availability")
async def product_availability(
context: Context,
series: str | None = None,
status: str = "Mass Production",
max_lead_weeks: int | None = None,
max_moq: int | None = None,
) -> dict[str, Any]:
"""Check product availability and procurement details.
Filter the catalog by production status, lead time, and MOQ
to find products you can actually source. Useful for procurement
planning and identifying alternatives for EOL/NRND parts.
Args:
series: Chip family filter, e.g. "ESP32-S3"
status: Production status filter (default: "Mass Production")
max_lead_weeks: Maximum acceptable lead time in weeks
max_moq: Maximum acceptable minimum order quantity
"""
return await self._product_availability_impl(
context, series, status, max_lead_weeks, max_moq
)
# ── Tool implementations ────────────────────────────────────────────
async def _product_search_impl(
self,
context: Context,
series: str | None = None,
product_type: str | None = None,
wifi: bool | None = None,
bluetooth: bool | None = None,
thread_zigbee: bool | None = None,
min_flash_mb: int | None = None,
min_psram_mb: int | None = None,
min_sram_kb: int | None = None,
min_gpio: int | None = None,
max_temp_c: int | None = None,
status: str | None = None,
antenna: str | None = None,
keyword: str | None = None,
) -> dict[str, Any]:
products = await self._ensure_data()
# Start with keyword pre-filter if provided
if keyword:
candidates = self._fuzzy_find(products, keyword, threshold=50)
else:
candidates = products
results = []
for p in candidates:
if series and series.lower() not in (p.get("seriesName") or "").lower():
continue
if product_type and (p.get("type") or "").lower() != product_type.lower():
continue
if wifi is True and not _has_capability(p.get("wifi", "")):
continue
if wifi is False and _has_capability(p.get("wifi", "")):
continue
if bluetooth is True and not _has_capability(p.get("bluetooth", "")):
continue
if bluetooth is False and _has_capability(p.get("bluetooth", "")):
continue
if thread_zigbee is True and not _has_capability(p.get("threadZigbee", "")):
continue
if thread_zigbee is False and _has_capability(p.get("threadZigbee", "")):
continue
flash_mb, _ = _parse_memory_field(p.get("flash", ""))
if min_flash_mb is not None and flash_mb < min_flash_mb:
continue
psram_mb, _ = _parse_memory_field(p.get("psram", ""))
if min_psram_mb is not None and psram_mb < min_psram_mb:
continue
sram_kb = _parse_sram_kb(p.get("sram", ""))
if min_sram_kb is not None and sram_kb < min_sram_kb:
continue
gpio = int(p.get("gpio", 0) or 0)
if min_gpio is not None and gpio < min_gpio:
continue
if max_temp_c is not None:
temp_max = _parse_temp_max(p.get("operatingTemp", ""))
if temp_max is None or temp_max < max_temp_c:
continue
if status and _normalize_status(p.get("status", "")) != _normalize_status(status):
continue
if antenna and antenna.lower() not in (p.get("antenna") or "").lower():
continue
results.append(_product_summary(p))
return {
"count": len(results),
"filters_applied": {
k: v
for k, v in {
"series": series,
"product_type": product_type,
"wifi": wifi,
"bluetooth": bluetooth,
"thread_zigbee": thread_zigbee,
"min_flash_mb": min_flash_mb,
"min_psram_mb": min_psram_mb,
"min_sram_kb": min_sram_kb,
"min_gpio": min_gpio,
"max_temp_c": max_temp_c,
"status": status,
"antenna": antenna,
"keyword": keyword,
}.items()
if v is not None
},
"products": results,
}
async def _product_info_impl(
self, context: Context, name: str
) -> dict[str, Any]:
products = await self._ensure_data()
matches = self._fuzzy_find(products, name, threshold=65)
if not matches:
return {"error": f"No product found matching '{name}'", "suggestion": "Try esp_product_search with a keyword to browse available products"}
return _product_detail(matches[0])
async def _chip_compare_impl(
self, context: Context, products: list[str]
) -> dict[str, Any]:
if len(products) < 2:
return {"error": "Provide at least 2 product names to compare"}
if len(products) > 4:
return {"error": "Compare up to 4 products at a time"}
all_products = await self._ensure_data()
resolved = []
not_found = []
for name in products:
matches = self._fuzzy_find(all_products, name, threshold=65)
if matches:
resolved.append(_product_detail(matches[0]))
else:
not_found.append(name)
if not_found:
return {"error": f"Could not find: {', '.join(not_found)}", "found": [r["name"] for r in resolved]}
# Build comparison — highlight fields that differ
all_keys = list(resolved[0].keys())
differences = {}
common = {}
for key in all_keys:
values = [r[key] for r in resolved]
if len({str(v) for v in values}) > 1:
differences[key] = {r["name"]: r[key] for r in resolved}
else:
common[key] = values[0]
return {
"products_compared": [r["name"] for r in resolved],
"differences": differences,
"common": common,
"full_specs": {r["name"]: r for r in resolved},
}
async def _product_recommend_impl(
self,
context: Context,
use_case: str,
constraints: str | None,
prefer_module: bool,
) -> dict[str, Any]:
products = await self._ensure_data()
# Parse use case and constraints into filter criteria
text = f"{use_case} {constraints or ''}".lower()
wants_wifi = any(w in text for w in ("wifi", "wi-fi", "wireless", "iot", "mqtt", "http"))
wants_ble = any(w in text for w in ("ble", "bluetooth", "beacon"))
wants_thread = any(w in text for w in ("thread", "zigbee", "matter", "802.15.4"))
wants_psram = "psram" in text
wants_industrial = any(w in text for w in ("industrial", "105", "automotive", "harsh"))
# Extract numeric constraints from text
min_flash = None
m = re.search(r"(\d+)\s*mb\s*flash", text)
if m:
min_flash = int(m.group(1))
min_psram = None
m = re.search(r"(\d+)\s*mb\s*psram", text)
if m:
min_psram = int(m.group(1))
candidates = []
for p in products:
# Only recommend actively produced parts
if _normalize_status(p.get("status", "")) != "Mass Production":
continue
if prefer_module and (p.get("type") or "").lower() != "module":
continue
if wants_wifi and not _has_capability(p.get("wifi", "")):
continue
if wants_ble and not _has_capability(p.get("bluetooth", "")):
continue
if wants_thread and not _has_capability(p.get("threadZigbee", "")):
continue
flash_mb, _ = _parse_memory_field(p.get("flash", ""))
psram_mb, _ = _parse_memory_field(p.get("psram", ""))
if wants_psram and psram_mb == 0:
continue
if min_flash is not None and flash_mb < min_flash:
continue
if min_psram is not None and psram_mb < min_psram:
continue
if wants_industrial:
temp_max = _parse_temp_max(p.get("operatingTemp", ""))
if temp_max is None or temp_max < 105:
continue
candidates.append(_product_summary(p))
# If module preference yields nothing, fall back to including SoCs
if not candidates and prefer_module:
return await self._product_recommend_impl(context, use_case, constraints, False)
return {
"use_case": use_case,
"constraints_parsed": {
"wifi": wants_wifi,
"bluetooth": wants_ble,
"thread_zigbee": wants_thread,
"psram_required": wants_psram,
"industrial_temp": wants_industrial,
"min_flash_mb": min_flash,
"min_psram_mb": min_psram,
"prefer_module": prefer_module,
},
"count": len(candidates),
"recommendations": candidates[:15],
}
async def _product_availability_impl(
self,
context: Context,
series: str | None,
status: str,
max_lead_weeks: int | None,
max_moq: int | None,
) -> dict[str, Any]:
products = await self._ensure_data()
results = []
for p in products:
if series and series.lower() not in (p.get("seriesName") or "").lower():
continue
if _normalize_status(p.get("status", "")) != _normalize_status(status):
continue
lead_weeks = _parse_lead_weeks(p.get("leadTime", ""))
if max_lead_weeks is not None and lead_weeks is not None and lead_weeks > max_lead_weeks:
continue
moq_val = p.get("moq", "")
moq_int = None
if moq_val:
m = re.search(r"(\d+)", str(moq_val))
if m:
moq_int = int(m.group(1))
if max_moq is not None and moq_int is not None and moq_int > max_moq:
continue
results.append({
**_product_summary(p),
"lead_time": p.get("leadTime", ""),
"moq": p.get("moq", ""),
"spq": p.get("spq", ""),
"mpn": p.get("mpn", ""),
"replaced_by": p.get("replacedByName", ""),
})
return {
"count": len(results),
"filters": {
"series": series,
"status": status,
"max_lead_weeks": max_lead_weeks,
"max_moq": max_moq,
},
"products": results,
}

13
src/mcesptool/server.py
View File

@ -22,6 +22,7 @@ from .components import (
FlashManager,
OTAManager,
PartitionManager,
ProductCatalog,
ProductionTools,
QemuManager,
SecurityManager,
@ -81,6 +82,9 @@ class ESPToolServer:
self.components["production_tools"] = ProductionTools(self.app, self.config)
self.components["diagnostics"] = Diagnostics(self.app, self.config)
# Product catalog (no hardware dependency — always enabled)
self.components["product_catalog"] = ProductCatalog(self.app, self.config)
# QEMU emulation (if available)
if self.config.get_qemu_available():
self.components["qemu_manager"] = QemuManager(self.app, self.config)
@ -114,7 +118,7 @@ class ESPToolServer:
# Get tool and resource counts via public API
tools = await self.app.get_tools()
# Note: FastMCP doesn't expose get_resources(), so we count our components
resource_count = 3 # esp://server/status, esp://config, esp://capabilities
resource_count = 4 # esp://server/status, esp://config, esp://capabilities, esp://products/catalog
return {
"server_name": "MCP ESPTool Server",
@ -175,6 +179,13 @@ class ESPToolServer:
"esp_performance_profile",
"esp_diagnostic_report",
],
"product_catalog": [
"esp_product_search",
"esp_product_info",
"esp_chip_compare",
"esp_product_recommend",
"esp_product_availability",
],
}
if self.config.get_qemu_available():

514
tests/test_product_catalog.py Normal file
View File

@ -0,0 +1,514 @@
"""
Tests for the Product Catalog component.
Uses mocked HTTP responses to avoid hitting the real Espressif API.
"""
import time
from unittest.mock import AsyncMock, patch
import httpx
import pytest
from fastmcp import FastMCP
from mcesptool.components.product_catalog import (
ProductCatalog,
_has_capability,
_normalize_status,
_parse_lead_weeks,
_parse_memory_field,
_parse_sram_kb,
_parse_temp_max,
)
from mcesptool.config import ESPToolServerConfig
# ── Fixtures ────────────────────────────────────────────────────────────
SAMPLE_PRODUCTS = [
{
"id": 1,
"name": "ESP32-S3",
"nameNew": "ESP32-S3",
"type": "SoC",
"seriesName": "ESP32-S3",
"wifi": "802.11 b/g/n",
"wifi6": "",
"bluetooth": "BLE 5.0",
"threadZigbee": "",
"flash": "0",
"psram": "0",
"sram": "512",
"rom": "384",
"gpio": "45",
"pins": "56",
"freq": "240",
"operatingTemp": "-40~85C",
"voltageRange": "3.0~3.6V",
"dimensions": "7x7 mm",
"status": "Mass Production",
"antenna": "",
"mpn": "ESP32-S3",
"sizeType": "QFN56",
"releaseTime": "2020",
"idfSupports": "v4.4+",
"spq": "490",
"moq": "4900",
"leadTime": "8-10 weeks",
"eccnCode": "5A992.c",
"ccatszCode": "",
"hsCode": "8542.31",
"preFirmware": "",
"replacedByName": "",
},
{
"id": 2,
"name": "ESP32-S3-WROOM-1-N16R8",
"nameNew": "ESP32-S3-WROOM-1-N16R8",
"type": "Module",
"seriesName": "ESP32-S3",
"wifi": "802.11 b/g/n",
"wifi6": "",
"bluetooth": "BLE 5.0",
"threadZigbee": "",
"flash": "16, Quad",
"psram": "8, Octal",
"sram": "512",
"rom": "384",
"gpio": "36",
"pins": "44",
"freq": "240",
"operatingTemp": "-40~85C",
"voltageRange": "3.0~3.6V",
"dimensions": "18x25.5x3.1 mm",
"status": "Mass Production",
"antenna": "PCB",
"mpn": "ESP32-S3-WROOM-1-N16R8",
"sizeType": "",
"releaseTime": "2021",
"idfSupports": "v4.4+",
"spq": "400",
"moq": "2000",
"leadTime": "6-8 weeks",
"eccnCode": "5A992.c",
"ccatszCode": "",
"hsCode": "8542.31",
"preFirmware": "",
"replacedByName": "",
},
{
"id": 3,
"name": "ESP32-C6-MINI-1-N4",
"nameNew": "ESP32-C6-MINI-1-N4",
"type": "Module",
"seriesName": "ESP32-C6",
"wifi": "802.11 b/g/n/ax",
"wifi6": "Yes",
"bluetooth": "BLE 5.3",
"threadZigbee": "Available",
"flash": "4, Quad",
"psram": "",
"sram": "512",
"rom": "320",
"gpio": "22",
"pins": "53",
"freq": "160",
"operatingTemp": "-40~105C",
"voltageRange": "3.0~3.6V",
"dimensions": "13.2x16.6x2.4 mm",
"status": "Mass Production",
"antenna": "PCB",
"mpn": "ESP32-C6-MINI-1-N4",
"sizeType": "",
"releaseTime": "2023",
"idfSupports": "v5.1+",
"spq": "500",
"moq": "2500",
"leadTime": "4-6 weeks",
"eccnCode": "5A992.c",
"ccatszCode": "",
"hsCode": "8542.31",
"preFirmware": "",
"replacedByName": "",
},
{
"id": 4,
"name": "ESP32-WROOM-32E",
"nameNew": "ESP32-WROOM-32E",
"type": "Module",
"seriesName": "ESP32",
"wifi": "802.11 b/g/n",
"wifi6": "",
"bluetooth": "BT/BLE 4.2",
"threadZigbee": "N/A",
"flash": "4, Quad",
"psram": "",
"sram": "520",
"rom": "448",
"gpio": "34",
"pins": "38",
"freq": "240",
"operatingTemp": "-40~85C",
"voltageRange": "3.0~3.6V",
"dimensions": "18x20x3.2 mm",
"status": "NRND",
"antenna": "PCB",
"mpn": "ESP32-WROOM-32E",
"sizeType": "",
"releaseTime": "2019",
"idfSupports": "v4.0+",
"spq": "500",
"moq": "5000",
"leadTime": "12-14 weeks",
"eccnCode": "5A992.c",
"ccatszCode": "",
"hsCode": "8542.31",
"preFirmware": "",
"replacedByName": "",
},
]
def _mock_response() -> httpx.Response:
"""Build a mock httpx response with sample product data."""
return httpx.Response(
status_code=200,
json={"options": {}, "results": SAMPLE_PRODUCTS},
request=httpx.Request("GET", "https://products.espressif.com/api/user/products"),
)
@pytest.fixture
def app():
return FastMCP("test-esp")
@pytest.fixture
def config():
with patch.object(ESPToolServerConfig, "_check_tool_availability", return_value=True):
return ESPToolServerConfig.from_environment()
@pytest.fixture
def catalog(app, config):
cat = ProductCatalog(app, config)
# Pre-load the cache so tests don't hit the network
cat._products = SAMPLE_PRODUCTS
cat._last_fetch = time.time()
return cat
@pytest.fixture
def context():
"""Minimal mock MCP context."""
return AsyncMock(spec=[])
# ── Helper function unit tests ──────────────────────────────────────────
class TestParsers:
def test_parse_memory_basic(self):
assert _parse_memory_field("16, Quad") == (16, "Quad")
assert _parse_memory_field("8, Octal") == (8, "Octal")
assert _parse_memory_field("4") == (4, "")
def test_parse_memory_bare_int(self):
assert _parse_memory_field(8) == (8, "")
assert _parse_memory_field(0) == (0, "")
def test_parse_memory_empty(self):
assert _parse_memory_field("") == (0, "")
assert _parse_memory_field("N/A") == (0, "")
assert _parse_memory_field("NA") == (0, "")
assert _parse_memory_field(None) == (0, "")
def test_parse_sram(self):
assert _parse_sram_kb("512") == 512
assert _parse_sram_kb("520") == 520
assert _parse_sram_kb("") == 0
assert _parse_sram_kb(512) == 512
assert _parse_sram_kb(None) == 0
def test_has_capability(self):
assert _has_capability("802.11 b/g/n") is True
assert _has_capability("BLE 5.0") is True
assert _has_capability("Available") is True
assert _has_capability("") is False
assert _has_capability("N/A") is False
assert _has_capability("NA") is False
def test_normalize_status(self):
assert _normalize_status("Mass Production") == "Mass Production"
assert _normalize_status("mass production") == "Mass Production"
assert _normalize_status("NRND") == "NRND"
assert _normalize_status("nrnd") == "NRND"
def test_parse_lead_weeks(self):
assert _parse_lead_weeks("8-10 weeks") == 8
assert _parse_lead_weeks("4-6 weeks") == 4
assert _parse_lead_weeks("") is None
assert _parse_lead_weeks("TBD") is None
def test_parse_temp_max(self):
assert _parse_temp_max("-40~85C") == 85
assert _parse_temp_max("-40~105C") == 105
assert _parse_temp_max("") is None
# ── Tool integration tests ──────────────────────────────────────────────
@pytest.mark.asyncio
class TestProductSearch:
async def test_search_no_filters(self, catalog, context):
result = await catalog._product_search_impl(context)
assert result["count"] == 4
assert len(result["products"]) == 4
async def test_search_by_series(self, catalog, context):
result = await catalog._product_search_impl(context, series="ESP32-S3")
assert result["count"] == 2
names = [p["name"] for p in result["products"]]
assert "ESP32-S3" in names
assert "ESP32-S3-WROOM-1-N16R8" in names
async def test_search_by_type(self, catalog, context):
result = await catalog._product_search_impl(context, product_type="SoC")
assert result["count"] == 1
assert result["products"][0]["name"] == "ESP32-S3"
async def test_search_by_wifi(self, catalog, context):
result = await catalog._product_search_impl(context, wifi=True)
assert result["count"] == 4 # all sample products have wifi
async def test_search_by_thread_zigbee(self, catalog, context):
result = await catalog._product_search_impl(context, thread_zigbee=True)
assert result["count"] == 1
assert result["products"][0]["name"] == "ESP32-C6-MINI-1-N4"
async def test_search_by_min_flash(self, catalog, context):
result = await catalog._product_search_impl(context, min_flash_mb=8)
assert result["count"] == 1
assert result["products"][0]["name"] == "ESP32-S3-WROOM-1-N16R8"
async def test_search_by_min_psram(self, catalog, context):
result = await catalog._product_search_impl(context, min_psram_mb=4)
assert result["count"] == 1
assert result["products"][0]["flash_mb"] == 16
async def test_search_by_status(self, catalog, context):
result = await catalog._product_search_impl(context, status="NRND")
assert result["count"] == 1
assert result["products"][0]["name"] == "ESP32-WROOM-32E"
async def test_search_by_antenna(self, catalog, context):
result = await catalog._product_search_impl(context, antenna="PCB")
assert result["count"] == 3
async def test_search_by_max_temp(self, catalog, context):
result = await catalog._product_search_impl(context, max_temp_c=105)
assert result["count"] == 1
assert result["products"][0]["name"] == "ESP32-C6-MINI-1-N4"
async def test_search_keyword(self, catalog, context):
result = await catalog._product_search_impl(context, keyword="WROOM")
assert result["count"] >= 1
names = [p["name"] for p in result["products"]]
assert any("WROOM" in n for n in names)
async def test_search_combined_filters(self, catalog, context):
result = await catalog._product_search_impl(
context, series="ESP32-S3", product_type="Module", min_flash_mb=8
)
assert result["count"] == 1
assert result["products"][0]["name"] == "ESP32-S3-WROOM-1-N16R8"
async def test_search_no_results(self, catalog, context):
result = await catalog._product_search_impl(context, min_flash_mb=128)
assert result["count"] == 0
assert result["products"] == []
@pytest.mark.asyncio
class TestProductInfo:
async def test_info_exact_name(self, catalog, context):
result = await catalog._product_info_impl(context, "ESP32-S3-WROOM-1-N16R8")
assert result["name"] == "ESP32-S3-WROOM-1-N16R8"
assert result["flash_mb"] == 16
assert result["psram_mb"] == 8
assert result["series"] == "ESP32-S3"
async def test_info_fuzzy_name(self, catalog, context):
result = await catalog._product_info_impl(context, "s3 wroom n16r8")
assert result["name"] == "ESP32-S3-WROOM-1-N16R8"
async def test_info_not_found(self, catalog, context):
result = await catalog._product_info_impl(context, "ESP99-NONEXISTENT")
assert "error" in result
@pytest.mark.asyncio
class TestChipCompare:
async def test_compare_two(self, catalog, context):
result = await catalog._chip_compare_impl(
context, ["ESP32-S3-WROOM-1-N16R8", "ESP32-C6-MINI-1-N4"]
)
assert len(result["products_compared"]) == 2
assert "differences" in result
assert "common" in result
# These two differ in series, flash, etc.
assert "series" in result["differences"]
async def test_compare_too_few(self, catalog, context):
result = await catalog._chip_compare_impl(context, ["ESP32-S3"])
assert "error" in result
async def test_compare_too_many(self, catalog, context):
result = await catalog._chip_compare_impl(
context, ["a", "b", "c", "d", "e"]
)
assert "error" in result
async def test_compare_not_found(self, catalog, context):
result = await catalog._chip_compare_impl(
context, ["ESP32-S3", "NONEXISTENT-CHIP"]
)
assert "error" in result
@pytest.mark.asyncio
class TestProductRecommend:
async def test_recommend_ble_sensor(self, catalog, context):
result = await catalog._product_recommend_impl(
context, "battery-powered BLE sensor", None, True
)
assert result["count"] >= 1
# All recommendations should be modules (prefer_module=True) and have BLE
for p in result["recommendations"]:
assert p["type"] == "Module"
assert p["bluetooth"] is True
async def test_recommend_thread(self, catalog, context):
result = await catalog._product_recommend_impl(
context, "Thread/Matter smart home device", None, True
)
assert result["count"] >= 1
for p in result["recommendations"]:
assert p["thread_zigbee"] is True
async def test_recommend_with_psram(self, catalog, context):
result = await catalog._product_recommend_impl(
context, "wifi camera needs psram", None, True
)
for p in result["recommendations"]:
assert p["psram_mb"] > 0
async def test_recommend_industrial(self, catalog, context):
result = await catalog._product_recommend_impl(
context, "industrial IoT gateway", "industrial temp range", True
)
# Only ESP32-C6-MINI-1-N4 has 105C rating in our sample data
assert result["count"] >= 1
async def test_recommend_fallback_to_soc(self, catalog, context):
"""When no modules match, falls back to including SoCs."""
result = await catalog._product_recommend_impl(
context, "wifi device", "needs 32MB flash", True
)
# No modules have 32MB flash — falls back to SoC search too
assert isinstance(result["count"], int)
@pytest.mark.asyncio
class TestProductAvailability:
async def test_availability_mass_production(self, catalog, context):
result = await catalog._product_availability_impl(
context, None, "Mass Production", None, None
)
assert result["count"] == 3 # NRND one excluded
async def test_availability_by_series(self, catalog, context):
result = await catalog._product_availability_impl(
context, "ESP32-S3", "Mass Production", None, None
)
assert result["count"] == 2
async def test_availability_lead_time(self, catalog, context):
result = await catalog._product_availability_impl(
context, None, "Mass Production", 6, None
)
# ESP32-C6-MINI-1-N4 (4 weeks) and ESP32-S3-WROOM-1-N16R8 (6 weeks) both pass <= 6
assert result["count"] == 2
names = [p["name"] for p in result["products"]]
assert "ESP32-C6-MINI-1-N4" in names
assert "ESP32-S3-WROOM-1-N16R8" in names
async def test_availability_max_moq(self, catalog, context):
result = await catalog._product_availability_impl(
context, None, "Mass Production", None, 2500
)
# ESP32-S3-WROOM-1-N16R8 (2000) and ESP32-C6-MINI-1-N4 (2500)
assert result["count"] == 2
# ── Cache behavior tests ────────────────────────────────────────────────
@pytest.mark.asyncio
class TestCaching:
async def test_cache_hit(self, catalog):
"""Cache should be used when data is fresh."""
original_fetch_time = catalog._last_fetch
products = await catalog._ensure_data()
assert len(products) == 4
assert catalog._last_fetch == original_fetch_time # no new fetch
async def test_cache_expired(self, catalog):
"""Cache should refresh when TTL expired."""
catalog._last_fetch = time.time() - 90000 # older than 24h
mock_resp = _mock_response()
with patch("mcesptool.components.product_catalog.httpx.AsyncClient") as mock_client_cls:
mock_client = AsyncMock()
mock_client.get.return_value = mock_resp
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client_cls.return_value = mock_client
products = await catalog._ensure_data()
assert len(products) == 4
mock_client.get.assert_called_once()
async def test_force_refresh(self, catalog):
"""Force refresh bypasses cache."""
mock_resp = _mock_response()
with patch("mcesptool.components.product_catalog.httpx.AsyncClient") as mock_client_cls:
mock_client = AsyncMock()
mock_client.get.return_value = mock_resp
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client_cls.return_value = mock_client
products = await catalog._ensure_data(force_refresh=True)
assert len(products) == 4
mock_client.get.assert_called_once()
# ── Fuzzy matching tests ────────────────────────────────────────────────
class TestFuzzyMatch:
def test_exact_match(self, catalog):
results = catalog._fuzzy_find(SAMPLE_PRODUCTS, "ESP32-S3")
assert len(results) >= 1
assert results[0]["name"] == "ESP32-S3"
def test_substring_match(self, catalog):
results = catalog._fuzzy_find(SAMPLE_PRODUCTS, "WROOM")
assert len(results) >= 1
names = [p["name"] for p in results]
assert any("WROOM" in n for n in names)
def test_no_match(self, catalog):
results = catalog._fuzzy_find(SAMPLE_PRODUCTS, "XYZNONEXISTENT", threshold=90)
assert len(results) == 0

2
uv.lock generated
View File

@ -893,6 +893,7 @@ source = { editable = "." }
dependencies = [
{ name = "click" },
{ name = "fastmcp" },
{ name = "httpx" },
{ name = "pydantic" },
{ name = "pyserial" },
{ name = "pyserial-asyncio" },
@ -933,6 +934,7 @@ requires-dist = [
{ name = "factory-boy", marker = "extra == 'testing'", specifier = ">=3.3.0" },
{ name = "fastmcp", specifier = ">=3.0.2,<4" },
{ name = "gunicorn", marker = "extra == 'production'", specifier = ">=21.0.0" },
{ name = "httpx", specifier = ">=0.28" },
{ name = "kconfiglib", marker = "extra == 'idf'", specifier = ">=14.1.0" },
{ name = "mypy", marker = "extra == 'dev'", specifier = ">=1.5.0" },
{ name = "pre-commit", marker = "extra == 'dev'", specifier = ">=3.0.0" },