diff --git a/build_kicad_v10_indexes.py b/build_kicad_v10_indexes.py new file mode 100644 index 0000000..3ef7412 --- /dev/null +++ b/build_kicad_v10_indexes.py @@ -0,0 +1,45 @@ +#!/usr/bin/env python3 +""" +Prebuild persistent indexes for the KiCad v10 library cache. +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import sys + + +SCRIPT_DIR = Path(__file__).resolve().parent +if str(SCRIPT_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPT_DIR)) + +from mcp_servers.kicad_text import ( # type: ignore + clear_container_indexes, + footprint_index_path, + prewarm_container_indexes, +) +from mcp_servers.kicad_symbol_sqlite import build_symbol_sqlite_index, symbol_sqlite_index_path # type: ignore + + +def main() -> int: + parser = argparse.ArgumentParser(description="Prebuild KiCad v10 symbol and footprint indexes") + parser.add_argument("--refresh", action="store_true", help="Clear existing indexes before rebuilding") + args = parser.parse_args() + + if args.refresh: + clear_container_indexes() + + symbol_summary = build_symbol_sqlite_index(refresh=False) + summary = prewarm_container_indexes(refresh=False) + summary["symbol_index"] = str(symbol_sqlite_index_path()) + summary["symbol_components"] = symbol_summary.get("symbol_components", summary.get("symbol_components", 0)) + summary["symbol_libraries"] = symbol_summary.get("symbol_libraries", summary.get("symbol_libraries", 0)) + summary["footprint_index"] = str(footprint_index_path()) + print(json.dumps(summary, ensure_ascii=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/mcp_servers/component_db.py b/mcp_servers/component_db.py index 61383c1..61b22c8 100644 --- a/mcp_servers/component_db.py +++ b/mcp_servers/component_db.py @@ -1,205 +1,482 @@ #!/usr/bin/env python3 """ -Simple MCP Server for component database functionality -This demonstrates how to create an MCP server for KiCad integration +KiCad component database MCP server. + +The server stays line-delimited JSON-RPC because the existing KIC-AI plugin +clients read one JSON object per line from stdout. """ +from __future__ import annotations + +import asyncio import json import sys -import asyncio -from typing import Dict, List, Any +from pathlib import Path +from typing import Any, Dict, List, Optional, Sequence + +try: + from .kicad_text import ( + KiCadComponent, + container_symbols_dir, + discover_symbol_library_files, + discover_schematic_files, + discover_search_roots, + parse_inline_library_symbols, + parse_schematic_components, + ) + from .kicad_symbol_sqlite import ( + catalog_size as sqlite_catalog_size, + resolve_symbol_component, + search_symbol_components, + source_count as sqlite_source_count, + source_sample as sqlite_source_sample, + ) +except ImportError: + from kicad_text import ( # type: ignore + KiCadComponent, + container_symbols_dir, + discover_symbol_library_files, + discover_schematic_files, + discover_search_roots, + parse_inline_library_symbols, + parse_schematic_components, + ) + from kicad_symbol_sqlite import ( # type: ignore + catalog_size as sqlite_catalog_size, + resolve_symbol_component, + search_symbol_components, + source_count as sqlite_source_count, + source_sample as sqlite_source_sample, + ) + class ComponentDatabaseServer: - """Simple component database MCP server""" - + """Component database backed by locally discoverable KiCad files.""" + def __init__(self): - self.component_db = { - # Sample component database - "R1": { - "type": "resistor", - "value": "10k", - "package": "0805", - "price": 0.02, - "stock": 5000, - "alternatives": ["R1206-10K", "R0603-10K"] - }, - "C1": { - "type": "capacitor", - "value": "100nF", - "package": "0805", - "price": 0.05, - "stock": 2000, - "alternatives": ["C1206-100N", "C0603-100N"] - } + self.base_dir = Path(__file__).resolve() + self.search_roots = discover_search_roots(self.base_dir) + self.container_symbols_root = container_symbols_dir() + self._catalog: Optional[List[KiCadComponent]] = None + self._sources: Optional[List[str]] = None + + def _load_catalog(self) -> List[KiCadComponent]: + if self._catalog is not None: + return self._catalog + + components: List[KiCadComponent] = [] + source_files: List[str] = [] + seen = set() + + for schematic_path in discover_schematic_files(self.search_roots): + source_files.append(str(schematic_path)) + + for component in parse_schematic_components(schematic_path): + key = ( + component.lib_id, + component.value, + component.footprint, + component.manufacturer, + component.part_number, + component.lcsc_id, + ) + if key in seen: + continue + seen.add(key) + components.append(component) + + # Inline libraries from template and expanded schematics are useful + # when no global KiCad libraries are installed on the host. + for component in parse_inline_library_symbols(schematic_path): + key = ( + "inline", + component.lib_id, + component.value, + component.footprint, + component.manufacturer, + component.part_number, + component.lcsc_id, + ) + if key in seen: + continue + seen.add(key) + components.append(component) + + for library_path in discover_symbol_library_files(self.search_roots): + try: + if library_path.resolve().parent == self.container_symbols_root.resolve(): + continue + except FileNotFoundError: + pass + source_files.append(str(library_path)) + for component in parse_inline_library_symbols(library_path): + key = ( + "library", + component.lib_id, + component.value, + component.footprint, + component.manufacturer, + component.part_number, + component.lcsc_id, + ) + if key in seen: + continue + seen.add(key) + components.append(component) + + self._catalog = components + self._sources = sorted(set(source_files)) + return self._catalog + + def _local_catalog_sources(self) -> List[str]: + self._load_catalog() + return self._sources or [] + + def _catalog_sources_sample(self, limit: int = 10) -> List[str]: + sample = list(self._local_catalog_sources()[:limit]) + if len(sample) >= limit: + return sample + + for source in sqlite_source_sample(limit=limit): + if source in sample: + continue + sample.append(source) + if len(sample) >= limit: + break + return sample + + def _catalog_source_count(self) -> int: + return len(self._local_catalog_sources()) + sqlite_source_count() + + def _catalog_size(self) -> int: + return len(self._load_catalog()) + sqlite_catalog_size() + + def _resolve_component(self, identifier: str) -> Optional[KiCadComponent]: + needle = identifier.strip().lower() + if not needle: + return None + + for component in self._load_catalog(): + candidates = [ + component.lib_id, + component.reference, + component.value, + component.part_number, + component.lcsc_id, + f"{component.lib_id}:{component.reference}", + ] + if any(candidate and candidate.lower() == needle for candidate in candidates): + return component + + return resolve_symbol_component(identifier) + + def _matches_type(self, component: KiCadComponent, component_type: str) -> bool: + if not component_type: + return True + wanted = component_type.strip().lower() + return wanted in component.component_type.lower() or wanted in component.lib_id.lower() + + def _matches_specs(self, component: KiCadComponent, specs: Dict[str, Any]) -> bool: + for key, value in specs.items(): + expected = str(value).strip().lower() + if not expected: + continue + haystacks: Sequence[str] + key_lower = key.lower() + if key_lower in {"value", "resistance", "capacitance"}: + haystacks = [component.value] + elif key_lower in {"package", "footprint"}: + haystacks = [component.package, component.footprint] + elif key_lower in {"manufacturer", "brand"}: + haystacks = [component.manufacturer] + elif key_lower in {"part", "part_number", "mpn"}: + haystacks = [component.part_number] + elif key_lower in {"lcsc", "lcsc_id"}: + haystacks = [component.lcsc_id] + elif key_lower in {"type", "category"}: + haystacks = [component.component_type, component.category] + else: + haystacks = [ + component.reference, + component.value, + component.footprint, + component.description, + component.manufacturer, + component.part_number, + component.lcsc_id, + component.lib_id, + component.category, + ] + if not any(expected in (candidate or "").lower() for candidate in haystacks): + return False + return True + + def _score_component(self, component: KiCadComponent, component_type: str, specs: Dict[str, Any]) -> int: + score = 0 + if component_type: + wanted = component_type.strip().lower() + if component.component_type == wanted: + score += 80 + elif wanted in component.lib_id.lower(): + score += 40 + + for key, value in specs.items(): + expected = str(value).strip().lower() + if not expected: + continue + if expected == component.value.lower(): + score += 60 + elif expected in component.value.lower(): + score += 30 + if expected and expected in component.footprint.lower(): + score += 25 + if expected and expected in component.part_number.lower(): + score += 35 + if expected and expected in component.description.lower(): + score += 15 + + if component.lcsc_id: + score += 10 + if component.manufacturer: + score += 5 + return score + + def _alternatives(self, target: KiCadComponent, limit: int = 3) -> List[str]: + matches: List[str] = [] + for component in self._load_catalog(): + if component.lib_id == target.lib_id: + continue + if component.component_type != target.component_type: + continue + if target.value and component.value and component.value.lower() != target.value.lower(): + continue + matches.append(component.lib_id) + if len(matches) >= limit: + return matches + + extra = search_symbol_components( + target.component_type, + {"value": target.value} if target.value else {}, + limit=limit + 3, + ) + for component in extra: + if component.lib_id == target.lib_id or component.lib_id in matches: + continue + matches.append(component.lib_id) + if len(matches) >= limit: + break + return matches + + def _search_components(self, arguments: Dict[str, Any]) -> Dict[str, Any]: + component_type = arguments.get("type", "") + specs = arguments.get("specs", {}) + limit = int(arguments.get("limit", 20)) + + matches = [ + component + for component in self._load_catalog() + if self._matches_type(component, component_type) and self._matches_specs(component, specs) + ] + matches.sort( + key=lambda component: self._score_component(component, component_type, specs), + reverse=True, + ) + seen_ids = {component.lib_id for component in matches} + for component in search_symbol_components(component_type, specs, limit=max(limit * 3, limit)): + if component.lib_id in seen_ids: + continue + seen_ids.add(component.lib_id) + matches.append(component) + + matches.sort( + key=lambda component: self._score_component(component, component_type, specs), + reverse=True, + ) + results = [] + for component in matches[:limit]: + payload = component.to_dict(include_values=True, include_footprints=True) + payload["part_id"] = component.lib_id + payload["alternatives"] = self._alternatives(component) + results.append(payload) + + return { + "components": results, + "total_found": len(results), + "indexed_source_count": self._catalog_source_count(), + "indexed_sources_sample": self._catalog_sources_sample(), + "catalog_size": self._catalog_size(), + "data_mode": "file-backed-local+container-sqlite", } - + + def _get_pricing(self, arguments: Dict[str, Any]) -> Dict[str, Any]: + identifiers = list(arguments.get("part_numbers", [])) + if not identifiers and arguments.get("part_number"): + identifiers.append(arguments["part_number"]) + + pricing: Dict[str, Dict[str, Any]] = {} + for identifier in identifiers: + component = self._resolve_component(identifier) + if component is None: + pricing[identifier] = { + "available": False, + "message": "Component not found in local KiCad catalog", + } + continue + + pricing[identifier] = { + "available": component.unit_price is not None, + "unit_price": component.unit_price, + "currency": "USD" if component.unit_price is not None else None, + "stock": component.stock, + "lcsc_id": component.lcsc_id or None, + "manufacturer": component.manufacturer or None, + "part_number": component.part_number or None, + "source": component.source_path, + } + + return { + "pricing": pricing, + "data_mode": "file-backed-local+container-sqlite", + "note": "Pricing is only returned when local or cached KiCad symbol metadata contains explicit price fields.", + } + + def _check_availability(self, arguments: Dict[str, Any]) -> Dict[str, Any]: + identifiers = list(arguments.get("part_numbers", [])) + availability: Dict[str, Any] = {} + details: Dict[str, Dict[str, Any]] = {} + for identifier in identifiers: + component = self._resolve_component(identifier) + if component is None: + availability[identifier] = None + details[identifier] = { + "available": False, + "message": "Component not found in local KiCad catalog", + } + continue + + availability[identifier] = component.stock + details[identifier] = { + "available": component.stock is not None, + "stock": component.stock, + "lcsc_id": component.lcsc_id or None, + "source": component.source_path, + } + + return { + "availability": availability, + "details": details, + "data_mode": "file-backed-local+container-sqlite", + "note": "Availability is only returned when local or cached KiCad symbol metadata contains explicit stock fields.", + } + async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]: - """Handle MCP requests""" method = request.get("method") params = request.get("params", {}) request_id = request.get("id") - + if method == "initialize": return { "jsonrpc": "2.0", "id": request_id, "result": { - "protocolVersion": "2024-11-05", - "capabilities": { - "tools": {} - }, + "protocolVersion": "2025-03-26", + "capabilities": {"tools": {}}, "serverInfo": { "name": "component-database", - "version": "1.0.0" - } - } + "version": "2.1.0", + }, + }, } - - elif method == "tools/list": + + if method == "tools/list": return { - "jsonrpc": "2.0", + "jsonrpc": "2.0", "id": request_id, "result": { "tools": [ { "name": "search_components", - "description": "Search for components by type and specifications", + "description": "Search components indexed from local KiCad schematics, inline symbol libraries, and cached KiCad v10 symbol libraries", "inputSchema": { "type": "object", "properties": { "type": {"type": "string"}, - "specs": {"type": "object"} - } - } + "specs": {"type": "object"}, + "limit": {"type": "integer", "default": 20}, + }, + }, }, { "name": "get_pricing", - "description": "Get pricing for component part numbers", - "inputSchema": { - "type": "object", - "properties": { - "part_numbers": {"type": "array", "items": {"type": "string"}} - } - } - }, - { - "name": "check_availability", - "description": "Check stock availability for components", + "description": "Return explicit price metadata present in local or cached KiCad symbol properties", "inputSchema": { "type": "object", "properties": { - "part_numbers": {"type": "array", "items": {"type": "string"}} - } - } - } + "part_numbers": {"type": "array", "items": {"type": "string"}}, + "part_number": {"type": "string"}, + }, + }, + }, + { + "name": "check_availability", + "description": "Return explicit stock metadata present in local or cached KiCad symbol properties", + "inputSchema": { + "type": "object", + "properties": { + "part_numbers": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["part_numbers"], + }, + }, ] - } + }, } - - elif method == "tools/call": + + if method == "tools/call": tool_name = params.get("name") arguments = params.get("arguments", {}) - if tool_name == "search_components": - component_type = arguments.get("type", "") - specs = arguments.get("specs", {}) - - # Search through component database - results = [] - for part_id, component in self.component_db.items(): - # Filter by type if specified - if component_type and component.get("type", "").lower() != component_type.lower(): - continue - - # Filter by specs if specified - match = True - for spec_key, spec_value in specs.items(): - if spec_key in component: - if str(component[spec_key]).lower() != str(spec_value).lower(): - match = False - break - - if match: - results.append({ - "part_id": part_id, - "type": component.get("type"), - "value": component.get("value"), - "package": component.get("package"), - "price": component.get("price"), - "stock": component.get("stock"), - "alternatives": component.get("alternatives", []) - }) - - return { - "jsonrpc": "2.0", - "id": request_id, - "result": { - "components": results, - "total_found": len(results) - } - } - + result = self._search_components(arguments) elif tool_name == "get_pricing": - part_numbers = arguments.get("part_numbers", []) - pricing = {} - for part in part_numbers: - if part in self.component_db: - pricing[part] = { - "unit_price": self.component_db[part]["price"], - "stock": self.component_db[part]["stock"] - } - - return { - "jsonrpc": "2.0", - "id": request_id, - "result": {"pricing": pricing} - } - + result = self._get_pricing(arguments) elif tool_name == "check_availability": - part_numbers = arguments.get("part_numbers", []) - availability = {} - for part in part_numbers: - if part in self.component_db: - availability[part] = self.component_db[part]["stock"] - + result = self._check_availability(arguments) + else: return { "jsonrpc": "2.0", "id": request_id, - "result": {"availability": availability} + "error": {"code": -32601, "message": f"Unknown tool: {tool_name}"}, } - - # Default error response + return {"jsonrpc": "2.0", "id": request_id, "result": result} + return { "jsonrpc": "2.0", "id": request_id, - "error": {"code": -32601, "message": "Method not found"} + "error": {"code": -32601, "message": "Method not found"}, } + async def main(): - """Main MCP server loop""" server = ComponentDatabaseServer() - - # Read from stdin, write to stdout (MCP protocol) while True: try: line = sys.stdin.readline() if not line: break - + request = json.loads(line.strip()) response = await server.handle_request(request) - print(json.dumps(response)) sys.stdout.flush() - - except Exception as e: + except Exception as exc: error_response = { "jsonrpc": "2.0", "id": None, - "error": {"code": -32603, "message": f"Internal error: {str(e)}"} + "error": {"code": -32603, "message": f"Internal error: {exc}"}, } print(json.dumps(error_response)) sys.stdout.flush() + if __name__ == "__main__": asyncio.run(main()) diff --git a/mcp_servers/kicad_symbol_sqlite.py b/mcp_servers/kicad_symbol_sqlite.py new file mode 100644 index 0000000..8cdaf99 --- /dev/null +++ b/mcp_servers/kicad_symbol_sqlite.py @@ -0,0 +1,523 @@ +#!/usr/bin/env python3 +""" +SQLite-backed KiCad v10 symbol catalog. + +This module indexes the exported KiCad v10 symbol libraries once and exposes +search/query helpers for lightweight MCP servers. +""" + +from __future__ import annotations + +import json +from pathlib import Path +import sqlite3 +import tempfile +from typing import Any, Dict, List, Optional, Sequence, Tuple + +try: + from .kicad_text import ( + KiCadComponent, + container_library_cache_root, + container_symbols_dir, + ensure_container_library_cache, + parse_inline_library_symbols, + ) +except ImportError: + from kicad_text import ( # type: ignore + KiCadComponent, + container_library_cache_root, + container_symbols_dir, + ensure_container_library_cache, + parse_inline_library_symbols, + ) + + +INDEX_VERSION = 1 +SYMBOL_SQLITE_FILENAME = ".symbol_index.sqlite3" + + +def symbol_sqlite_index_path(cache_root: Optional[Path] = None) -> Path: + root = cache_root or container_library_cache_root() + return root / SYMBOL_SQLITE_FILENAME + + +def _normalize(value: Any) -> str: + return str(value or "").strip().lower() + + +def _fts_term(value: str) -> str: + escaped = value.replace('"', '""').strip() + return f'"{escaped}"' + + +def _connect(index_path: Path) -> sqlite3.Connection: + connection = sqlite3.connect(index_path) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA journal_mode=WAL") + connection.execute("PRAGMA synchronous=NORMAL") + connection.execute("PRAGMA temp_store=MEMORY") + return connection + + +def _index_meta(cache_root: Path) -> Dict[str, Any]: + symbols_dir = container_symbols_dir(cache_root) + library_files = sorted(symbols_dir.glob("*.kicad_sym")) + return { + "version": INDEX_VERSION, + "symbols_dir": str(symbols_dir), + "symbols_dir_mtime_ns": symbols_dir.stat().st_mtime_ns if symbols_dir.exists() else None, + "library_count": len(library_files), + "library_names": [path.name for path in library_files], + } + + +def _metadata_matches(connection: sqlite3.Connection, current_meta: Dict[str, Any]) -> bool: + row = connection.execute("SELECT value FROM metadata WHERE key = 'meta_json'").fetchone() + if row is None: + return False + try: + cached_meta = json.loads(row["value"]) + except json.JSONDecodeError: + return False + return cached_meta == current_meta + + +def _create_schema(connection: sqlite3.Connection) -> None: + connection.executescript( + """ + CREATE TABLE metadata ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + + CREATE TABLE sources ( + path TEXT PRIMARY KEY + ); + + CREATE TABLE symbols ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + lib_id TEXT NOT NULL, + reference TEXT NOT NULL, + value TEXT NOT NULL, + footprint TEXT NOT NULL, + datasheet TEXT NOT NULL, + description TEXT NOT NULL, + manufacturer TEXT NOT NULL, + part_number TEXT NOT NULL, + lcsc_id TEXT NOT NULL, + source_path TEXT NOT NULL, + category TEXT NOT NULL, + component_type TEXT NOT NULL, + package TEXT NOT NULL, + unit_price REAL, + stock INTEGER, + lib_id_lc TEXT NOT NULL, + reference_lc TEXT NOT NULL, + value_lc TEXT NOT NULL, + footprint_lc TEXT NOT NULL, + datasheet_lc TEXT NOT NULL, + description_lc TEXT NOT NULL, + manufacturer_lc TEXT NOT NULL, + part_number_lc TEXT NOT NULL, + lcsc_id_lc TEXT NOT NULL, + source_path_lc TEXT NOT NULL, + category_lc TEXT NOT NULL, + component_type_lc TEXT NOT NULL, + package_lc TEXT NOT NULL + ); + + CREATE INDEX idx_symbols_component_type ON symbols(component_type_lc); + CREATE INDEX idx_symbols_manufacturer ON symbols(manufacturer_lc); + CREATE INDEX idx_symbols_part_number ON symbols(part_number_lc); + CREATE INDEX idx_symbols_lib_id ON symbols(lib_id_lc); + CREATE INDEX idx_symbols_value ON symbols(value_lc); + CREATE INDEX idx_symbols_footprint ON symbols(footprint_lc); + CREATE INDEX idx_symbols_category ON symbols(category_lc); + CREATE INDEX idx_symbols_reference ON symbols(reference_lc); + CREATE INDEX idx_symbols_lcsc ON symbols(lcsc_id_lc); + + CREATE VIRTUAL TABLE symbols_fts USING fts5( + lib_id, + reference, + value, + footprint, + description, + manufacturer, + part_number, + lcsc_id, + category, + component_type, + package + ); + """ + ) + + +def _insert_component(connection: sqlite3.Connection, component: KiCadComponent) -> None: + cursor = connection.execute( + """ + INSERT INTO symbols ( + lib_id, reference, value, footprint, datasheet, description, + manufacturer, part_number, lcsc_id, source_path, category, + component_type, package, unit_price, stock, + lib_id_lc, reference_lc, value_lc, footprint_lc, datasheet_lc, + description_lc, manufacturer_lc, part_number_lc, lcsc_id_lc, + source_path_lc, category_lc, component_type_lc, package_lc + ) VALUES ( + ?, ?, ?, ?, ?, ?, + ?, ?, ?, ?, ?, + ?, ?, ?, ?, + ?, ?, ?, ?, ?, + ?, ?, ?, ?, + ?, ?, ?, ? + ) + """, + ( + component.lib_id, + component.reference, + component.value, + component.footprint, + component.datasheet, + component.description, + component.manufacturer, + component.part_number, + component.lcsc_id, + component.source_path, + component.category, + component.component_type, + component.package, + component.unit_price, + component.stock, + _normalize(component.lib_id), + _normalize(component.reference), + _normalize(component.value), + _normalize(component.footprint), + _normalize(component.datasheet), + _normalize(component.description), + _normalize(component.manufacturer), + _normalize(component.part_number), + _normalize(component.lcsc_id), + _normalize(component.source_path), + _normalize(component.category), + _normalize(component.component_type), + _normalize(component.package), + ), + ) + rowid = cursor.lastrowid + connection.execute( + """ + INSERT INTO symbols_fts ( + rowid, lib_id, reference, value, footprint, description, + manufacturer, part_number, lcsc_id, category, component_type, package + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + rowid, + component.lib_id, + component.reference, + component.value, + component.footprint, + component.description, + component.manufacturer, + component.part_number, + component.lcsc_id, + component.category, + component.component_type, + component.package, + ), + ) + + +def _row_to_component(row: sqlite3.Row) -> KiCadComponent: + return KiCadComponent( + reference=row["reference"], + value=row["value"], + footprint=row["footprint"], + datasheet=row["datasheet"], + description=row["description"], + manufacturer=row["manufacturer"], + part_number=row["part_number"], + lcsc_id=row["lcsc_id"], + lib_id=row["lib_id"], + source_path=row["source_path"], + category=row["category"], + component_type=row["component_type"], + unit_price=row["unit_price"], + stock=row["stock"], + ) + + +def build_symbol_sqlite_index( + cache_root: Optional[Path] = None, + *, + refresh: bool = False, +) -> Dict[str, Any]: + resolved_root = cache_root or ensure_container_library_cache(refresh=refresh) + if resolved_root is None: + return { + "index_path": None, + "symbol_libraries": 0, + "symbol_components": 0, + "source_count": 0, + } + + resolved_root = resolved_root.expanduser() + index_path = symbol_sqlite_index_path(resolved_root) + current_meta = _index_meta(resolved_root) + + if index_path.exists() and not refresh: + with _connect(index_path) as connection: + if _metadata_matches(connection, current_meta): + return symbol_sqlite_summary(resolved_root) + + fd, tmp_name = tempfile.mkstemp( + prefix=".symbol-index-", + suffix=".sqlite3", + dir=str(resolved_root), + ) + Path(tmp_name).unlink(missing_ok=True) + tmp_path = Path(tmp_name) + + try: + with _connect(tmp_path) as connection: + _create_schema(connection) + connection.execute( + "INSERT INTO metadata(key, value) VALUES('meta_json', ?)", + (json.dumps(current_meta, ensure_ascii=True),), + ) + + sources = sorted(container_symbols_dir(resolved_root).glob("*.kicad_sym")) + for source in sources: + connection.execute("INSERT INTO sources(path) VALUES (?)", (str(source),)) + for component in parse_inline_library_symbols(source): + _insert_component(connection, component) + connection.commit() + + tmp_path.replace(index_path) + return symbol_sqlite_summary(resolved_root) + finally: + tmp_path.unlink(missing_ok=True) + + +def symbol_sqlite_summary(cache_root: Optional[Path] = None) -> Dict[str, Any]: + resolved_root = cache_root or ensure_container_library_cache() + if resolved_root is None: + return { + "index_path": None, + "symbol_libraries": 0, + "symbol_components": 0, + "source_count": 0, + } + + index_path = symbol_sqlite_index_path(resolved_root) + if not index_path.exists(): + return build_symbol_sqlite_index(resolved_root, refresh=False) + + with _connect(index_path) as connection: + symbol_components = connection.execute("SELECT COUNT(*) AS count FROM symbols").fetchone()["count"] + symbol_libraries = connection.execute("SELECT COUNT(*) AS count FROM sources").fetchone()["count"] + return { + "index_path": str(index_path), + "symbol_libraries": symbol_libraries, + "symbol_components": symbol_components, + "source_count": symbol_libraries, + } + + +def load_all_symbol_components( + cache_root: Optional[Path] = None, + *, + refresh: bool = False, +) -> Tuple[List[KiCadComponent], List[Path]]: + resolved_root = cache_root or ensure_container_library_cache(refresh=refresh) + if resolved_root is None: + return [], [] + + build_symbol_sqlite_index(resolved_root, refresh=refresh) + index_path = symbol_sqlite_index_path(resolved_root) + with _connect(index_path) as connection: + rows = connection.execute("SELECT * FROM symbols ORDER BY lib_id").fetchall() + sources = [Path(row["path"]) for row in connection.execute("SELECT path FROM sources ORDER BY path")] + return [_row_to_component(row) for row in rows], sources + + +def source_sample(cache_root: Optional[Path] = None, limit: int = 10) -> List[str]: + resolved_root = cache_root or ensure_container_library_cache() + if resolved_root is None: + return [] + build_symbol_sqlite_index(resolved_root, refresh=False) + with _connect(symbol_sqlite_index_path(resolved_root)) as connection: + rows = connection.execute( + "SELECT path FROM sources ORDER BY path LIMIT ?", + (limit,), + ).fetchall() + return [row["path"] for row in rows] + + +def source_count(cache_root: Optional[Path] = None) -> int: + resolved_root = cache_root or ensure_container_library_cache() + if resolved_root is None: + return 0 + build_symbol_sqlite_index(resolved_root, refresh=False) + with _connect(symbol_sqlite_index_path(resolved_root)) as connection: + return connection.execute("SELECT COUNT(*) AS count FROM sources").fetchone()["count"] + + +def catalog_size(cache_root: Optional[Path] = None) -> int: + resolved_root = cache_root or ensure_container_library_cache() + if resolved_root is None: + return 0 + build_symbol_sqlite_index(resolved_root, refresh=False) + with _connect(symbol_sqlite_index_path(resolved_root)) as connection: + return connection.execute("SELECT COUNT(*) AS count FROM symbols").fetchone()["count"] + + +def resolve_symbol_component(identifier: str, cache_root: Optional[Path] = None) -> Optional[KiCadComponent]: + needle = _normalize(identifier) + if not needle: + return None + + resolved_root = cache_root or ensure_container_library_cache() + if resolved_root is None: + return None + build_symbol_sqlite_index(resolved_root, refresh=False) + + query = """ + SELECT * + FROM symbols + WHERE lib_id_lc = ? + OR reference_lc = ? + OR value_lc = ? + OR part_number_lc = ? + OR lcsc_id_lc = ? + OR (lib_id_lc || ':' || reference_lc) = ? + LIMIT 1 + """ + with _connect(symbol_sqlite_index_path(resolved_root)) as connection: + row = connection.execute(query, (needle, needle, needle, needle, needle, needle)).fetchone() + return _row_to_component(row) if row is not None else None + + +def _apply_spec_filter( + where_clauses: List[str], + params: List[Any], + score_terms: List[str], + score_params: List[Any], + key: str, + value: str, +) -> None: + normalized = _normalize(value) + if not normalized: + return + + if key in {"value", "resistance", "capacitance"}: + where_clauses.append("instr(value_lc, ?) > 0") + params.append(normalized) + score_terms.append("CASE WHEN value_lc = ? THEN 60 WHEN instr(value_lc, ?) > 0 THEN 30 ELSE 0 END") + score_params.extend([normalized, normalized]) + return + + if key in {"package"}: + where_clauses.append("instr(package_lc, ?) > 0") + params.append(normalized) + score_terms.append("CASE WHEN package_lc = ? THEN 35 WHEN instr(package_lc, ?) > 0 THEN 20 ELSE 0 END") + score_params.extend([normalized, normalized]) + return + + if key in {"footprint"}: + where_clauses.append("(instr(package_lc, ?) > 0 OR instr(footprint_lc, ?) > 0)") + params.extend([normalized, normalized]) + score_terms.append( + "CASE WHEN footprint_lc = ? THEN 35 WHEN instr(footprint_lc, ?) > 0 THEN 25 " + "WHEN package_lc = ? THEN 25 WHEN instr(package_lc, ?) > 0 THEN 15 ELSE 0 END" + ) + score_params.extend([normalized, normalized, normalized, normalized]) + return + + if key in {"manufacturer", "brand"}: + where_clauses.append("instr(manufacturer_lc, ?) > 0") + params.append(normalized) + score_terms.append( + "CASE WHEN manufacturer_lc = ? THEN 55 WHEN instr(manufacturer_lc, ?) > 0 THEN 25 ELSE 0 END" + ) + score_params.extend([normalized, normalized]) + return + + if key in {"part", "part_number", "mpn"}: + where_clauses.append("instr(part_number_lc, ?) > 0") + params.append(normalized) + score_terms.append( + "CASE WHEN part_number_lc = ? THEN 90 WHEN instr(part_number_lc, ?) > 0 THEN 45 ELSE 0 END" + ) + score_params.extend([normalized, normalized]) + return + + if key in {"lcsc", "lcsc_id"}: + where_clauses.append("instr(lcsc_id_lc, ?) > 0") + params.append(normalized) + score_terms.append( + "CASE WHEN lcsc_id_lc = ? THEN 90 WHEN instr(lcsc_id_lc, ?) > 0 THEN 45 ELSE 0 END" + ) + score_params.extend([normalized, normalized]) + return + + if key in {"type", "category"}: + where_clauses.append("(instr(component_type_lc, ?) > 0 OR instr(category_lc, ?) > 0)") + params.extend([normalized, normalized]) + score_terms.append( + "CASE WHEN component_type_lc = ? THEN 70 WHEN instr(component_type_lc, ?) > 0 THEN 35 " + "WHEN instr(category_lc, ?) > 0 THEN 20 ELSE 0 END" + ) + score_params.extend([normalized, normalized, normalized]) + return + + where_clauses.append( + "id IN (SELECT rowid FROM symbols_fts WHERE symbols_fts MATCH ?)" + ) + params.append(_fts_term(normalized)) + score_terms.append("20") + + +def search_symbol_components( + component_type: str, + specs: Dict[str, Any], + *, + limit: int = 20, + cache_root: Optional[Path] = None, +) -> List[KiCadComponent]: + resolved_root = cache_root or ensure_container_library_cache() + if resolved_root is None: + return [] + build_symbol_sqlite_index(resolved_root, refresh=False) + + where_clauses = ["1 = 1"] + params: List[Any] = [] + score_terms: List[str] = [] + score_params: List[Any] = [] + + wanted_type = _normalize(component_type) + if wanted_type: + where_clauses.append("(component_type_lc = ? OR instr(lib_id_lc, ?) > 0)") + params.extend([wanted_type, wanted_type]) + score_terms.append( + "CASE WHEN component_type_lc = ? THEN 80 WHEN instr(lib_id_lc, ?) > 0 THEN 40 ELSE 0 END" + ) + score_params.extend([wanted_type, wanted_type]) + + for key, value in specs.items(): + _apply_spec_filter(where_clauses, params, score_terms, score_params, key.lower(), str(value)) + + score_sql = " + ".join(score_terms) if score_terms else "0" + sql = f""" + SELECT *, + ({score_sql}) AS score + FROM symbols + WHERE {' AND '.join(where_clauses)} + ORDER BY score DESC, lib_id ASC + LIMIT ? + """ + + with _connect(symbol_sqlite_index_path(resolved_root)) as connection: + rows = connection.execute(sql, tuple(score_params + params + [max(limit * 10, limit)])).fetchall() + + return [_row_to_component(row) for row in rows[:limit]] diff --git a/mcp_servers/kicad_text.py b/mcp_servers/kicad_text.py new file mode 100644 index 0000000..2778cf4 --- /dev/null +++ b/mcp_servers/kicad_text.py @@ -0,0 +1,871 @@ +#!/usr/bin/env python3 +""" +Shared KiCad text parsing helpers for lightweight MCP servers. + +These helpers intentionally work without pcbnew or KiCad system libraries. +They read `.kicad_sch` and `.kicad_pcb` files directly and only report facts +that can be derived from the available local files. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Dict, Iterable, List, Optional, Sequence, Tuple +import math +import json +import os +import re +import shutil +import subprocess + + +DEFAULT_KICAD_V10_IMAGE = "kill_life_cad-kicad-mcp:latest" +_CONTAINER_CACHE_STATE: Optional[Path] = None +_CONTAINER_CACHE_FAILED = False +_CONTAINER_FOOTPRINT_INDEX_CACHE: Optional[Tuple[str, int, Dict[str, Path]]] = None +_FOOTPRINT_NAME_CACHE: Dict[str, Tuple[str, ...]] = {} +INDEX_VERSION = 1 +SYMBOL_INDEX_FILENAME = ".symbol_index.sqlite3" +FOOTPRINT_INDEX_FILENAME = ".footprint_index.json" + + +PROPERTY_PATTERN = re.compile(r'\(property\s+"([^"]+)"\s+"([^"]*)"') +LIB_ID_PATTERN = re.compile(r'\(lib_id\s+"([^"]+)"') +QUOTED_SYMBOL_PATTERN = re.compile(r'^\(symbol\s+"([^"]+)"') +LABEL_PATTERN = re.compile(r'\((?:global_label|label|hierarchical_label)\s+"([^"]+)"') +TITLE_PATTERN = re.compile(r'\(title\s+"([^"]*)"') +LAYER_PATTERN = re.compile(r'\(\s*\d+\s+"([^"]+)"\s+([^)]+)\)') +EDGE_LINE_PATTERN = re.compile( + r'\(gr_line\b.*?\(start\s+([-0-9.]+)\s+([-0-9.]+)\)\s*' + r'\(end\s+([-0-9.]+)\s+([-0-9.]+)\).*?' + r'\(layer\s+"Edge\.Cuts"\)', + re.DOTALL, +) +EDGE_RECT_PATTERN = re.compile( + r'\(gr_rect\b.*?\(start\s+([-0-9.]+)\s+([-0-9.]+)\)\s*' + r'\(end\s+([-0-9.]+)\s+([-0-9.]+)\).*?' + r'\(layer\s+"Edge\.Cuts"\)', + re.DOTALL, +) +EDGE_CIRCLE_PATTERN = re.compile( + r'\(gr_circle\b.*?\(center\s+([-0-9.]+)\s+([-0-9.]+)\)\s*' + r'\(end\s+([-0-9.]+)\s+([-0-9.]+)\).*?' + r'\(layer\s+"Edge\.Cuts"\)', + re.DOTALL, +) +EDGE_ARC_PATTERN = re.compile( + r'\(gr_arc\b.*?\(start\s+([-0-9.]+)\s+([-0-9.]+)\)\s*' + r'\(mid\s+([-0-9.]+)\s+([-0-9.]+)\)\s*' + r'\(end\s+([-0-9.]+)\s+([-0-9.]+)\).*?' + r'\(layer\s+"Edge\.Cuts"\)', + re.DOTALL, +) + + +def read_text(path: Path) -> str: + return path.read_text(encoding="utf-8", errors="ignore") + + +def env_is_enabled(name: str, default: bool = True) -> bool: + value = os.getenv(name) + if value is None: + return default + return value.strip().lower() not in {"0", "false", "no", "off", ""} + + +def container_library_cache_root() -> Path: + override = os.getenv("KICAD_AUX_CONTAINER_CACHE_DIR") + if override: + return Path(override).expanduser() + return Path.home() / "Kill_LIFE" / ".cad-home" / "kicad-mcp" / "kicad-v10-libs" + + +def container_symbols_dir(cache_root: Optional[Path] = None) -> Path: + root = cache_root or container_library_cache_root() + return root / "symbols" + + +def container_footprints_dir(cache_root: Optional[Path] = None) -> Path: + root = cache_root or container_library_cache_root() + return root / "footprints" + + +def symbol_index_path(cache_root: Optional[Path] = None) -> Path: + root = cache_root or container_library_cache_root() + return root / SYMBOL_INDEX_FILENAME + + +def footprint_index_path(cache_root: Optional[Path] = None) -> Path: + root = cache_root or container_library_cache_root() + return root / FOOTPRINT_INDEX_FILENAME + + +def ensure_container_library_cache(refresh: bool = False) -> Optional[Path]: + global _CONTAINER_CACHE_STATE, _CONTAINER_CACHE_FAILED + + if _CONTAINER_CACHE_STATE is not None and not refresh: + return _CONTAINER_CACHE_STATE + if _CONTAINER_CACHE_FAILED and not refresh: + return None + if not env_is_enabled("KICAD_AUX_USE_CONTAINER_LIBS", True): + _CONTAINER_CACHE_FAILED = True + return None + if shutil.which("docker") is None: + _CONTAINER_CACHE_FAILED = True + return None + + image = os.getenv("KICAD_MCP_IMAGE", DEFAULT_KICAD_V10_IMAGE) + cache_root = container_library_cache_root() + symbols_dir = cache_root / "symbols" + footprints_dir = cache_root / "footprints" + + if not refresh and symbols_dir.is_dir() and footprints_dir.is_dir(): + _CONTAINER_CACHE_STATE = cache_root + return cache_root + + if subprocess.run( + ["docker", "image", "inspect", image], + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ).returncode != 0: + _CONTAINER_CACHE_FAILED = True + return None + + container_id = "" + tmp_root = cache_root.with_name(f".{cache_root.name}.tmp") + try: + if tmp_root.exists(): + shutil.rmtree(tmp_root) + tmp_root.parent.mkdir(parents=True, exist_ok=True) + tmp_root.mkdir(parents=True, exist_ok=True) + + container_id = subprocess.check_output(["docker", "create", image], text=True).strip() + subprocess.run( + ["docker", "cp", f"{container_id}:/usr/share/kicad/symbols", str(tmp_root)], + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + subprocess.run( + ["docker", "cp", f"{container_id}:/usr/share/kicad/footprints", str(tmp_root)], + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + if cache_root.exists(): + shutil.rmtree(cache_root) + tmp_root.rename(cache_root) + _CONTAINER_CACHE_STATE = cache_root + _CONTAINER_CACHE_FAILED = False + return cache_root + except (subprocess.CalledProcessError, OSError): + _CONTAINER_CACHE_FAILED = True + return None + finally: + if container_id: + subprocess.run( + ["docker", "rm", "-f", container_id], + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + if tmp_root.exists(): + shutil.rmtree(tmp_root, ignore_errors=True) + + +def _safe_relpath(path: Path, root: Path) -> str: + try: + return str(path.resolve().relative_to(root.resolve())) + except ValueError: + return str(path) + + +def _footprint_index_metadata(cache_root: Path) -> Dict[str, object]: + footprints_dir = container_footprints_dir(cache_root) + dirs = sorted(footprints_dir.glob("*.pretty")) + return { + "version": INDEX_VERSION, + "footprints_dir": str(footprints_dir), + "footprints_dir_mtime_ns": footprints_dir.stat().st_mtime_ns if footprints_dir.exists() else None, + "library_count": len(dirs), + "library_names": [path.name for path in dirs], + } + + +def _metadata_matches(current: Dict[str, object], cached: Dict[str, object]) -> bool: + for key, value in current.items(): + if cached.get(key) != value: + return False + return True + + +def load_container_symbol_index( + cache_root: Optional[Path] = None, + *, + refresh: bool = False, +) -> Tuple[List["KiCadComponent"], List[Path]]: + try: + from .kicad_symbol_sqlite import load_all_symbol_components + except ImportError: + from kicad_symbol_sqlite import load_all_symbol_components # type: ignore + + return load_all_symbol_components(cache_root, refresh=refresh) + + +def load_container_footprint_index( + cache_root: Optional[Path] = None, + *, + refresh: bool = False, +) -> Dict[str, Path]: + global _CONTAINER_FOOTPRINT_INDEX_CACHE + + resolved_root = cache_root or ensure_container_library_cache(refresh=refresh) + if resolved_root is None: + return {} + + resolved_root = resolved_root.expanduser() + current_meta = _footprint_index_metadata(resolved_root) + cache_signature = int(current_meta.get("footprints_dir_mtime_ns") or 0) + cache_key = str(resolved_root) + + if ( + _CONTAINER_FOOTPRINT_INDEX_CACHE is not None + and _CONTAINER_FOOTPRINT_INDEX_CACHE[0] == cache_key + and _CONTAINER_FOOTPRINT_INDEX_CACHE[1] == cache_signature + and not refresh + ): + return _CONTAINER_FOOTPRINT_INDEX_CACHE[2] + + index_file = footprint_index_path(resolved_root) + if index_file.exists() and not refresh: + try: + payload = json.loads(index_file.read_text(encoding="utf-8")) + cached_meta = payload.get("meta", {}) + if isinstance(cached_meta, dict) and _metadata_matches(current_meta, cached_meta): + libraries: Dict[str, Path] = {} + _FOOTPRINT_NAME_CACHE.clear() + for library_name, entry in payload.get("libraries", {}).items(): + relative_path = entry.get("relative_path") + if not relative_path: + continue + full_path = resolved_root / relative_path + libraries[library_name] = full_path + _FOOTPRINT_NAME_CACHE[str(full_path)] = tuple(entry.get("footprints", [])) + _CONTAINER_FOOTPRINT_INDEX_CACHE = (cache_key, cache_signature, libraries) + return libraries + except (OSError, json.JSONDecodeError, TypeError, ValueError): + pass + + libraries: Dict[str, Path] = {} + serialized: Dict[str, Dict[str, object]] = {} + _FOOTPRINT_NAME_CACHE.clear() + footprints_dir = container_footprints_dir(resolved_root) + for library_path in sorted(footprints_dir.glob("*.pretty")): + if not library_path.is_dir(): + continue + library_name = library_path.stem + if library_name in libraries: + continue + footprint_names = tuple(sorted(path.stem for path in library_path.glob("*.kicad_mod"))) + libraries[library_name] = library_path + _FOOTPRINT_NAME_CACHE[str(library_path)] = footprint_names + serialized[library_name] = { + "relative_path": _safe_relpath(library_path, resolved_root), + "footprints": list(footprint_names), + } + + payload = {"meta": current_meta, "libraries": serialized} + index_file.write_text(json.dumps(payload, ensure_ascii=True), encoding="utf-8") + _CONTAINER_FOOTPRINT_INDEX_CACHE = (cache_key, cache_signature, libraries) + return libraries + + +def prewarm_container_indexes(refresh: bool = False) -> Dict[str, object]: + cache_root = ensure_container_library_cache(refresh=refresh) + if cache_root is None: + return { + "cache_root": None, + "symbol_libraries": 0, + "symbol_components": 0, + "footprint_libraries": 0, + } + + try: + from .kicad_symbol_sqlite import build_symbol_sqlite_index + except ImportError: + from kicad_symbol_sqlite import build_symbol_sqlite_index # type: ignore + + symbol_summary = build_symbol_sqlite_index(cache_root, refresh=refresh) + footprint_libraries = load_container_footprint_index(cache_root, refresh=refresh) + return { + "cache_root": str(cache_root), + "symbol_libraries": symbol_summary.get("symbol_libraries", 0), + "symbol_components": symbol_summary.get("symbol_components", 0), + "footprint_libraries": len(footprint_libraries), + } + + +def clear_container_indexes(cache_root: Optional[Path] = None) -> None: + global _CONTAINER_FOOTPRINT_INDEX_CACHE + + resolved_root = cache_root or container_library_cache_root() + legacy_symbol_index = resolved_root / ".symbol_index.json" + for index_file in (symbol_index_path(resolved_root), footprint_index_path(resolved_root)): + try: + index_file.unlink() + except FileNotFoundError: + pass + try: + legacy_symbol_index.unlink() + except FileNotFoundError: + pass + _CONTAINER_FOOTPRINT_INDEX_CACHE = None + + +def extract_blocks(text: str, keyword: str) -> List[str]: + """Return every S-expression block whose head is `keyword`.""" + target = f"({keyword}" + blocks: List[str] = [] + start = 0 + + while True: + idx = text.find(target, start) + if idx == -1: + return blocks + + depth = 0 + in_string = False + escaped = False + + for pos in range(idx, len(text)): + char = text[pos] + if in_string: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == '"': + in_string = False + continue + + if char == '"': + in_string = True + continue + if char == "(": + depth += 1 + elif char == ")": + depth -= 1 + if depth == 0: + blocks.append(text[idx : pos + 1]) + start = pos + 1 + break + else: + return blocks + + +def parse_properties(block: str) -> Dict[str, str]: + return {match.group(1): match.group(2) for match in PROPERTY_PATTERN.finditer(block)} + + +def parse_float(value: Optional[str]) -> Optional[float]: + if value in (None, ""): + return None + cleaned = str(value).strip().replace(",", ".") + match = re.search(r"-?\d+(?:\.\d+)?", cleaned) + if not match: + return None + try: + return float(match.group(0)) + except ValueError: + return None + + +def parse_int(value: Optional[str]) -> Optional[int]: + if value in (None, ""): + return None + cleaned = str(value).strip().replace(",", "") + match = re.search(r"-?\d+", cleaned) + if not match: + return None + try: + return int(match.group(0)) + except ValueError: + return None + + +def derive_package(footprint: str) -> str: + if not footprint: + return "" + footprint_name = footprint.split(":", 1)[-1] + match = re.search( + r"(0201|0402|0603|0805|1206|1210|SOT-23(?:-[0-9]+)?|SOT-223|" + r"SOIC-[0-9]+|TSSOP-[0-9]+|QFN-[0-9]+|TQFP-[0-9]+|LQFP-[0-9]+|" + r"DIP-[0-9]+|PinHeader_[^:_]+|JST_[^:_]+)", + footprint_name, + re.IGNORECASE, + ) + return match.group(1) if match else footprint_name + + +def infer_component_type( + reference: str = "", + lib_id: str = "", + value: str = "", + description: str = "", + category: str = "", +) -> str: + reference = (reference or "").upper() + lib_id_lower = (lib_id or "").lower() + haystack = " ".join([lib_id_lower, value.lower(), description.lower(), category.lower()]) + + if "regulator" in haystack or "ldo" in haystack: + return "regulator" + if "microcontroller" in haystack or "mcu" in haystack: + return "microcontroller" + if "opamp" in haystack or "amplifier" in haystack: + return "opamp" + if "connector" in haystack or lib_id_lower.startswith("connector"): + return "connector" + if "switch" in haystack: + return "switch" + if "mosfet" in haystack: + return "mosfet" + if "transistor" in haystack or lib_id_lower.startswith("device:q_"): + return "transistor" + if "crystal" in haystack or "oscillator" in haystack: + return "crystal" + if "inductor" in haystack or lib_id_lower.startswith("device:l"): + return "inductor" + if "capacitor" in haystack or lib_id_lower.startswith("device:c"): + return "capacitor" + if "resistor" in haystack or lib_id_lower.startswith("device:r"): + return "resistor" + if "led" in haystack: + return "led" + if "diode" in haystack or lib_id_lower.startswith("device:d"): + return "diode" + if "power" in haystack or lib_id_lower.startswith("power:"): + return "power" + + if reference.startswith("R"): + return "resistor" + if reference.startswith("C"): + return "capacitor" + if reference.startswith("L"): + return "inductor" + if reference.startswith("D"): + return "diode" + if reference.startswith("Q"): + return "transistor" + if reference.startswith("U"): + return "ic" + if reference.startswith("J"): + return "connector" + if reference.startswith("Y"): + return "crystal" + if reference.startswith("SW"): + return "switch" + + return "other" + + +def derive_manufacturer(lib_id: str, manufacturer: str) -> str: + if manufacturer: + return manufacturer + library_name = lib_id.split(":", 1)[0] + parts = [part for part in library_name.split("_") if part] + if len(parts) >= 2 and parts[0] in {"MCU", "CPU", "DSP", "FPGA", "CPLD"}: + aliases = { + "ST": "STMicroelectronics", + "NXP": "NXP", + "TI": "Texas Instruments", + } + return aliases.get(parts[1], parts[1]) + return manufacturer + + +def derive_category(lib_id: str, category: str) -> str: + if category: + return category + return lib_id.split(":", 1)[0].replace("_", " ") + + +@dataclass +class KiCadComponent: + reference: str + value: str + footprint: str + datasheet: str + description: str + manufacturer: str + part_number: str + lcsc_id: str + lib_id: str + source_path: str + category: str + component_type: str + unit_price: Optional[float] = None + stock: Optional[int] = None + + @property + def package(self) -> str: + return derive_package(self.footprint) + + def to_dict( + self, + include_values: bool = True, + include_footprints: bool = True, + ) -> Dict[str, object]: + result = asdict(self) + result["type"] = result.pop("component_type") + result["package"] = self.package if include_footprints else "" + result["value"] = self.value if include_values else None + result["footprint"] = self.footprint if include_footprints else None + result["price"] = self.unit_price + return result + + +def _build_component( + *, + reference: str, + value: str, + footprint: str, + datasheet: str, + description: str, + manufacturer: str, + part_number: str, + lcsc_id: str, + lib_id: str, + source_path: Path, + category: str, + price_raw: str, + stock_raw: str, +) -> KiCadComponent: + component_type = infer_component_type( + reference=reference, + lib_id=lib_id, + value=value, + description=description, + category=category, + ) + return KiCadComponent( + reference=reference, + value=value, + footprint=footprint, + datasheet=datasheet, + description=description, + manufacturer=derive_manufacturer(lib_id, manufacturer), + part_number=part_number, + lcsc_id=lcsc_id, + lib_id=lib_id, + source_path=str(source_path), + category=derive_category(lib_id, category), + component_type=component_type, + unit_price=parse_float(price_raw), + stock=parse_int(stock_raw), + ) + + +def parse_inline_library_symbols(path: Path) -> List[KiCadComponent]: + text = read_text(path) + components: List[KiCadComponent] = [] + for block in extract_blocks(text, "symbol"): + match = QUOTED_SYMBOL_PATTERN.match(block) + if not match: + continue + symbol_name = match.group(1) + if re.search(r"_\d+_\d+$", symbol_name): + continue + if path.suffix == ".kicad_sym" and ":" not in symbol_name: + symbol_ref = f"{path.stem}:{symbol_name}" + else: + symbol_ref = symbol_name + properties = parse_properties(block) + components.append( + _build_component( + reference=properties.get("Reference", symbol_ref.split(":")[-1]), + value=properties.get("Value", symbol_ref.split(":")[-1]), + footprint=properties.get("Footprint", ""), + datasheet=properties.get("Datasheet", ""), + description=properties.get("Description", ""), + manufacturer=properties.get("Manufacturer", ""), + part_number=properties.get( + "Part", + properties.get("MPN", symbol_ref.split(":")[-1]), + ), + lcsc_id=properties.get("LCSC", properties.get("LCSC Part", "")), + lib_id=symbol_ref, + source_path=path, + category=properties.get("Category", ""), + price_raw=properties.get("Price", ""), + stock_raw=properties.get("Stock", ""), + ) + ) + return components + + +def parse_schematic_components( + path: Path, + *, + include_templates: bool = False, +) -> List[KiCadComponent]: + text = read_text(path) + components: List[KiCadComponent] = [] + for block in extract_blocks(text, "symbol"): + lib_id_match = LIB_ID_PATTERN.search(block) + if not lib_id_match: + continue + properties = parse_properties(block) + reference = properties.get("Reference", "") + if not reference: + continue + if not include_templates and reference.startswith("_TEMPLATE"): + continue + components.append( + _build_component( + reference=reference, + value=properties.get("Value", lib_id_match.group(1).split(":")[-1]), + footprint=properties.get("Footprint", ""), + datasheet=properties.get("Datasheet", ""), + description=properties.get("Description", ""), + manufacturer=properties.get("Manufacturer", ""), + part_number=properties.get( + "Part", + properties.get("MPN", lib_id_match.group(1).split(":")[-1]), + ), + lcsc_id=properties.get("LCSC", properties.get("LCSC Part", "")), + lib_id=lib_id_match.group(1), + source_path=path, + category=properties.get("Category", ""), + price_raw=properties.get("Price", ""), + stock_raw=properties.get("Stock", ""), + ) + ) + return sorted(components, key=lambda item: item.reference) + + +def parse_schematic_metrics(path: Path) -> Dict[str, object]: + text = read_text(path) + labels = sorted(set(LABEL_PATTERN.findall(text))) + return { + "wire_count": text.count("(wire "), + "junction_count": text.count("(junction "), + "label_count": len(labels), + "labels": labels, + "sheet_count": text.count("(sheet "), + } + + +def _extract_layer_block(text: str) -> str: + blocks = extract_blocks(text, "layers") + return blocks[0] if blocks else "" + + +def _edge_cut_points(text: str) -> List[Tuple[float, float]]: + points: List[Tuple[float, float]] = [] + + for match in EDGE_LINE_PATTERN.finditer(text): + x1, y1, x2, y2 = map(float, match.groups()) + points.extend([(x1, y1), (x2, y2)]) + + for match in EDGE_RECT_PATTERN.finditer(text): + x1, y1, x2, y2 = map(float, match.groups()) + points.extend([(x1, y1), (x2, y2)]) + + for match in EDGE_CIRCLE_PATTERN.finditer(text): + cx, cy, ex, ey = map(float, match.groups()) + radius = math.dist((cx, cy), (ex, ey)) + points.extend( + [ + (cx - radius, cy - radius), + (cx + radius, cy + radius), + ] + ) + + for match in EDGE_ARC_PATTERN.finditer(text): + coords = list(map(float, match.groups())) + points.extend([(coords[0], coords[1]), (coords[2], coords[3]), (coords[4], coords[5])]) + + return points + + +def parse_pcb_summary(path: Path) -> Dict[str, object]: + text = read_text(path) + layer_block = _extract_layer_block(text) + copper_layers: List[str] = [] + for name, layer_type in LAYER_PATTERN.findall(layer_block): + if name.endswith(".Cu") or "signal" in layer_type: + copper_layers.append(name) + + points = _edge_cut_points(text) + width_mm = None + height_mm = None + if points: + xs = [point[0] for point in points] + ys = [point[1] for point in points] + width_mm = round(max(xs) - min(xs), 3) + height_mm = round(max(ys) - min(ys), 3) + + title_match = TITLE_PATTERN.search(text) + return { + "title": title_match.group(1) if title_match else path.stem, + "copper_layers": copper_layers, + "footprint_count": len(re.findall(r"\(footprint\b", text)), + "segment_count": len(re.findall(r"\(segment\b", text)), + "via_count": len(re.findall(r"\(via\b", text)), + "zone_count": len(re.findall(r"\(zone\b", text)), + "net_count": len(set(re.findall(r'\(net\s+\d+\s+"([^"]*)"', text))), + "edge_cuts_count": text.count('Edge.Cuts'), + "width_mm": width_mm, + "height_mm": height_mm, + } + + +def discover_search_roots(base_dir: Path, env_var: str = "KICAD_AUX_MCP_ROOTS") -> List[Path]: + extra_roots: List[Path] = [] + env_value = os.getenv(env_var, "") + if env_value: + extra_roots.extend(Path(entry).expanduser() for entry in env_value.split(os.pathsep) if entry) + + sibling_templates = base_dir.parents[1] / "kicad_mcp_server" / "python" / "templates" + container_cache_root = ensure_container_library_cache( + refresh=env_is_enabled("KICAD_AUX_CONTAINER_REFRESH", False) + ) + default_roots = [ + Path.cwd(), + base_dir.parents[1], + sibling_templates, + Path.home() / "Documents" / "KiCad", + Path.home() / ".local" / "share" / "kicad", + Path.home() / "Kill_LIFE" / ".cad-home" / "kicad-mcp", + container_cache_root, + ] + + roots: List[Path] = [] + seen = set() + for root in [*extra_roots, *default_roots]: + if root is None: + continue + root = root.expanduser() + key = str(root.resolve()) if root.exists() else str(root) + if key in seen or not root.exists(): + continue + seen.add(key) + roots.append(root) + return roots + + +def discover_schematic_files(roots: Sequence[Path], limit: int = 64) -> List[Path]: + files: List[Path] = [] + seen = set() + for root in roots: + for path in root.rglob("*.kicad_sch"): + key = str(path.resolve()) + if key in seen: + continue + seen.add(key) + files.append(path) + if len(files) >= limit: + return files + return files + + +def discover_symbol_library_files(roots: Sequence[Path], limit: int = 512) -> List[Path]: + files: List[Path] = [] + seen = set() + for root in roots: + for path in root.rglob("*.kicad_sym"): + key = str(path.resolve()) + if key in seen: + continue + seen.add(key) + files.append(path) + if len(files) >= limit: + return files + return files + + +def discover_footprint_libraries(roots: Sequence[Path], limit: int = 256) -> Dict[str, Path]: + libraries: Dict[str, Path] = {} + container_root = ensure_container_library_cache( + refresh=env_is_enabled("KICAD_AUX_CONTAINER_REFRESH", False) + ) + container_libraries_loaded = False + for root in roots: + if container_root is not None: + try: + root_resolved = root.resolve() + container_resolved = container_root.resolve() + if root_resolved == container_resolved or container_resolved in root_resolved.parents: + if not container_libraries_loaded: + for name, path in load_container_footprint_index(container_root).items(): + libraries.setdefault(name, path) + if len(libraries) >= limit: + return libraries + container_libraries_loaded = True + continue + except FileNotFoundError: + pass + for path in root.rglob("*.pretty"): + if not path.is_dir(): + continue + libraries.setdefault(path.stem, path) + if len(libraries) >= limit: + return libraries + return libraries + + +def footprint_exists(footprint_spec: str, libraries: Dict[str, Path]) -> bool: + if not footprint_spec: + return False + if ":" in footprint_spec: + library_name, footprint_name = footprint_spec.split(":", 1) + library_path = libraries.get(library_name) + if not library_path: + return False + cached_names = _FOOTPRINT_NAME_CACHE.get(str(library_path)) + if cached_names: + return footprint_name in cached_names + return (library_path / f"{footprint_name}.kicad_mod").exists() + + footprint_name = footprint_spec + for path in libraries.values(): + cached_names = _FOOTPRINT_NAME_CACHE.get(str(path)) + if cached_names: + if footprint_name in cached_names: + return True + continue + if (path / f"{footprint_name}.kicad_mod").exists(): + return True + return False + + +def search_footprints( + libraries: Dict[str, Path], + patterns: Iterable[str], + limit: int = 5, +) -> List[str]: + results: List[str] = [] + seen = set() + lowered_patterns = [pattern.lower() for pattern in patterns if pattern] + + for library_name, library_path in libraries.items(): + cached_names = _FOOTPRINT_NAME_CACHE.get(str(library_path)) + if cached_names: + footprint_names = list(cached_names) + else: + footprint_names = [candidate.stem for candidate in library_path.glob("*.kicad_mod")] + for footprint_name in footprint_names: + search_text = f"{library_name}:{footprint_name}".lower() + if lowered_patterns and not any(pattern in search_text for pattern in lowered_patterns): + continue + full_name = f"{library_name}:{footprint_name}" + if full_name in seen: + continue + seen.add(full_name) + results.append(full_name) + if len(results) >= limit: + return results + return results diff --git a/mcp_servers/kicad_tools.py b/mcp_servers/kicad_tools.py index e7d90d5..6144ce3 100644 --- a/mcp_servers/kicad_tools.py +++ b/mcp_servers/kicad_tools.py @@ -1,44 +1,538 @@ #!/usr/bin/env python3 """ -KiCad Tools MCP Server -Provides tools for manipulating KiCad designs and schematics +KiCad Tools MCP Server. + +This server stays compatible with the existing KIC-AI plugin transport +(one JSON-RPC object per line) while replacing canned answers with +real file-backed analysis of KiCad schematics and PCB files. """ +from __future__ import annotations + +import asyncio import json import sys -import asyncio -import os -from typing import Dict, List, Any, Optional +from collections import Counter, defaultdict +from pathlib import Path +from typing import Any, Dict, List, Optional, Sequence + +try: + from .kicad_text import ( + KiCadComponent, + discover_footprint_libraries, + discover_search_roots, + footprint_exists, + parse_pcb_summary, + parse_schematic_components, + parse_schematic_metrics, + search_footprints, + ) +except ImportError: + from kicad_text import ( # type: ignore + KiCadComponent, + discover_footprint_libraries, + discover_search_roots, + footprint_exists, + parse_pcb_summary, + parse_schematic_components, + parse_schematic_metrics, + search_footprints, + ) + class KiCadToolsServer: - """MCP server for KiCad design manipulation tools""" - + """MCP server for analyzing KiCad design files without pcbnew.""" + def __init__(self): - self.current_project_path = None - + self.base_dir = Path(__file__).resolve() + self.search_roots = discover_search_roots(self.base_dir) + + def _check_file(self, path_value: str, expected_suffix: str) -> Path: + if not path_value: + raise FileNotFoundError(f"Missing path for {expected_suffix}") + path = Path(path_value).expanduser() + if not path.exists(): + raise FileNotFoundError(f"File not found: {path}") + if path.suffix != expected_suffix: + raise ValueError(f"Expected {expected_suffix} file, got {path.suffix}: {path}") + return path + + def _resolve_project_files(self, project_path: str) -> Dict[str, Optional[Path]]: + path = Path(project_path).expanduser() + if not path.exists(): + raise FileNotFoundError(f"Project path not found: {path}") + + directory = path if path.is_dir() else path.parent + stem = path.stem if path.is_file() else path.name + + schematic = directory / f"{stem}.kicad_sch" + pcb = directory / f"{stem}.kicad_pcb" + project = directory / f"{stem}.kicad_pro" + + if path.suffix == ".kicad_sch": + schematic = path + elif path.suffix == ".kicad_pcb": + pcb = path + elif path.suffix == ".kicad_pro": + project = path + + return { + "directory": directory, + "schematic": schematic if schematic.exists() else None, + "pcb": pcb if pcb.exists() else None, + "project": project if project.exists() else None, + } + + def _summarize_components(self, components: Sequence[KiCadComponent]) -> Dict[str, int]: + return dict(sorted(Counter(component.component_type for component in components).items())) + + def _schematic_issues( + self, + components: Sequence[KiCadComponent], + metrics: Dict[str, Any], + ) -> List[Dict[str, Any]]: + issues: List[Dict[str, Any]] = [] + references = [component.reference for component in components] + duplicates = [ref for ref, count in Counter(references).items() if count > 1] + + if not components: + issues.append( + { + "type": "error", + "category": "components", + "message": "No placed components were detected in the schematic file.", + "suggestion": "Place symbols in the schematic or pass the correct .kicad_sch path.", + } + ) + + if metrics["wire_count"] == 0 and len(components) > 1: + issues.append( + { + "type": "warning", + "category": "connectivity", + "message": "The schematic contains multiple components but no wires.", + "suggestion": "Add wires or labels to express connectivity.", + } + ) + + power_labels = { + label.upper() + for label in metrics["labels"] + if label.upper() in {"VCC", "3V3", "5V", "VBAT", "VIN", "GND"} + } + if ( + not any(component.component_type == "power" for component in components) + and metrics["label_count"] > 0 + and not power_labels + ): + issues.append( + { + "type": "warning", + "category": "power", + "message": "No explicit power symbols were detected.", + "suggestion": "Verify that supply rails are represented with power symbols or labels.", + } + ) + + for reference in duplicates: + issues.append( + { + "type": "error", + "category": "components", + "component": reference, + "message": f"Duplicate component reference detected: {reference}", + "suggestion": "Re-annotate the schematic so each component reference is unique.", + } + ) + + for component in components: + if not component.value: + issues.append( + { + "type": "warning", + "category": "components", + "component": component.reference, + "message": f"{component.reference} has no value property.", + "suggestion": "Set a component value before generating a BOM.", + } + ) + if not component.footprint: + issues.append( + { + "type": "warning", + "category": "components", + "component": component.reference, + "message": f"{component.reference} has no footprint assigned.", + "suggestion": "Assign a footprint before PCB layout or fabrication export.", + } + ) + + return issues + + def _filter_issues(self, issues: Sequence[Dict[str, Any]], allowed: Sequence[str]) -> List[Dict[str, Any]]: + if not allowed: + return list(issues) + allowed_lower = {entry.lower() for entry in allowed} + return [issue for issue in issues if issue.get("category", "").lower() in allowed_lower] + + async def _analyze_schematic(self, args: Dict[str, Any]) -> Dict[str, Any]: + schematic_path = self._check_file(args.get("schematic_path"), ".kicad_sch") + check_types = args.get("check_types", ["erc", "power", "connectivity", "components"]) + + components = parse_schematic_components(schematic_path) + metrics = parse_schematic_metrics(schematic_path) + issues = self._filter_issues(self._schematic_issues(components, metrics), check_types) + warnings = sum(1 for issue in issues if issue.get("type") == "warning") + errors = sum(1 for issue in issues if issue.get("type") == "error") + power_labels = [label for label in metrics["labels"] if label.upper() in {"VCC", "3V3", "5V", "VBAT", "VIN", "GND"}] + + return { + "file_path": str(schematic_path), + "analysis_summary": { + "total_components": len(components), + "total_nets_estimate": max(metrics["wire_count"], metrics["label_count"]), + "warnings": warnings, + "errors": errors, + "wires": metrics["wire_count"], + "labels": metrics["label_count"], + }, + "issues": issues, + "power_analysis": { + "power_symbols": sum(1 for component in components if component.component_type == "power"), + "power_rails": sorted(set(power_labels)), + "status": "inferred-from-file", + }, + "component_summary": self._summarize_components(components), + "data_mode": "file-backed-local", + } + + async def _analyze_pcb(self, args: Dict[str, Any]) -> Dict[str, Any]: + pcb_path = self._check_file(args.get("pcb_path"), ".kicad_pcb") + check_types = args.get("check_types", ["drc", "layer_stack", "thermal", "impedance"]) + summary = parse_pcb_summary(pcb_path) + + issues: List[Dict[str, Any]] = [] + copper_layers = summary["copper_layers"] + if summary["edge_cuts_count"] == 0: + issues.append( + { + "type": "violation", + "category": "drc", + "rule": "board_outline", + "message": "No Edge.Cuts geometry was detected in the board file.", + "severity": "error", + } + ) + if summary["footprint_count"] > 1 and summary["segment_count"] == 0: + issues.append( + { + "type": "warning", + "category": "drc", + "rule": "routing", + "message": "The PCB contains multiple footprints but no routed segments.", + "severity": "warning", + } + ) + if len(copper_layers) >= 4 and summary["via_count"] == 0: + issues.append( + { + "type": "warning", + "category": "layer_stack", + "rule": "vias", + "message": "A multilayer board was detected without vias.", + "severity": "warning", + } + ) + if len(copper_layers) >= 2 and summary["zone_count"] == 0: + issues.append( + { + "type": "warning", + "category": "thermal", + "rule": "zones", + "message": "No copper zones were detected on a multi-layer board.", + "severity": "warning", + } + ) + + issues = self._filter_issues(issues, check_types) + manufacturing_notes = [ + issue["message"] + for issue in issues + if issue.get("severity") in {"warning", "error"} + ] + + return { + "file_path": str(pcb_path), + "board_info": { + "title": summary["title"], + "size": None + if summary["width_mm"] is None + else f'{summary["width_mm"]}mm x {summary["height_mm"]}mm', + "layers": len(copper_layers), + "copper_layers": copper_layers, + "components": summary["footprint_count"], + "nets": summary["net_count"], + }, + "drc_results": { + "violations": sum(1 for issue in issues if issue.get("severity") == "error"), + "warnings": sum(1 for issue in issues if issue.get("severity") == "warning"), + "issues": issues, + }, + "layer_usage": { + "signal_layers": copper_layers, + "segments": summary["segment_count"], + "vias": summary["via_count"], + "zones": summary["zone_count"], + }, + "manufacturing_notes": manufacturing_notes, + "data_mode": "file-backed-local", + } + + async def _get_component_list(self, args: Dict[str, Any]) -> Dict[str, Any]: + schematic_path = self._check_file(args.get("schematic_path"), ".kicad_sch") + include_values = args.get("include_values", True) + include_footprints = args.get("include_footprints", True) + components = parse_schematic_components(schematic_path) + + return { + "schematic_path": str(schematic_path), + "component_count": len(components), + "components": [ + component.to_dict(include_values=include_values, include_footprints=include_footprints) + for component in components + ], + "data_mode": "file-backed-local", + } + + async def _generate_bom(self, args: Dict[str, Any]) -> Dict[str, Any]: + schematic_path = self._check_file(args.get("schematic_path"), ".kicad_sch") + include_pricing = args.get("include_pricing", False) + group_by = args.get("group_by", "value") + components = parse_schematic_components(schematic_path) + + groups: Dict[str, List[KiCadComponent]] = defaultdict(list) + for component in components: + if group_by == "footprint": + key = component.footprint or "(no-footprint)" + elif group_by == "manufacturer": + key = component.manufacturer or "(no-manufacturer)" + else: + key = component.value or "(no-value)" + groups[key].append(component) + + bom_items: List[Dict[str, Any]] = [] + total_cost = 0.0 + for index, (_, group) in enumerate(sorted(groups.items(), key=lambda item: item[0]), start=1): + first = group[0] + unit_price = first.unit_price if include_pricing else None + total_price = round(unit_price * len(group), 4) if include_pricing and unit_price is not None else None + if total_price is not None: + total_cost += total_price + bom_items.append( + { + "item": index, + "quantity": len(group), + "references": [component.reference for component in group], + "value": first.value, + "footprint": first.footprint, + "manufacturer": first.manufacturer, + "part_number": first.part_number, + "lcsc_id": first.lcsc_id, + "unit_price": unit_price, + "total_price": total_price, + } + ) + + return { + "schematic_path": str(schematic_path), + "generated_date": "2026-03-07", + "group_by": group_by, + "total_items": len(bom_items), + "total_components": len(components), + "total_cost": round(total_cost, 4) if include_pricing and total_cost else None, + "bom_items": bom_items, + "data_mode": "file-backed-local", + } + + def _footprint_patterns(self, component: KiCadComponent) -> List[str]: + patterns = [component.component_type, component.package, component.reference[:1], component.value] + if component.footprint and ":" in component.footprint: + library_name, footprint_name = component.footprint.split(":", 1) + patterns.extend([library_name, footprint_name]) + return [pattern.lower() for pattern in patterns if pattern] + + async def _validate_footprints(self, args: Dict[str, Any]) -> Dict[str, Any]: + schematic_path = self._check_file(args.get("schematic_path"), ".kicad_sch") + suggest_alternatives = args.get("suggest_alternatives", True) + components = parse_schematic_components(schematic_path) + roots = [schematic_path.parent, *self.search_roots] + libraries = discover_footprint_libraries(roots) + + issues: List[Dict[str, Any]] = [] + valid_footprints = 0 + unresolved_assigned = 0 + missing_footprints = 0 + + for component in components: + if not component.footprint: + missing_footprints += 1 + issues.append( + { + "component": component.reference, + "issue": "No footprint assigned", + "assigned_footprint": None, + "alternatives": search_footprints(libraries, self._footprint_patterns(component), limit=5) + if suggest_alternatives and libraries + else [], + } + ) + continue + + if libraries and footprint_exists(component.footprint, libraries): + valid_footprints += 1 + continue + + if libraries: + unresolved_assigned += 1 + issues.append( + { + "component": component.reference, + "issue": "Assigned footprint not found in local footprint libraries", + "assigned_footprint": component.footprint, + "alternatives": search_footprints(libraries, self._footprint_patterns(component), limit=5) + if suggest_alternatives + else [], + } + ) + else: + issues.append( + { + "component": component.reference, + "issue": "Footprint could not be verified because no local .pretty libraries were found", + "assigned_footprint": component.footprint, + "alternatives": [], + } + ) + + return { + "schematic_path": str(schematic_path), + "total_components": len(components), + "valid_footprints": valid_footprints, + "missing_footprints": missing_footprints, + "unresolved_assigned_footprints": unresolved_assigned, + "local_library_count": len(libraries), + "issues": issues, + "data_mode": "file-backed-local", + } + + async def _suggest_improvements(self, args: Dict[str, Any]) -> Dict[str, Any]: + project_info = self._resolve_project_files(args.get("project_path", "")) + schematic_path = project_info["schematic"] + pcb_path = project_info["pcb"] + focus_areas = args.get("focus_areas", ["cost", "performance", "reliability", "manufacturability"]) + + suggestions = { + "cost_optimization": [], + "performance_improvements": [], + "reliability_enhancements": [], + "manufacturability": [], + } + + if schematic_path: + components = parse_schematic_components(schematic_path) + component_types = Counter(component.component_type for component in components) + unique_footprints = {component.footprint for component in components if component.footprint} + components_without_part_numbers = [component.reference for component in components if not component.part_number] + components_without_manufacturers = [component.reference for component in components if not component.manufacturer] + footprint_validation = await self._validate_footprints({"schematic_path": str(schematic_path)}) + + if "cost" in focus_areas and len(unique_footprints) > max(3, len(components) // 2): + suggestions["cost_optimization"].append( + { + "suggestion": "Standardize footprints and passive package sizes to reduce setup variance.", + "impact": f"{len(unique_footprints)} unique footprints detected across {len(components)} components.", + } + ) + if "cost" in focus_areas and components_without_part_numbers: + suggestions["cost_optimization"].append( + { + "suggestion": "Add manufacturer part numbers for sourcing automation and cost comparison.", + "impact": f"{len(components_without_part_numbers)} components are missing part numbers.", + } + ) + if "performance" in focus_areas and component_types.get("ic", 0) + component_types.get("microcontroller", 0) > component_types.get("capacitor", 0): + suggestions["performance_improvements"].append( + { + "suggestion": "Add or verify local decoupling capacitors for ICs and MCUs.", + "impact": "The schematic contains fewer capacitors than active IC-class components.", + } + ) + if "reliability" in focus_areas and components_without_manufacturers: + suggestions["reliability_enhancements"].append( + { + "suggestion": "Add manufacturer metadata to reduce substitution ambiguity.", + "impact": f"{len(components_without_manufacturers)} components are missing manufacturer information.", + } + ) + if "manufacturability" in focus_areas and footprint_validation["missing_footprints"] > 0: + suggestions["manufacturability"].append( + { + "suggestion": "Assign footprints to all symbols before board handoff.", + "impact": f'{footprint_validation["missing_footprints"]} components are missing footprints.', + } + ) + if "manufacturability" in focus_areas and footprint_validation["unresolved_assigned_footprints"] > 0: + suggestions["manufacturability"].append( + { + "suggestion": "Bring the referenced footprint libraries into the project or workspace.", + "impact": f'{footprint_validation["unresolved_assigned_footprints"]} assigned footprints could not be resolved locally.', + } + ) + + if pcb_path: + pcb_summary = parse_pcb_summary(pcb_path) + if "performance" in focus_areas and len(pcb_summary["copper_layers"]) >= 2 and pcb_summary["zone_count"] == 0: + suggestions["performance_improvements"].append( + { + "suggestion": "Add at least one copper zone for power or ground distribution.", + "impact": "Multi-layer board detected without copper zones.", + } + ) + if "manufacturability" in focus_areas and pcb_summary["edge_cuts_count"] == 0: + suggestions["manufacturability"].append( + { + "suggestion": "Define the board outline on Edge.Cuts before fabrication export.", + "impact": "No board outline geometry was found in the PCB file.", + } + ) + + return { + "project_path": str(project_info["directory"]), + "focus_areas": focus_areas, + "suggestions": suggestions, + "data_mode": "file-backed-local", + } + async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]: - """Handle MCP requests""" method = request.get("method") params = request.get("params", {}) request_id = request.get("id") - + if method == "initialize": return { "jsonrpc": "2.0", "id": request_id, "result": { - "protocolVersion": "2024-11-05", - "capabilities": { - "tools": {} - }, + "protocolVersion": "2025-03-26", + "capabilities": {"tools": {}}, "serverInfo": { "name": "kicad-tools", - "version": "1.0.0" - } - } + "version": "2.0.0", + }, + }, } - - elif method == "tools/list": + + if method == "tools/list": return { "jsonrpc": "2.0", "id": request_id, @@ -46,454 +540,135 @@ class KiCadToolsServer: "tools": [ { "name": "analyze_schematic", - "description": "Analyze schematic for design rule violations and suggestions", + "description": "Analyze a .kicad_sch file from its real file contents", "inputSchema": { "type": "object", "properties": { - "schematic_path": {"type": "string", "description": "Path to .kicad_sch file"}, - "check_types": { - "type": "array", - "items": {"type": "string"}, - "description": "Types of checks: erc, power, connectivity, components" - } + "schematic_path": {"type": "string"}, + "check_types": {"type": "array", "items": {"type": "string"}}, }, - "required": ["schematic_path"] - } + "required": ["schematic_path"], + }, }, { "name": "analyze_pcb", - "description": "Analyze PCB layout for design issues", + "description": "Analyze a .kicad_pcb file from its real file contents", "inputSchema": { "type": "object", "properties": { - "pcb_path": {"type": "string", "description": "Path to .kicad_pcb file"}, - "check_types": { - "type": "array", - "items": {"type": "string"}, - "description": "Types of checks: drc, layer_stack, thermal, impedance" - } + "pcb_path": {"type": "string"}, + "check_types": {"type": "array", "items": {"type": "string"}}, }, - "required": ["pcb_path"] - } + "required": ["pcb_path"], + }, }, { "name": "get_component_list", - "description": "Extract component list from schematic", + "description": "Extract the placed components from a schematic", "inputSchema": { "type": "object", "properties": { "schematic_path": {"type": "string"}, - "include_values": {"type": "boolean", "default": true}, - "include_footprints": {"type": "boolean", "default": true} + "include_values": {"type": "boolean", "default": True}, + "include_footprints": {"type": "boolean", "default": True}, }, - "required": ["schematic_path"] - } + "required": ["schematic_path"], + }, }, { "name": "generate_bom", - "description": "Generate Bill of Materials with pricing if available", + "description": "Generate a BOM from the schematic file", "inputSchema": { "type": "object", "properties": { "schematic_path": {"type": "string"}, - "include_pricing": {"type": "boolean", "default": false}, - "group_by": {"type": "string", "enum": ["value", "footprint", "manufacturer"], "default": "value"} + "include_pricing": {"type": "boolean", "default": False}, + "group_by": { + "type": "string", + "enum": ["value", "footprint", "manufacturer"], + "default": "value", + }, }, - "required": ["schematic_path"] - } + "required": ["schematic_path"], + }, }, { "name": "suggest_improvements", - "description": "Suggest design improvements based on analysis", + "description": "Generate file-backed design suggestions for a KiCad project directory or file", "inputSchema": { "type": "object", "properties": { "project_path": {"type": "string"}, - "focus_areas": { - "type": "array", - "items": {"type": "string"}, - "description": "Areas to focus on: cost, performance, reliability, manufacturability" - } + "focus_areas": {"type": "array", "items": {"type": "string"}}, }, - "required": ["project_path"] - } + "required": ["project_path"], + }, }, { "name": "validate_footprints", - "description": "Check if all components have valid footprints assigned", + "description": "Validate that assigned footprints can be resolved from local .pretty libraries", "inputSchema": { "type": "object", "properties": { "schematic_path": {"type": "string"}, - "suggest_alternatives": {"type": "boolean", "default": true} + "suggest_alternatives": {"type": "boolean", "default": True}, }, - "required": ["schematic_path"] - } - } + "required": ["schematic_path"], + }, + }, ] - } + }, } - - elif method == "tools/call": + + if method == "tools/call": tool_name = params.get("name") arguments = params.get("arguments", {}) - - if tool_name == "analyze_schematic": - result = await self._analyze_schematic(arguments) + handlers = { + "analyze_schematic": self._analyze_schematic, + "analyze_pcb": self._analyze_pcb, + "get_component_list": self._get_component_list, + "generate_bom": self._generate_bom, + "suggest_improvements": self._suggest_improvements, + "validate_footprints": self._validate_footprints, + } + handler = handlers.get(tool_name) + if handler is None: return { "jsonrpc": "2.0", "id": request_id, - "result": result + "error": {"code": -32601, "message": f"Unknown tool: {tool_name}"}, } - - elif tool_name == "analyze_pcb": - result = await self._analyze_pcb(arguments) - return { - "jsonrpc": "2.0", - "id": request_id, - "result": result - } - - elif tool_name == "get_component_list": - result = await self._get_component_list(arguments) - return { - "jsonrpc": "2.0", - "id": request_id, - "result": result - } - - elif tool_name == "generate_bom": - result = await self._generate_bom(arguments) - return { - "jsonrpc": "2.0", - "id": request_id, - "result": result - } - - elif tool_name == "suggest_improvements": - result = await self._suggest_improvements(arguments) - return { - "jsonrpc": "2.0", - "id": request_id, - "result": result - } - - elif tool_name == "validate_footprints": - result = await self._validate_footprints(arguments) - return { - "jsonrpc": "2.0", - "id": request_id, - "result": result - } - + result = await handler(arguments) + return {"jsonrpc": "2.0", "id": request_id, "result": result} + return { "jsonrpc": "2.0", "id": request_id, - "error": {"code": -32601, "message": "Method not found"} + "error": {"code": -32601, "message": "Method not found"}, } - - async def _analyze_schematic(self, args: Dict[str, Any]) -> Dict[str, Any]: - """Analyze schematic file""" - try: - schematic_path = args.get("schematic_path") - check_types = args.get("check_types", ["erc", "connectivity"]) - - # Simulated schematic analysis - results = { - "file_path": schematic_path, - "analysis_summary": { - "total_components": 45, - "total_nets": 67, - "warnings": 2, - "errors": 0 - }, - "issues": [ - { - "type": "warning", - "category": "erc", - "message": "Power input not connected on U1 pin 8", - "component": "U1", - "pin": "8", - "suggestion": "Connect VCC to power rail or add decoupling capacitor" - }, - { - "type": "warning", - "category": "connectivity", - "message": "Net 'LED_CTL' only has one connection", - "net": "LED_CTL", - "suggestion": "Verify if this signal should connect to additional components" - } - ], - "power_analysis": { - "total_current_draw": "125mA", - "power_rails": [ - {"rail": "VCC", "voltage": "3.3V", "current": "85mA"}, - {"rail": "VDD", "voltage": "5V", "current": "40mA"} - ] - }, - "component_summary": { - "resistors": 15, - "capacitors": 12, - "ics": 3, - "connectors": 2, - "other": 13 - } - } - - return results - - except Exception as e: - return {"error": f"Schematic analysis failed: {str(e)}"} - - async def _analyze_pcb(self, args: Dict[str, Any]) -> Dict[str, Any]: - """Analyze PCB layout""" - try: - pcb_path = args.get("pcb_path") - check_types = args.get("check_types", ["drc"]) - - # Simulated PCB analysis - results = { - "file_path": pcb_path, - "board_info": { - "size": "50mm x 80mm", - "layers": 4, - "thickness": "1.6mm", - "components": 45 - }, - "drc_results": { - "violations": 1, - "warnings": 3, - "issues": [ - { - "type": "violation", - "rule": "minimum_trace_width", - "location": "(25.4, 12.7)", - "message": "Trace width 0.1mm below minimum 0.15mm", - "severity": "error" - }, - { - "type": "warning", - "rule": "via_size", - "location": "(45.2, 33.1)", - "message": "Via size may be too small for reliable manufacturing", - "severity": "warning" - } - ] - }, - "layer_usage": { - "signal_layers": ["F.Cu", "In1.Cu", "In2.Cu", "B.Cu"], - "plane_layers": ["In1.Cu (GND)", "In2.Cu (VCC)"], - "utilization": "75%" - }, - "manufacturing_notes": [ - "Consider increasing trace width for power rails", - "Add more thermal vias under high-power components", - "Verify minimum drill sizes with manufacturer" - ] - } - - return results - - except Exception as e: - return {"error": f"PCB analysis failed: {str(e)}"} - - async def _get_component_list(self, args: Dict[str, Any]) -> Dict[str, Any]: - """Extract component list from schematic""" - try: - schematic_path = args.get("schematic_path") - include_values = args.get("include_values", True) - include_footprints = args.get("include_footprints", True) - - # Simulated component list - components = [ - { - "reference": "R1", - "value": "10kΩ" if include_values else None, - "footprint": "Resistor_SMD:R_0805_2012Metric" if include_footprints else None, - "description": "Resistor", - "manufacturer": "Yageo", - "part_number": "RC0805FR-0710KL" - }, - { - "reference": "C1", - "value": "100nF" if include_values else None, - "footprint": "Capacitor_SMD:C_0805_2012Metric" if include_footprints else None, - "description": "Capacitor", - "manufacturer": "Samsung", - "part_number": "CL21B104KBCNNNC" - }, - { - "reference": "U1", - "value": "ATmega328P-AU" if include_values else None, - "footprint": "Package_QFP:TQFP-32_7x7mm_P0.8mm" if include_footprints else None, - "description": "Microcontroller", - "manufacturer": "Microchip", - "part_number": "ATMEGA328P-AU" - } - ] - - return { - "schematic_path": schematic_path, - "component_count": len(components), - "components": components - } - - except Exception as e: - return {"error": f"Failed to extract component list: {str(e)}"} - - async def _generate_bom(self, args: Dict[str, Any]) -> Dict[str, Any]: - """Generate Bill of Materials""" - try: - schematic_path = args.get("schematic_path") - include_pricing = args.get("include_pricing", False) - group_by = args.get("group_by", "value") - - # Simulated BOM - bom_items = [ - { - "item": 1, - "quantity": 15, - "references": ["R1", "R2", "R3", "R4", "R5", "R6", "R7", "R8", "R9", "R10", "R11", "R12", "R13", "R14", "R15"], - "value": "10kΩ", - "footprint": "Resistor_SMD:R_0805_2012Metric", - "manufacturer": "Yageo", - "part_number": "RC0805FR-0710KL", - "unit_price": 0.10 if include_pricing else None, - "total_price": 1.50 if include_pricing else None - }, - { - "item": 2, - "quantity": 12, - "references": ["C1", "C2", "C3", "C4", "C5", "C6", "C7", "C8", "C9", "C10", "C11", "C12"], - "value": "100nF", - "footprint": "Capacitor_SMD:C_0805_2012Metric", - "manufacturer": "Samsung", - "part_number": "CL21B104KBCNNNC", - "unit_price": 0.05 if include_pricing else None, - "total_price": 0.60 if include_pricing else None - } - ] - - total_cost = sum(item.get("total_price", 0) for item in bom_items) if include_pricing else None - - return { - "schematic_path": schematic_path, - "generated_date": "2025-08-01", - "total_items": len(bom_items), - "total_components": sum(item["quantity"] for item in bom_items), - "total_cost": total_cost, - "bom_items": bom_items - } - - except Exception as e: - return {"error": f"BOM generation failed: {str(e)}"} - - async def _suggest_improvements(self, args: Dict[str, Any]) -> Dict[str, Any]: - """Suggest design improvements""" - try: - project_path = args.get("project_path") - focus_areas = args.get("focus_areas", ["cost", "performance"]) - - suggestions = { - "cost_optimization": [ - { - "component": "R1-R15", - "suggestion": "Consider using resistor arrays instead of individual resistors", - "potential_savings": "15%", - "impact": "Reduced component count and assembly cost" - } - ], - "performance_improvements": [ - { - "area": "Power supply decoupling", - "suggestion": "Add more decoupling capacitors near high-speed ICs", - "impact": "Improved signal integrity and reduced noise" - } - ], - "reliability_enhancements": [ - { - "component": "U1", - "suggestion": "Add ESD protection on exposed pins", - "impact": "Increased robustness against electrostatic discharge" - } - ], - "manufacturability": [ - { - "suggestion": "Standardize on 0805 package size where possible", - "impact": "Simplified pick-and-place setup and reduced setup costs" - } - ] - } - - return { - "project_path": project_path, - "focus_areas": focus_areas, - "suggestions": suggestions - } - - except Exception as e: - return {"error": f"Failed to generate suggestions: {str(e)}"} - - async def _validate_footprints(self, args: Dict[str, Any]) -> Dict[str, Any]: - """Validate component footprints""" - try: - schematic_path = args.get("schematic_path") - suggest_alternatives = args.get("suggest_alternatives", True) - - validation_results = { - "total_components": 45, - "valid_footprints": 43, - "missing_footprints": 2, - "issues": [ - { - "component": "J1", - "issue": "No footprint assigned", - "alternatives": [ - "Connector_PinHeader_2.54mm:PinHeader_1x04_P2.54mm_Vertical", - "Connector_JST:JST_XH_B4B-XH-A_1x04_P2.50mm_Vertical" - ] if suggest_alternatives else None - }, - { - "component": "D1", - "issue": "Footprint library not found", - "alternatives": [ - "LED_SMD:LED_0805_2012Metric", - "LED_THT:LED_D5.0mm" - ] if suggest_alternatives else None - } - ] - } - - return validation_results - - except Exception as e: - return {"error": f"Footprint validation failed: {str(e)}"} async def main(): - """Main MCP server loop""" server = KiCadToolsServer() - while True: try: line = sys.stdin.readline() if not line: break - + request = json.loads(line.strip()) response = await server.handle_request(request) - print(json.dumps(response)) sys.stdout.flush() - - except Exception as e: + except Exception as exc: error_response = { "jsonrpc": "2.0", "id": None, - "error": {"code": -32603, "message": f"Internal error: {str(e)}"} + "error": {"code": -32603, "message": f"Internal error: {exc}"}, } print(json.dumps(error_response)) sys.stdout.flush() + if __name__ == "__main__": asyncio.run(main()) diff --git a/mcp_servers/nexar.py b/mcp_servers/nexar.py index bb31724..8f15f6e 100644 --- a/mcp_servers/nexar.py +++ b/mcp_servers/nexar.py @@ -9,6 +9,7 @@ import json import sys import asyncio import os +import logging from typing import Dict, List, Any, Optional try: @@ -17,17 +18,55 @@ try: except ImportError: REQUESTS_AVAILABLE = False + +LOG_LEVEL = os.getenv("NEXAR_MCP_LOG_LEVEL", "WARNING").upper() +logging.basicConfig(level=getattr(logging, LOG_LEVEL, logging.WARNING)) +logger = logging.getLogger("kicad_kic_ai.nexar_mcp") + class NexarServer: """MCP server for Nexar API integration (formerly Octopart)""" def __init__(self): - self.api_token = os.getenv('NEXAR_TOKEN') + self.api_token = os.getenv('NEXAR_TOKEN') or os.getenv('NEXAR_API_KEY') self.base_url = "https://api.nexar.com/graphql" # Demo mode with realistic data self.demo_mode = not self.api_token if self.demo_mode: - print("Info: No NEXAR_TOKEN found, using demo pricing mode", file=sys.stderr) + logger.info("No Nexar token configured, using demo mode") + + def _result_payload(self, summary_text: str, **meta: Any) -> Dict[str, Any]: + return { + "content": [{"type": "text", "text": summary_text}], + "isError": False, + "_meta": meta, + **meta, + } + + def _part_with_client_aliases(self, part: Dict[str, Any]) -> Dict[str, Any]: + normalized = dict(part) + mpn = normalized.get("mpn") or normalized.get("part_number") + if mpn: + normalized["mpn"] = mpn + normalized["part_number"] = mpn + + distributors = normalized.get("distributors", {}) + pricing = {} + for distributor, details in distributors.items(): + if "price_1" in details: + pricing[distributor] = details + continue + unit_price = details.get("unit_price") + pricing[distributor] = { + "price_1": unit_price, + "stock": details.get("stock"), + "minimum_order": details.get("minimum_order"), + "url": details.get("url"), + } + if pricing: + normalized["pricing"] = pricing + + return normalized async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]: """Handle MCP requests""" @@ -40,7 +79,7 @@ class NexarServer: "jsonrpc": "2.0", "id": request_id, "result": { - "protocolVersion": "2024-11-05", + "protocolVersion": "2025-03-26", "capabilities": { "tools": {} }, @@ -174,10 +213,7 @@ class NexarServer: try: return await self._search_parts_api(args) except Exception as e: - print(f"Nexar API call failed, using demo mode: {e}", file=sys.stderr) - - # Demo mode - comprehensive pricing examples - print("Warning: No NEXAR_TOKEN configured, using demo data", file=sys.stderr) + logger.warning("Nexar API call failed, falling back to demo mode: %s", e) # Create demo responses based on search query demo_parts = [] @@ -193,7 +229,7 @@ class NexarServer: value = "2.2K" demo_parts = [ - { + self._part_with_client_aliases({ "mpn": f"RC0805FR-07{value}L", "manufacturer": "Yageo", "description": f"RES SMD {value} OHM 1% 1/8W 0805", @@ -241,12 +277,12 @@ class NexarServer: "package": "0805", "temperature_coefficient": "±100ppm/°C" } - } + }) ] elif 'capacitor' in query.lower() or 'cap' in query.lower(): demo_parts = [ - { + self._part_with_client_aliases({ "mpn": "CL21B104KBCNNNC", "manufacturer": "Samsung Electro-Mechanics", "description": "CAP CER 100NF 50V X7R 0805", @@ -284,13 +320,13 @@ class NexarServer: "dielectric": "X7R", "package": "0805" } - } + }) ] else: # Generic search result demo_parts = [ - { + self._part_with_client_aliases({ "mpn": "DEMO-PART-001", "manufacturer": "Demo Manufacturer", "description": f"Demo component for: {query}", @@ -305,16 +341,25 @@ class NexarServer: "stock": 8000, "minimum_order": 1 } - } + }) ] - return { - "parts": demo_parts, - "total_count": len(demo_parts), - "demo_mode": True, - "message": "Demo data - add NEXAR_TOKEN for live pricing from distributors", - "distributors_covered": ["Digi-Key", "Mouser", "Farnell", "Newark", "Arrow", "RS Components", "Avnet"] - } + return self._result_payload( + f"Found {len(demo_parts)} parts in demo mode", + parts=demo_parts, + total_count=len(demo_parts), + demo_mode=True, + message="Demo data - add NEXAR_TOKEN for live distributor pricing", + distributors_covered=[ + "Digi-Key", + "Mouser", + "Farnell", + "Newark", + "Arrow", + "RS Components", + "Avnet", + ], + ) except Exception as e: return {"error": f"Search failed: {str(e)}"} @@ -385,15 +430,18 @@ class NexarServer: "specifications": {spec.get('attribute', {}).get('name', ''): spec.get('value', '') for spec in part.get('specs', [])} } - parts.append(part_info) + parts.append(self._part_with_client_aliases(part_info)) - return { - "parts": parts, - "total_count": len(parts), - "demo_mode": False, - "message": f"Found {len(parts)} parts via Nexar API", - "distributors_covered": list(set([d for p in parts for d in p.get('distributors', {}).keys()])) - } + return self._result_payload( + f"Found {len(parts)} parts via Nexar API", + parts=parts, + total_count=len(parts), + demo_mode=False, + message=f"Found {len(parts)} parts via Nexar API", + distributors_covered=list( + set([d for p in parts for d in p.get('distributors', {}).keys()]) + ), + ) else: raise Exception(f"API error: {response.status_code} - {response.text}") @@ -458,7 +506,11 @@ class NexarServer: "demo_mode": True } - return demo_pricing + return self._result_payload( + f"Pricing for {mpn} ({'demo' if self.demo_mode else 'api'} mode)", + pricing=demo_pricing, + demo_mode=self.demo_mode, + ) except Exception as e: return {"error": f"Failed to get pricing: {str(e)}"} @@ -503,16 +555,18 @@ class NexarServer: "lead_time": "In Stock" }) - return { - "parts": best_prices, - "total_bom_cost": round(total_cost, 2), - "average_unit_price": round(total_cost / sum(p.get('quantity', 1) for p in parts), 4), - "recommended_distributor": "Mouser", - "estimated_shipping": 15.00, - "grand_total": round(total_cost + 15.00, 2), - "demo_mode": True, - "message": "Demo BOM pricing - real API provides accurate multi-distributor comparison" - } + total_quantity = sum(p.get('quantity', 1) for p in parts) or 1 + return self._result_payload( + f"Calculated best pricing for {len(parts)} parts", + parts=best_prices, + total_bom_cost=round(total_cost, 2), + average_unit_price=round(total_cost / total_quantity, 4), + recommended_distributor="Mouser", + estimated_shipping=15.00, + grand_total=round(total_cost + 15.00, 2), + demo_mode=True, + message="Demo BOM pricing - real API provides accurate multi-distributor comparison", + ) except Exception as e: return {"error": f"Failed to calculate best prices: {str(e)}"} @@ -545,12 +599,13 @@ class NexarServer: } ] - return { - "original_part": mpn, - "alternatives": alternatives, - "total_alternatives": len(alternatives), - "demo_mode": True - } + return self._result_payload( + f"Found {len(alternatives)} alternatives for {mpn}", + original_part=mpn, + alternatives=alternatives, + total_alternatives=len(alternatives), + demo_mode=True, + ) except Exception as e: return {"error": f"Failed to find alternatives: {str(e)}"} diff --git a/plugins/config_dialog.py b/plugins/config_dialog.py index f6a59b8..2ae19d9 100644 --- a/plugins/config_dialog.py +++ b/plugins/config_dialog.py @@ -204,7 +204,7 @@ class MCPServerConfigPanel(wx.Panel): "id": 1, "method": "initialize", "params": { - "protocolVersion": "2024-11-05", + "protocolVersion": "2025-03-26", "capabilities": {"tools": {}}, "clientInfo": {"name": "config-test", "version": "1.0.0"} } diff --git a/plugins/mcp_client.py b/plugins/mcp_client.py index 5f0572d..c75250f 100644 --- a/plugins/mcp_client.py +++ b/plugins/mcp_client.py @@ -41,7 +41,7 @@ class MCPClient: "id": 1, "method": "initialize", "params": { - "protocolVersion": "2024-11-05", + "protocolVersion": "2025-03-26", "capabilities": { "tools": {} }, diff --git a/plugins/mcp_config.json b/plugins/mcp_config.json index 99dcddf..fb358ca 100644 --- a/plugins/mcp_config.json +++ b/plugins/mcp_config.json @@ -2,8 +2,8 @@ "mcp_servers": { "component_database": { "command": ["python3", "-m", "mcp_servers.component_db"], - "description": "Component database with pricing and availability", - "enabled": false + "description": "Component database indexed from local KiCad schematics and inline symbol libraries", + "enabled": true }, "nexar_api": { "command": ["python3", "-m", "mcp_servers.nexar"], @@ -13,8 +13,8 @@ }, "kicad_tools": { "command": ["python3", "-m", "mcp_servers.kicad_tools"], - "description": "KiCad design manipulation tools", - "enabled": false + "description": "KiCad design analysis and BOM tools backed by local KiCad files", + "enabled": true } }, "ai_enhancement": { diff --git a/plugins/nexar_server.py b/plugins/nexar_server.py index a18bbb5..dec1198 100644 --- a/plugins/nexar_server.py +++ b/plugins/nexar_server.py @@ -58,7 +58,7 @@ class NexarServer: def handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]: """Handle MCP initialize request""" return { - "protocolVersion": "2024-11-05", + "protocolVersion": "2025-03-26", "capabilities": self.capabilities, "serverInfo": { "name": "nexar-server", diff --git a/plugins/simple_mcp_client_embedded.py b/plugins/simple_mcp_client_embedded.py index 9fd70ce..f47d210 100644 --- a/plugins/simple_mcp_client_embedded.py +++ b/plugins/simple_mcp_client_embedded.py @@ -38,7 +38,7 @@ class EmbeddedNexarServer: def handle_initialize(self, params): mode_info = "Demo mode" if self.demo_mode else f"API mode (key: {self.api_key[:8]}...)" return { - "protocolVersion": "2024-11-05", + "protocolVersion": "2025-03-26", "capabilities": self.capabilities, "serverInfo": { "name": "embedded-nexar-server", @@ -156,11 +156,10 @@ class SimpleMCPClient: try: # Try to find external server files first possible_paths = [ - os.path.join(os.path.dirname(__file__), 'nexar_server.py'), os.path.join(os.path.dirname(os.path.dirname(__file__)), 'mcp_servers', 'nexar.py'), os.path.join(os.path.dirname(__file__), '..', 'mcp_servers', 'nexar.py'), os.path.join('mcp_servers', 'nexar.py'), - os.path.join(os.path.dirname(os.path.dirname(__file__)), 'mcp_servers', 'nexar.py') + os.path.join(os.path.dirname(__file__), 'nexar_server.py'), ] server_path = None diff --git a/test_aux_mcp_smoke.py b/test_aux_mcp_smoke.py new file mode 100644 index 0000000..d025b1e --- /dev/null +++ b/test_aux_mcp_smoke.py @@ -0,0 +1,259 @@ +#!/usr/bin/env python3 +""" +Smoke test for the auxiliary KIC-AI MCP servers. + +This validates the line-delimited JSON-RPC transport and a minimal set of +real tool calls against deterministic local fixtures. +""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import tempfile +from pathlib import Path +from typing import Any, Dict, List, Optional + + +ROOT = Path(__file__).resolve().parent +CACHE_ROOT = Path.home() / "Kill_LIFE" / ".cad-home" / "kicad-mcp" / "kicad-v10-libs" + + +SCHEMATIC_FIXTURE = """\ +(kicad_sch (version 20250114) (generator "aux-mcp-smoke") + (uuid 11111111-1111-1111-1111-111111111111) + (paper "A4") + (symbol (lib_id "Device:R") (at 10 10 0) + (property "Reference" "R1" (at 10 12 0)) + (property "Value" "10k" (at 10 8 0)) + (property "Footprint" "Resistor_SMD:R_0805_2012Metric" (at 10 6 0)) + (property "Datasheet" "~" (at 10 4 0)) + ) + (symbol (lib_id "Device:C") (at 20 10 0) + (property "Reference" "C1" (at 20 12 0)) + (property "Value" "100nF" (at 20 8 0)) + (property "Footprint" "Capacitor_SMD:C_0805_2012Metric" (at 20 6 0)) + (property "Datasheet" "~" (at 20 4 0)) + ) + (symbol (lib_id "MCU_Microchip_ATmega:ATmega328P-AU") (at 30 10 0) + (property "Reference" "U1" (at 30 12 0)) + (property "Value" "ATmega328P-AU" (at 30 8 0)) + (property "Footprint" "Package_QFP:TQFP-32_7x7mm_P0.8mm" (at 30 6 0)) + (property "Manufacturer" "Microchip" (at 30 4 0)) + (property "Part" "ATMEGA328P-AU" (at 30 2 0)) + ) + (symbol (lib_id "Connector_Generic:Conn_01x02") (at 40 10 0) + (property "Reference" "J1" (at 40 12 0)) + (property "Value" "Conn_01x02" (at 40 8 0)) + (property "Footprint" "" (at 40 6 0)) + ) + (global_label "VCC" (shape input) (at 5 5 0) (fields_autoplaced)) + (label "SCL" (at 15 5 0) (fields_autoplaced)) + (wire (pts (xy 5 5) (xy 15 5)) (stroke (width 0) (type default)) (uuid abcdefab-cdef-abcd-efab-cdefabcdefab)) + (wire (pts (xy 15 5) (xy 25 5)) (stroke (width 0) (type default)) (uuid bcdefabc-defa-bcde-fabc-defabcdefabc)) +) +""" + + +PCB_FIXTURE = """\ +(kicad_pcb + (version 20260206) + (generator "pcbnew") + (title_block (title "aux-mcp-smoke")) + (layers + (0 "F.Cu" signal) + (2 "B.Cu" signal) + (25 "Edge.Cuts" user) + ) + (gr_rect (start 0 0) (end 50 40) (stroke (width 0.1) (type default)) (fill none) (layer "Edge.Cuts")) + (footprint "Resistor_SMD:R_0805_2012Metric" (layer "F.Cu") (tstamp 1)) + (footprint "Package_QFP:TQFP-32_7x7mm_P0.8mm" (layer "F.Cu") (tstamp 2)) + (segment (start 5 5) (end 10 5) (width 0.25) (layer "F.Cu") (net 1) (tstamp 3)) + (zone (net 0) (net_name "") (layer "F.Cu") (tstamp 4) (name "") (hatch edge 0.5) + (connect_pads yes (clearance 0.5)) + (min_thickness 0.25) + (filled_areas_thickness no) + (fill yes) + (polygon (pts (xy 0 0) (xy 10 0) (xy 10 10) (xy 0 10))) + ) +) +""" + + +class MCPLineClient: + def __init__(self, command: List[str], *, cwd: Optional[Path] = None, env: Optional[Dict[str, str]] = None): + process_env = os.environ.copy() + if env: + process_env.update(env) + self.process = subprocess.Popen( + command, + cwd=cwd or ROOT, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + env=process_env, + ) + self._request_id = 1 + + def close(self) -> None: + if self.process.poll() is None: + self.process.terminate() + try: + self.process.wait(timeout=5) + except subprocess.TimeoutExpired: + self.process.kill() + + def request(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]: + request = {"jsonrpc": "2.0", "id": self._request_id, "method": method, "params": params} + self._request_id += 1 + assert self.process.stdin is not None + assert self.process.stdout is not None + self.process.stdin.write(json.dumps(request) + "\n") + self.process.stdin.flush() + response = self.process.stdout.readline() + if not response: + stderr = "" + if self.process.stderr is not None: + stderr = self.process.stderr.read() + raise RuntimeError(f"No response for {method}; stderr={stderr.strip()}") + payload = json.loads(response) + if "error" in payload: + raise RuntimeError(f"{method} failed: {payload['error']}") + return payload["result"] + + +def assert_true(condition: bool, message: str) -> None: + if not condition: + raise AssertionError(message) + + +def create_fixture_project() -> Path: + tmpdir = Path(tempfile.mkdtemp(prefix="aux-mcp-smoke-")) + (tmpdir / "aux_test.kicad_sch").write_text(SCHEMATIC_FIXTURE, encoding="utf-8") + (tmpdir / "aux_test.kicad_pcb").write_text(PCB_FIXTURE, encoding="utf-8") + return tmpdir + + +def run_component_db_checks(project_dir: Path) -> Dict[str, Any]: + client = MCPLineClient( + ["python3", "-m", "mcp_servers.component_db"], + env={"KICAD_AUX_MCP_ROOTS": str(project_dir)}, + ) + try: + init = client.request("initialize", {}) + tools = client.request("tools/list", {}) + search = client.request( + "tools/call", + {"name": "search_components", "arguments": {"type": "microcontroller", "limit": 5}}, + ) + pricing = client.request( + "tools/call", + {"name": "get_pricing", "arguments": {"part_numbers": ["Device:R", "Device:C"]}}, + ) + assert_true(init["serverInfo"]["name"] == "component-database", "component_db initialize failed") + assert_true(len(tools["tools"]) == 3, "component_db tools/list mismatch") + assert_true(search["total_found"] >= 1, "component_db search returned no results") + assert_true("pricing" in pricing, "component_db pricing payload missing") + assert_true((CACHE_ROOT / ".symbol_index.sqlite3").exists(), "symbol sqlite index was not created") + return {"catalog_size": search["catalog_size"], "search_results": search["total_found"]} + finally: + client.close() + + +def run_nexar_checks() -> Dict[str, Any]: + client = MCPLineClient(["python3", "-m", "mcp_servers.nexar"]) + try: + init = client.request("initialize", {}) + tools = client.request("tools/list", {}) + search = client.request( + "tools/call", + {"name": "search_parts", "arguments": {"query": "STM32", "limit": 2}}, + ) + assert_true(init["serverInfo"]["name"] == "nexar-api", "nexar initialize failed") + assert_true(len(tools["tools"]) >= 4, "nexar tools/list mismatch") + assert_true(search["_meta"]["parts"], "nexar search did not expose _meta.parts") + return {"parts_found": len(search["_meta"]["parts"]), "demo_mode": search["_meta"]["demo_mode"]} + finally: + client.close() + + +def run_kicad_tools_checks(project_dir: Path) -> Dict[str, Any]: + schematic = project_dir / "aux_test.kicad_sch" + pcb = project_dir / "aux_test.kicad_pcb" + client = MCPLineClient(["python3", "-m", "mcp_servers.kicad_tools"]) + try: + init = client.request("initialize", {}) + tools = client.request("tools/list", {}) + component_list = client.request( + "tools/call", + {"name": "get_component_list", "arguments": {"schematic_path": str(schematic)}}, + ) + analysis = client.request( + "tools/call", + {"name": "analyze_schematic", "arguments": {"schematic_path": str(schematic)}}, + ) + bom = client.request( + "tools/call", + {"name": "generate_bom", "arguments": {"schematic_path": str(schematic), "group_by": "value"}}, + ) + footprints = client.request( + "tools/call", + {"name": "validate_footprints", "arguments": {"schematic_path": str(schematic)}}, + ) + pcb_analysis = client.request( + "tools/call", + {"name": "analyze_pcb", "arguments": {"pcb_path": str(pcb)}}, + ) + suggestions = client.request( + "tools/call", + {"name": "suggest_improvements", "arguments": {"project_path": str(schematic)}}, + ) + + assert_true(init["serverInfo"]["name"] == "kicad-tools", "kicad_tools initialize failed") + assert_true(len(tools["tools"]) == 6, "kicad_tools tools/list mismatch") + assert_true(component_list["component_count"] == 4, "unexpected schematic component count") + assert_true(analysis["analysis_summary"]["total_components"] == 4, "schematic analysis mismatch") + assert_true(bom["total_components"] == 4, "BOM generation mismatch") + assert_true("issues" in footprints, "footprint validation payload missing") + assert_true(footprints["local_library_count"] > 0, "container KiCad libraries were not detected") + assert_true(footprints["valid_footprints"] >= 3, "expected fixture footprints to resolve against KiCad v10 libraries") + assert_true((CACHE_ROOT / ".footprint_index.json").exists(), "footprint index file was not created") + assert_true(pcb_analysis["board_info"]["components"] == 2, "pcb analysis mismatch") + assert_true("suggestions" in suggestions, "suggest improvements payload missing") + + return { + "components": component_list["component_count"], + "bom_items": bom["total_items"], + "pcb_components": pcb_analysis["board_info"]["components"], + } + finally: + client.close() + + +def main() -> int: + project_dir = create_fixture_project() + try: + component_db = run_component_db_checks(project_dir) + nexar = run_nexar_checks() + kicad_tools = run_kicad_tools_checks(project_dir) + print( + "OK:" + f" component_db.catalog={component_db['catalog_size']}" + f" component_db.search={component_db['search_results']}" + f" nexar.parts={nexar['parts_found']}" + f" nexar.demo={nexar['demo_mode']}" + f" kicad_tools.components={kicad_tools['components']}" + f" kicad_tools.bom_items={kicad_tools['bom_items']}" + f" kicad_tools.pcb_components={kicad_tools['pcb_components']}" + ) + return 0 + finally: + shutil.rmtree(project_dir, ignore_errors=True) + + +if __name__ == "__main__": + raise SystemExit(main())