nearxos/kicad-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: generate_netlist schematic_path, PinLocator cache, server.ts Python detection

Browse Source 
- connection_schematic.py: generate_netlist() now accepts schematic_path param,
  threaded through to get_net_connections() so PinLocator is actually invoked
  (previously only 1 connection per component was returned due to fallback break)
- kicad_interface.py: pass schematic_path to generate_netlist()
- pin_locator.py: add _schematic_cache to avoid loading Schematic() once per pin
  (was causing timeout: O(nets x components x pins) Schematic() calls)
- server.ts: remove fragile PYTHONPATH?.includes('KiCad') condition,
  always prefer KiCAD bundled Python on Windows when executable exists
- CHANGELOG.md: document fixes under v2.2.0-alpha
This commit is contained in:
Tom
2026-02-28 01:23:36 +01:00
parent 2945b52eae
commit 76503b144c
5 changed files with 967 additions and 650 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
CHANGELOG.md
View File

@@ -36,6 +36,27 @@ All notable changes to the KiCAD MCP Server project are documented here.
- `library.py`: Fix loop variable shadowing `Path` object (mypy)
- `design_rules.py`: Add type annotation for `violation_counts` (mypy)
### Pending fixes (not yet committed)
- `connection_schematic.py` / `kicad_interface.py`: Fix `generate_netlist` missing
`schematic_path` parameter – without it `get_net_connections` always fell back to
proximity matching which only returns one connection per component (first wire hit,
then `break`). PinLocator was never invoked. Fix: added `schematic_path: Optional[Path]`
to `generate_netlist` signature and threaded it through to `get_net_connections`,
and updated `_handle_generate_netlist` in `kicad_interface.py` to pass `schematic_path`.
- `server.ts`: Fix KiCAD bundled Python (3.11.5) not being selected on Windows – the
detection condition `process.env.PYTHONPATH?.includes("KiCad")` was fragile and failed
in some environments, causing System Python 3.12 to be used instead. Since `pcbnew.pyd`
is compiled for KiCAD's Python 3.11.5, this resulted in `No module named 'pcbnew'`.
Fix: removed the condition, KiCAD bundled Python is now always preferred on Windows
when it exists at `C:\Program Files\KiCad\9.0\bin\python.exe`.
Also added `KICAD_PYTHON` to `claude_desktop_config.json` as explicit override.
- `pin_locator.py`: Fix `generate_netlist` timeout – `get_pin_location` and
`get_all_symbol_pins` called `Schematic(schematic_path)` on every single pin lookup,
causing O(nets × components × pins) schematic file loads (e.g. 400+ loads for a
medium schematic). Fix: added `_schematic_cache` dict to `PinLocator.__init__`,
schematic is now loaded once per path and reused.
---
## [2.1.0-alpha] - 2026-01-10

192
python/commands/connection_schematic.py
View File

@@ -10,11 +10,13 @@ logger = logging.getLogger(__name__)
try:
from commands.wire_manager import WireManager
from commands.pin_locator import PinLocator
WIRE_MANAGER_AVAILABLE = True
except ImportError:
logger.warning("WireManager/PinLocator not available")
WIRE_MANAGER_AVAILABLE = False
class ConnectionManager:
"""Manage connections between components in schematics"""
@@ -29,7 +31,12 @@ class ConnectionManager:
return cls._pin_locator
@staticmethod
def add_wire(schematic_path: Path, start_point: list, end_point: list, properties: dict = None):
def add_wire(
schematic_path: Path,
start_point: list,
end_point: list,
properties: dict = None,
):
"""
Add a wire between two points using WireManager
@@ -47,11 +54,18 @@ class ConnectionManager:
logger.error("WireManager not available")
return False
stroke_width = properties.get('stroke_width', 0) if properties else 0
stroke_type = properties.get('stroke_type', 'default') if properties else 'default'
stroke_width = properties.get("stroke_width", 0) if properties else 0
stroke_type = (
properties.get("stroke_type", "default") if properties else "default"
)
success = WireManager.add_wire(schematic_path, start_point, end_point,
stroke_width=stroke_width, stroke_type=stroke_type)
success = WireManager.add_wire(
schematic_path,
start_point,
end_point,
stroke_width=stroke_width,
stroke_type=stroke_type,
)
return success
except Exception as e:
logger.error(f"Error adding wire: {e}")
@@ -70,7 +84,7 @@ class ConnectionManager:
[x, y] coordinates or None if pin not found
"""
try:
if not hasattr(symbol, 'pin'):
if not hasattr(symbol, "pin"):
logger.warning(f"Symbol {symbol.property.Reference.value} has no pins")
return None
@@ -82,7 +96,9 @@ class ConnectionManager:
break
if not target_pin:
logger.warning(f"Pin '{pin_name}' not found on {symbol.property.Reference.value}")
logger.warning(
f"Pin '{pin_name}' not found on {symbol.property.Reference.value}"
)
return None
# Get pin location relative to symbol
@@ -101,8 +117,14 @@ class ConnectionManager:
return None
@staticmethod
def add_connection(schematic_path: Path, source_ref: str, source_pin: str,
target_ref: str, target_pin: str, routing: str = 'direct'):
def add_connection(
schematic_path: Path,
source_ref: str,
source_pin: str,
target_ref: str,
target_pin: str,
routing: str = "direct",
):
"""
Add a wire connection between two component pins
@@ -128,31 +150,41 @@ class ConnectionManager:
return False
# Get pin locations
source_loc = locator.get_pin_location(schematic_path, source_ref, source_pin)
target_loc = locator.get_pin_location(schematic_path, target_ref, target_pin)
source_loc = locator.get_pin_location(
schematic_path, source_ref, source_pin
)
target_loc = locator.get_pin_location(
schematic_path, target_ref, target_pin
)
if not source_loc or not target_loc:
logger.error("Could not determine pin locations")
return False
# Create wire based on routing style
if routing == 'direct':
if routing == "direct":
# Simple direct wire
success = WireManager.add_wire(schematic_path, source_loc, target_loc)
elif routing == 'orthogonal_h':
elif routing == "orthogonal_h":
# Orthogonal routing (horizontal first)
path = WireManager.create_orthogonal_path(source_loc, target_loc, prefer_horizontal_first=True)
path = WireManager.create_orthogonal_path(
source_loc, target_loc, prefer_horizontal_first=True
)
success = WireManager.add_polyline_wire(schematic_path, path)
elif routing == 'orthogonal_v':
elif routing == "orthogonal_v":
# Orthogonal routing (vertical first)
path = WireManager.create_orthogonal_path(source_loc, target_loc, prefer_horizontal_first=False)
path = WireManager.create_orthogonal_path(
source_loc, target_loc, prefer_horizontal_first=False
)
success = WireManager.add_polyline_wire(schematic_path, path)
else:
logger.error(f"Unknown routing style: {routing}")
return False
if success:
logger.info(f"Connected {source_ref}/{source_pin} to {target_ref}/{target_pin} (routing: {routing})")
logger.info(
f"Connected {source_ref}/{source_pin} to {target_ref}/{target_pin} (routing: {routing})"
)
return True
else:
return False
@@ -160,6 +192,7 @@ class ConnectionManager:
except Exception as e:
logger.error(f"Error adding connection: {e}")
import traceback
logger.error(traceback.format_exc())
return False
@@ -177,13 +210,12 @@ class ConnectionManager:
Label object or None on error
"""
try:
if not hasattr(schematic, 'label'):
if not hasattr(schematic, "label"):
logger.error("Schematic does not have label collection")
return None
label = schematic.label.append(
text=net_name,
at={'x': position[0], 'y': position[1]}
text=net_name, at={"x": position[0], "y": position[1]}
)
logger.info(f"Added net label '{net_name}' at {position}")
return label
@@ -192,7 +224,9 @@ class ConnectionManager:
return None
@staticmethod
def connect_to_net(schematic_path: Path, component_ref: str, pin_name: str, net_name: str):
def connect_to_net(
schematic_path: Path, component_ref: str, pin_name: str, net_name: str
):
"""
Connect a component pin to a named net using a wire stub and label
@@ -231,7 +265,9 @@ class ConnectionManager:
return False
# Add label at the end of the stub using WireManager
label_success = WireManager.add_label(schematic_path, net_name, stub_end, label_type='label')
label_success = WireManager.add_label(
schematic_path, net_name, stub_end, label_type="label"
)
if not label_success:
logger.error(f"Failed to add net label '{net_name}'")
return False
@@ -242,11 +278,14 @@ class ConnectionManager:
except Exception as e:
logger.error(f"Error connecting to net: {e}")
import traceback
logger.error(traceback.format_exc())
return False
@staticmethod
def get_net_connections(schematic: Schematic, net_name: str, schematic_path: Optional[Path] = None):
def get_net_connections(
schematic: Schematic, net_name: str, schematic_path: Optional[Path] = None
):
"""
Get all connections for a named net using wire graph analysis
@@ -273,14 +312,14 @@ class ConnectionManager:
return dx < tolerance and dy < tolerance
# 1. Find all labels with this net name
if not hasattr(schematic, 'label'):
if not hasattr(schematic, "label"):
logger.warning("Schematic has no labels")
return connections
net_label_positions = []
for label in schematic.label:
if hasattr(label, 'value') and label.value == net_name:
if hasattr(label, 'at') and hasattr(label.at, 'value'):
if hasattr(label, "value") and label.value == net_name:
if hasattr(label, "at") and hasattr(label.at, "value"):
pos = label.at.value
net_label_positions.append([float(pos[0]), float(pos[1])])
@@ -288,21 +327,25 @@ class ConnectionManager:
logger.info(f"No labels found for net '{net_name}'")
return connections
logger.debug(f"Found {len(net_label_positions)} labels for net '{net_name}'")
logger.debug(
f"Found {len(net_label_positions)} labels for net '{net_name}'"
)
# 2. Find all wires connected to these label positions
if not hasattr(schematic, 'wire'):
if not hasattr(schematic, "wire"):
logger.warning("Schematic has no wires")
return connections
connected_wire_points = set()
for wire in schematic.wire:
if hasattr(wire, 'pts') and hasattr(wire.pts, 'xy'):
if hasattr(wire, "pts") and hasattr(wire.pts, "xy"):
# Get all points in this wire (polyline)
wire_points = []
for point in wire.pts.xy:
if hasattr(point, 'value'):
wire_points.append([float(point.value[0]), float(point.value[1])])
if hasattr(point, "value"):
wire_points.append(
[float(point.value[0]), float(point.value[1])]
)
# Check if any wire point touches a label
wire_connected = False
@@ -323,10 +366,12 @@ class ConnectionManager:
logger.debug(f"No wires connected to net '{net_name}' labels")
return connections
logger.debug(f"Found {len(connected_wire_points)} wire connection points for net '{net_name}'")
logger.debug(
f"Found {len(connected_wire_points)} wire connection points for net '{net_name}'"
)
# 3. Find component pins at wire endpoints
if not hasattr(schematic, 'symbol'):
if not hasattr(schematic, "symbol"):
logger.warning("Schematic has no symbols")
return connections
@@ -337,15 +382,15 @@ class ConnectionManager:
for symbol in schematic.symbol:
# Skip template symbols
if not hasattr(symbol.property, 'Reference'):
if not hasattr(symbol.property, "Reference"):
continue
ref = symbol.property.Reference.value
if ref.startswith('_TEMPLATE'):
if ref.startswith("_TEMPLATE"):
continue
# Get lib_id for pin location lookup
lib_id = symbol.lib_id.value if hasattr(symbol, 'lib_id') else None
lib_id = symbol.lib_id.value if hasattr(symbol, "lib_id") else None
if not lib_id:
continue
@@ -360,17 +405,18 @@ class ConnectionManager:
# Check each pin
for pin_num, pin_data in pins.items():
# Get pin location
pin_loc = locator.get_pin_location(schematic_path, ref, pin_num)
pin_loc = locator.get_pin_location(
schematic_path, ref, pin_num
)
if not pin_loc:
continue
# Check if pin coincides with any wire point
for wire_pt in connected_wire_points:
if points_coincide(pin_loc, list(wire_pt)):
connections.append({
"component": ref,
"pin": pin_num
})
connections.append(
{"component": ref, "pin": pin_num}
)
break # Pin found, no need to check more wire points
except Exception as e:
@@ -380,7 +426,7 @@ class ConnectionManager:
# Fallback: proximity-based matching if no PinLocator
if not locator or not schematic_path:
symbol_pos = symbol.at.value if hasattr(symbol, 'at') else None
symbol_pos = symbol.at.value if hasattr(symbol, "at") else None
if not symbol_pos:
continue
@@ -389,12 +435,11 @@ class ConnectionManager:
# Check if symbol is near any wire point (within 10mm)
for wire_pt in connected_wire_points:
dist = ((symbol_x - wire_pt[0])**2 + (symbol_y - wire_pt[1])**2)**0.5
dist = (
(symbol_x - wire_pt[0]) ** 2 + (symbol_y - wire_pt[1]) ** 2
) ** 0.5
if dist < 10.0: # 10mm proximity threshold
connections.append({
"component": ref,
"pin": "unknown"
})
connections.append({"component": ref, "pin": "unknown"})
break # Only add once per component
logger.info(f"Found {len(connections)} connections for net '{net_name}'")
@@ -403,14 +448,20 @@ class ConnectionManager:
except Exception as e:
logger.error(f"Error getting net connections: {e}")
import traceback
logger.error(traceback.format_exc())
return []
@staticmethod
def generate_netlist(schematic: Schematic):
def generate_netlist(schematic: Schematic, schematic_path: Optional[Path] = None):
"""
Generate a netlist from the schematic
Args:
schematic: Schematic object
schematic_path: Optional path to schematic file (enables accurate pin matching
via PinLocator; without it, only one connection per component is found)
Returns:
Dictionary with net information:
{
@@ -431,47 +482,58 @@ class ConnectionManager:
}
"""
try:
netlist = {
"nets": [],
"components": []
}
netlist = {"nets": [], "components": []}
# Gather all components
if hasattr(schematic, 'symbol'):
if hasattr(schematic, "symbol"):
for symbol in schematic.symbol:
component_info = {
"reference": symbol.property.Reference.value,
"value": symbol.property.Value.value if hasattr(symbol.property, 'Value') else "",
"footprint": symbol.property.Footprint.value if hasattr(symbol.property, 'Footprint') else ""
"value": (
symbol.property.Value.value
if hasattr(symbol.property, "Value")
else ""
),
"footprint": (
symbol.property.Footprint.value
if hasattr(symbol.property, "Footprint")
else ""
),
}
netlist["components"].append(component_info)
# Gather all nets from labels
if hasattr(schematic, 'label'):
if hasattr(schematic, "label"):
net_names = set()
for label in schematic.label:
if hasattr(label, 'value'):
if hasattr(label, "value"):
net_names.add(label.value)
# For each net, get connections
for net_name in net_names:
connections = ConnectionManager.get_net_connections(schematic, net_name)
connections = ConnectionManager.get_net_connections(
schematic, net_name, schematic_path
)
if connections:
netlist["nets"].append({
"name": net_name,
"connections": connections
})
netlist["nets"].append(
{"name": net_name, "connections": connections}
)
logger.info(f"Generated netlist with {len(netlist['nets'])} nets and {len(netlist['components'])} components")
logger.info(
f"Generated netlist with {len(netlist['nets'])} nets and {len(netlist['components'])} components"
)
return netlist
except Exception as e:
logger.error(f"Error generating netlist: {e}")
return {"nets": [], "components": []}
if __name__ == '__main__':
if __name__ == "__main__":
# Example Usage (for testing)
from schematic import SchematicManager # Assuming schematic.py is in the same directory
from schematic import (
SchematicManager,
) # Assuming schematic.py is in the same directory
# Create a new schematic
test_sch = SchematicManager.create_schematic("ConnectionTestSchematic")

150
python/commands/pin_locator.py
View File

@@ -14,7 +14,7 @@ import sexpdata
from sexpdata import Symbol
from skip import Schematic
logger = logging.getLogger('kicad_interface')
logger = logging.getLogger("kicad_interface")
class PinLocator:
@@ -23,6 +23,7 @@ class PinLocator:
def __init__(self):
"""Initialize pin locator with empty cache"""
self.pin_definition_cache = {} # Cache: "lib_id:symbol_name" -> pin_data
self._schematic_cache: Dict[str, object] = {} # Cache: path -> loaded Schematic
@staticmethod
def parse_symbol_definition(symbol_def: list) -> Dict[str, Dict]:
@@ -47,39 +48,39 @@ class PinLocator:
return
# Check if this is a pin definition
if len(sexp) > 0 and sexp[0] == Symbol('pin'):
if len(sexp) > 0 and sexp[0] == Symbol("pin"):
# Pin format: (pin type shape (at x y angle) (length len) (name "name") (number "num"))
pin_data = {
'x': 0,
'y': 0,
'angle': 0,
'length': 0,
'name': '',
'number': '',
'type': str(sexp[1]) if len(sexp) > 1 else 'passive'
"x": 0,
"y": 0,
"angle": 0,
"length": 0,
"name": "",
"number": "",
"type": str(sexp[1]) if len(sexp) > 1 else "passive",
}
# Extract pin attributes
for item in sexp:
if isinstance(item, list) and len(item) > 0:
if item[0] == Symbol('at') and len(item) >= 3:
pin_data['x'] = float(item[1])
pin_data['y'] = float(item[2])
if item[0] == Symbol("at") and len(item) >= 3:
pin_data["x"] = float(item[1])
pin_data["y"] = float(item[2])
if len(item) >= 4:
pin_data['angle'] = float(item[3])
pin_data["angle"] = float(item[3])
elif item[0] == Symbol('length') and len(item) >= 2:
pin_data['length'] = float(item[1])
elif item[0] == Symbol("length") and len(item) >= 2:
pin_data["length"] = float(item[1])
elif item[0] == Symbol('name') and len(item) >= 2:
pin_data['name'] = str(item[1]).strip('"')
elif item[0] == Symbol("name") and len(item) >= 2:
pin_data["name"] = str(item[1]).strip('"')
elif item[0] == Symbol('number') and len(item) >= 2:
pin_data['number'] = str(item[1]).strip('"')
elif item[0] == Symbol("number") and len(item) >= 2:
pin_data["number"] = str(item[1]).strip('"')
# Store by pin number
if pin_data['number']:
pins[pin_data['number']] = pin_data
if pin_data["number"]:
pins[pin_data["number"]] = pin_data
# Recurse into sublists
for item in sexp:
@@ -108,7 +109,7 @@ class PinLocator:
try:
# Read schematic
with open(schematic_path, 'r', encoding='utf-8') as f:
with open(schematic_path, "r", encoding="utf-8") as f:
sch_content = f.read()
sch_data = sexpdata.loads(sch_content)
@@ -116,7 +117,11 @@ class PinLocator:
# Find lib_symbols section
lib_symbols = None
for item in sch_data:
if isinstance(item, list) and len(item) > 0 and item[0] == Symbol('lib_symbols'):
if (
isinstance(item, list)
and len(item) > 0
and item[0] == Symbol("lib_symbols")
):
lib_symbols = item
break
@@ -126,7 +131,11 @@ class PinLocator:
# Find the specific symbol definition
for item in lib_symbols[1:]: # Skip 'lib_symbols' itself
if isinstance(item, list) and len(item) > 1 and item[0] == Symbol('symbol'):
if (
isinstance(item, list)
and len(item) > 1
and item[0] == Symbol("symbol")
):
symbol_name = str(item[1]).strip('"')
if symbol_name == lib_id:
# Found the symbol, parse pins
@@ -141,6 +150,7 @@ class PinLocator:
except Exception as e:
logger.error(f"Error getting symbol pins: {e}")
import traceback
logger.error(traceback.format_exc())
return {}
@@ -169,8 +179,9 @@ class PinLocator:
return (rotated_x, rotated_y)
def get_pin_location(self, schematic_path: Path, symbol_reference: str,
pin_number: str) -> Optional[List[float]]:
def get_pin_location(
self, schematic_path: Path, symbol_reference: str, pin_number: str
) -> Optional[List[float]]:
"""
Get the absolute location of a pin on a symbol instance
@@ -184,7 +195,11 @@ class PinLocator:
"""
try:
# Load schematic with kicad-skip to get symbol instance
sch = Schematic(str(schematic_path))
# Use cache to avoid reloading the file for every pin lookup
sch_key = str(schematic_path)
if sch_key not in self._schematic_cache:
self._schematic_cache[sch_key] = Schematic(sch_key)
sch = self._schematic_cache[sch_key]
# Find the symbol instance
target_symbol = None
@@ -205,12 +220,16 @@ class PinLocator:
symbol_rotation = float(symbol_at[2]) if len(symbol_at) > 2 else 0.0
# Get symbol lib_id
lib_id = target_symbol.lib_id.value if hasattr(target_symbol, 'lib_id') else None
lib_id = (
target_symbol.lib_id.value if hasattr(target_symbol, "lib_id") else None
)
if not lib_id:
logger.error(f"Symbol {symbol_reference} has no lib_id")
return None
logger.debug(f"Symbol {symbol_reference}: pos=({symbol_x}, {symbol_y}), rot={symbol_rotation}, lib_id={lib_id}")
logger.debug(
f"Symbol {symbol_reference}: pos=({symbol_x}, {symbol_y}), rot={symbol_rotation}, lib_id={lib_id}"
)
# Get pin definitions for this symbol
pins = self.get_symbol_pins(schematic_path, lib_id)
@@ -220,36 +239,49 @@ class PinLocator:
# Find the requested pin
if pin_number not in pins:
logger.error(f"Pin {pin_number} not found on {symbol_reference}. Available pins: {list(pins.keys())}")
logger.error(
f"Pin {pin_number} not found on {symbol_reference}. Available pins: {list(pins.keys())}"
)
return None
pin_data = pins[pin_number]
# Get pin position relative to symbol origin
pin_rel_x = pin_data['x']
pin_rel_y = pin_data['y']
pin_rel_x = pin_data["x"]
pin_rel_y = pin_data["y"]
logger.debug(f"Pin {pin_number} relative position: ({pin_rel_x}, {pin_rel_y})")
logger.debug(
f"Pin {pin_number} relative position: ({pin_rel_x}, {pin_rel_y})"
)
# Apply symbol rotation to pin position
if symbol_rotation != 0:
pin_rel_x, pin_rel_y = self.rotate_point(pin_rel_x, pin_rel_y, symbol_rotation)
logger.debug(f"After rotation {symbol_rotation}°: ({pin_rel_x}, {pin_rel_y})")
pin_rel_x, pin_rel_y = self.rotate_point(
pin_rel_x, pin_rel_y, symbol_rotation
)
logger.debug(
f"After rotation {symbol_rotation}°: ({pin_rel_x}, {pin_rel_y})"
)
# Calculate absolute position
abs_x = symbol_x + pin_rel_x
abs_y = symbol_y + pin_rel_y
logger.info(f"Pin {symbol_reference}/{pin_number} located at ({abs_x}, {abs_y})")
logger.info(
f"Pin {symbol_reference}/{pin_number} located at ({abs_x}, {abs_y})"
)
return [abs_x, abs_y]
except Exception as e:
logger.error(f"Error getting pin location: {e}")
import traceback
logger.error(traceback.format_exc())
return None
def get_all_symbol_pins(self, schematic_path: Path, symbol_reference: str) -> Dict[str, List[float]]:
def get_all_symbol_pins(
self, schematic_path: Path, symbol_reference: str
) -> Dict[str, List[float]]:
"""
Get locations of all pins on a symbol instance
@@ -261,8 +293,11 @@ class PinLocator:
Dictionary mapping pin number -> [x, y] coordinates
"""
try:
# Load schematic
sch = Schematic(str(schematic_path))
# Load schematic (use cache)
sch_key = str(schematic_path)
if sch_key not in self._schematic_cache:
self._schematic_cache[sch_key] = Schematic(sch_key)
sch = self._schematic_cache[sch_key]
# Find symbol
target_symbol = None
@@ -276,7 +311,9 @@ class PinLocator:
return {}
# Get lib_id
lib_id = target_symbol.lib_id.value if hasattr(target_symbol, 'lib_id') else None
lib_id = (
target_symbol.lib_id.value if hasattr(target_symbol, "lib_id") else None
)
if not lib_id:
logger.error(f"Symbol {symbol_reference} has no lib_id")
return {}
@@ -289,7 +326,9 @@ class PinLocator:
# Calculate location for each pin
result = {}
for pin_num in pins.keys():
location = self.get_pin_location(schematic_path, symbol_reference, pin_num)
location = self.get_pin_location(
schematic_path, symbol_reference, pin_num
)
if location:
result[pin_num] = location
@@ -301,10 +340,11 @@ class PinLocator:
return {}
if __name__ == '__main__':
if __name__ == "__main__":
# Test pin location discovery
import sys
sys.path.insert(0, '/home/chris/MCP/KiCAD-MCP-Server/python')
sys.path.insert(0, "/home/chris/MCP/KiCAD-MCP-Server/python")
from pathlib import Path
from commands.component_schematic import ComponentManager
@@ -316,8 +356,10 @@ if __name__ == '__main__':
print("=" * 80)
# Create test schematic with components (cross-platform temp directory)
test_path = Path(tempfile.gettempdir()) / 'test_pin_locator.kicad_sch'
template_path = Path('/home/chris/MCP/KiCAD-MCP-Server/python/templates/template_with_symbols_expanded.kicad_sch')
test_path = Path(tempfile.gettempdir()) / "test_pin_locator.kicad_sch"
template_path = Path(
"/home/chris/MCP/KiCAD-MCP-Server/python/templates/template_with_symbols_expanded.kicad_sch"
)
shutil.copy(template_path, test_path)
print(f"\n✓ Created test schematic: {test_path}")
@@ -327,11 +369,25 @@ if __name__ == '__main__':
sch = SchematicManager.load_schematic(str(test_path))
# Add resistor at (100, 100), rotation 0
r1_def = {'type': 'R', 'reference': 'R1', 'value': '10k', 'x': 100, 'y': 100, 'rotation': 0}
r1_def = {
"type": "R",
"reference": "R1",
"value": "10k",
"x": 100,
"y": 100,
"rotation": 0,
}
ComponentManager.add_component(sch, r1_def, test_path)
# Add capacitor at (150, 100), rotation 90
c1_def = {'type': 'C', 'reference': 'C1', 'value': '100nF', 'x': 150, 'y': 100, 'rotation': 90}
c1_def = {
"type": "C",
"reference": "C1",
"value": "100nF",
"x": 150,
"y": 100,
"rotation": 90,
}
ComponentManager.add_component(sch, c1_def, test_path)
SchematicManager.save_schematic(sch, str(test_path))

782
python/kicad_interface.py
View File

File diff suppressed because it is too large Load Diff

382
src/server.ts
View File

@@ -2,46 +2,46 @@
* KiCAD MCP Server implementation
*/
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import express from 'express';
import { spawn, exec, execSync, ChildProcess } from 'child_process';
import { existsSync } from 'fs';
import { join, dirname } from 'path';
import { logger } from './logger.js';
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import express from "express";
import { spawn, exec, execSync, ChildProcess } from "child_process";
import { existsSync } from "fs";
import { join, dirname } from "path";
import { logger } from "./logger.js";
// Import tool registration functions
import { registerProjectTools } from './tools/project.js';
import { registerBoardTools } from './tools/board.js';
import { registerComponentTools } from './tools/component.js';
import { registerRoutingTools } from './tools/routing.js';
import { registerDesignRuleTools } from './tools/design-rules.js';
import { registerExportTools } from './tools/export.js';
import { registerSchematicTools } from './tools/schematic.js';
import { registerLibraryTools } from './tools/library.js';
import { registerSymbolLibraryTools } from './tools/library-symbol.js';
import { registerJLCPCBApiTools } from './tools/jlcpcb-api.js';
import { registerUITools } from './tools/ui.js';
import { registerRouterTools } from './tools/router.js';
import { registerProjectTools } from "./tools/project.js";
import { registerBoardTools } from "./tools/board.js";
import { registerComponentTools } from "./tools/component.js";
import { registerRoutingTools } from "./tools/routing.js";
import { registerDesignRuleTools } from "./tools/design-rules.js";
import { registerExportTools } from "./tools/export.js";
import { registerSchematicTools } from "./tools/schematic.js";
import { registerLibraryTools } from "./tools/library.js";
import { registerSymbolLibraryTools } from "./tools/library-symbol.js";
import { registerJLCPCBApiTools } from "./tools/jlcpcb-api.js";
import { registerUITools } from "./tools/ui.js";
import { registerRouterTools } from "./tools/router.js";
// Import resource registration functions
import { registerProjectResources } from './resources/project.js';
import { registerBoardResources } from './resources/board.js';
import { registerComponentResources } from './resources/component.js';
import { registerLibraryResources } from './resources/library.js';
import { registerProjectResources } from "./resources/project.js";
import { registerBoardResources } from "./resources/board.js";
import { registerComponentResources } from "./resources/component.js";
import { registerLibraryResources } from "./resources/library.js";
// Import prompt registration functions
import { registerComponentPrompts } from './prompts/component.js';
import { registerRoutingPrompts } from './prompts/routing.js';
import { registerDesignPrompts } from './prompts/design.js';
import { registerComponentPrompts } from "./prompts/component.js";
import { registerRoutingPrompts } from "./prompts/routing.js";
import { registerDesignPrompts } from "./prompts/design.js";
/**
* Find the Python executable to use
* Prioritizes virtual environment if available, falls back to system Python
*/
function findPythonExecutable(scriptPath: string): string {
const isWindows = process.platform === 'win32';
const isMac = process.platform === 'darwin';
const isWindows = process.platform === "win32";
const isMac = process.platform === "darwin";
const isLinux = !isWindows && !isMac;
// Get the project root (parent of the python/ directory)
@@ -49,8 +49,18 @@ function findPythonExecutable(scriptPath: string): string {
// Check for virtual environment
const venvPaths = [
join(projectRoot, 'venv', isWindows ? 'Scripts' : 'bin', isWindows ? 'python.exe' : 'python'),
join(projectRoot, '.venv', isWindows ? 'Scripts' : 'bin', isWindows ? 'python.exe' : 'python'),
join(
projectRoot,
"venv",
isWindows ? "Scripts" : "bin",
isWindows ? "python.exe" : "python",
),
join(
projectRoot,
".venv",
isWindows ? "Scripts" : "bin",
isWindows ? "python.exe" : "python",
),
];
for (const venvPath of venvPaths) {
@@ -62,26 +72,28 @@ function findPythonExecutable(scriptPath: string): string {
// Allow override via KICAD_PYTHON environment variable (any platform)
if (process.env.KICAD_PYTHON) {
logger.info(`Using KICAD_PYTHON environment variable: ${process.env.KICAD_PYTHON}`);
logger.info(
`Using KICAD_PYTHON environment variable: ${process.env.KICAD_PYTHON}`,
);
return process.env.KICAD_PYTHON;
}
// Platform-specific KiCAD bundled Python detection
if (isWindows && process.env.PYTHONPATH?.includes('KiCad')) {
// Windows: Try KiCAD's bundled Python
const kicadPython = 'C:\\Program Files\\KiCad\\9.0\\bin\\python.exe';
if (isWindows) {
// Windows: Always prefer KiCAD's bundled Python (pcbnew.pyd is compiled for it)
const kicadPython = "C:\\Program Files\\KiCad\\9.0\\bin\\python.exe";
if (existsSync(kicadPython)) {
logger.info(`Found KiCAD bundled Python at: ${kicadPython}`);
return kicadPython;
}
} else if (isMac) {
// macOS: Try KiCAD's bundled Python (check multiple versions and locations)
const kicadPythonVersions = ['3.9', '3.10', '3.11', '3.12', '3.13'];
const kicadPythonVersions = ["3.9", "3.10", "3.11", "3.12", "3.13"];
// Standard KiCAD installation paths
const kicadAppPaths = [
'/Applications/KiCad/KiCad.app',
'/Applications/KiCAD/KiCad.app', // Alternative capitalization
"/Applications/KiCad/KiCad.app",
"/Applications/KiCAD/KiCad.app", // Alternative capitalization
`${process.env.HOME}/Applications/KiCad/KiCad.app`, // User Applications folder
];
@@ -98,24 +110,26 @@ function findPythonExecutable(scriptPath: string): string {
// Fallback to Homebrew Python (if pcbnew is installed via pip)
const homebrewPaths = [
'/opt/homebrew/bin/python3', // Apple Silicon
'/usr/local/bin/python3', // Intel Mac
'/opt/homebrew/bin/python3.12',
'/opt/homebrew/bin/python3.11',
"/opt/homebrew/bin/python3", // Apple Silicon
"/usr/local/bin/python3", // Intel Mac
"/opt/homebrew/bin/python3.12",
"/opt/homebrew/bin/python3.11",
];
for (const path of homebrewPaths) {
if (existsSync(path)) {
logger.info(`Found Homebrew Python at: ${path} (ensure pcbnew is importable)`);
logger.info(
`Found Homebrew Python at: ${path} (ensure pcbnew is importable)`,
);
return path;
}
}
} else if (isLinux) {
// Linux: Try KiCAD bundled Python locations first
const linuxKicadPaths = [
'/usr/lib/kicad/bin/python3',
'/usr/local/lib/kicad/bin/python3',
'/opt/kicad/bin/python3',
"/usr/lib/kicad/bin/python3",
"/usr/local/lib/kicad/bin/python3",
"/opt/kicad/bin/python3",
];
for (const path of linuxKicadPaths) {
@@ -127,17 +141,17 @@ function findPythonExecutable(scriptPath: string): string {
// Resolve system python3 to full path using 'which'
try {
const result = execSync('which python3', { encoding: 'utf-8' }).trim();
const result = execSync("which python3", { encoding: "utf-8" }).trim();
if (result && existsSync(result)) {
logger.info(`Resolved system Python via which: ${result}`);
return result;
}
} catch (e) {
logger.warn('Failed to resolve python3 via which command');
logger.warn("Failed to resolve python3 via which command");
}
// Fallback to common system paths
const systemPaths = ['/usr/bin/python3', '/bin/python3'];
const systemPaths = ["/usr/bin/python3", "/bin/python3"];
for (const path of systemPaths) {
if (existsSync(path)) {
logger.info(`Found system Python at: ${path}`);
@@ -147,8 +161,8 @@ function findPythonExecutable(scriptPath: string): string {
}
// Default to system Python (last resort)
logger.info('Using system Python (no venv found)');
return isWindows ? 'python.exe' : 'python3';
logger.info("Using system Python (no venv found)");
return isWindows ? "python.exe" : "python3";
}
/**
@@ -159,10 +173,18 @@ export class KiCADMcpServer {
private pythonProcess: ChildProcess | null = null;
private kicadScriptPath: string;
private stdioTransport!: StdioServerTransport;
private requestQueue: Array<{ request: any, resolve: Function, reject: Function }> = [];
private requestQueue: Array<{
request: any;
resolve: Function;
reject: Function;
}> = [];
private processingRequest = false;
private responseBuffer: string = '';
private currentRequestHandler: { resolve: Function, reject: Function, timeoutHandle: NodeJS.Timeout } | null = null;
private responseBuffer: string = "";
private currentRequestHandler: {
resolve: Function;
reject: Function;
timeoutHandle: NodeJS.Timeout;
} | null = null;
/**
* Constructor for the KiCAD MCP Server
@@ -171,7 +193,7 @@ export class KiCADMcpServer {
*/
constructor(
kicadScriptPath: string,
logLevel: 'error' | 'warn' | 'info' | 'debug' = 'info'
logLevel: "error" | "warn" | "info" | "debug" = "info",
) {
// Set up the logger
logger.setLogLevel(logLevel);
@@ -179,19 +201,21 @@ export class KiCADMcpServer {
// Check if KiCAD script exists
this.kicadScriptPath = kicadScriptPath;
if (!existsSync(this.kicadScriptPath)) {
throw new Error(`KiCAD interface script not found: ${this.kicadScriptPath}`);
throw new Error(
`KiCAD interface script not found: ${this.kicadScriptPath}`,
);
}
// Initialize the MCP server
this.server = new McpServer({
name: 'kicad-mcp-server',
version: '1.0.0',
description: 'MCP server for KiCAD PCB design operations'
name: "kicad-mcp-server",
version: "1.0.0",
description: "MCP server for KiCAD PCB design operations",
});
// Initialize STDIO transport
this.stdioTransport = new StdioServerTransport();
logger.info('Using STDIO transport for local communication');
logger.info("Using STDIO transport for local communication");
// Register tools, resources, and prompts
this.registerAll();
@@ -201,7 +225,7 @@ export class KiCADMcpServer {
* Register all tools, resources, and prompts
*/
private registerAll(): void {
logger.info('Registering KiCAD tools, resources, and prompts...');
logger.info("Registering KiCAD tools, resources, and prompts...");
// Register router tools FIRST (for tool discovery and execution)
registerRouterTools(this.server, this.callKicadScript.bind(this));
@@ -230,20 +254,26 @@ export class KiCADMcpServer {
registerRoutingPrompts(this.server);
registerDesignPrompts(this.server);
logger.info('All KiCAD tools, resources, and prompts registered');
logger.info('Router pattern enabled: 4 router tools + direct tools for discovery');
logger.info("All KiCAD tools, resources, and prompts registered");
logger.info(
"Router pattern enabled: 4 router tools + direct tools for discovery",
);
}
/**
* Validate prerequisites before starting the server
*/
private async validatePrerequisites(pythonExe: string): Promise<boolean> {
const isWindows = process.platform === 'win32';
const isLinux = process.platform !== 'win32' && process.platform !== 'darwin';
const isWindows = process.platform === "win32";
const isLinux =
process.platform !== "win32" && process.platform !== "darwin";
const errors: string[] = [];
// Check if Python executable exists (for absolute paths) or is executable (for commands)
const isAbsolutePath = pythonExe.startsWith('/') || pythonExe.startsWith('C:') || pythonExe.startsWith('\\');
const isAbsolutePath =
pythonExe.startsWith("/") ||
pythonExe.startsWith("C:") ||
pythonExe.startsWith("\\");
if (isAbsolutePath) {
// Absolute path: use existsSync
@@ -251,42 +281,61 @@ export class KiCADMcpServer {
errors.push(`Python executable not found: ${pythonExe}`);
if (isWindows) {
errors.push('Windows: Install KiCAD 9.0+ from https://www.kicad.org/download/windows/');
errors.push('Or run: .\\setup-windows.ps1 for automatic configuration');
errors.push(
"Windows: Install KiCAD 9.0+ from https://www.kicad.org/download/windows/",
);
errors.push(
"Or run: .\\setup-windows.ps1 for automatic configuration",
);
} else if (isLinux) {
errors.push('Linux: Install KiCAD 9.0+ or set KICAD_PYTHON environment variable');
errors.push('Set KICAD_PYTHON to specify a custom Python path');
errors.push(
"Linux: Install KiCAD 9.0+ or set KICAD_PYTHON environment variable",
);
errors.push("Set KICAD_PYTHON to specify a custom Python path");
}
}
} else {
// Command name: verify it's executable via --version test
logger.info(`Validating command-based Python executable: ${pythonExe}`);
try {
const { stdout } = await new Promise<{stdout: string, stderr: string}>((resolve, reject) => {
exec(`"${pythonExe}" --version`, {
const { stdout } = await new Promise<{
stdout: string;
stderr: string;
}>((resolve, reject) => {
exec(
`"${pythonExe}" --version`,
{
timeout: 3000,
env: { ...process.env }
}, (error: any, stdout: string, stderr: string) => {
env: { ...process.env },
},
(error: any, stdout: string, stderr: string) => {
if (error) {
reject(error);
} else {
resolve({ stdout, stderr });
}
});
},
);
});
logger.info(`Python version check passed: ${stdout.trim()}`);
} catch (error: any) {
errors.push(`Python executable not found in PATH: ${pythonExe}`);
errors.push(`Error: ${error.message}`);
errors.push('Set KICAD_PYTHON environment variable to specify full path');
errors.push(
"Set KICAD_PYTHON environment variable to specify full path",
);
if (isLinux) {
errors.push('');
errors.push('Linux troubleshooting:');
errors.push('1. Check if python3 is installed: which python3');
errors.push('2. Install KiCAD: sudo apt install kicad (Ubuntu/Debian)');
errors.push('3. Set KICAD_PYTHON=/usr/bin/python3 in your MCP config');
errors.push("");
errors.push("Linux troubleshooting:");
errors.push("1. Check if python3 is installed: which python3");
errors.push(
"2. Install KiCAD: sudo apt install kicad (Ubuntu/Debian)",
);
errors.push(
"3. Set KICAD_PYTHON=/usr/bin/python3 in your MCP config",
);
}
}
}
@@ -297,76 +346,91 @@ export class KiCADMcpServer {
}
// Check if dist/index.js exists (if running from compiled code)
const distPath = join(dirname(dirname(this.kicadScriptPath)), 'dist', 'index.js');
const distPath = join(
dirname(dirname(this.kicadScriptPath)),
"dist",
"index.js",
);
if (!existsSync(distPath)) {
errors.push('Project not built. Run: npm run build');
errors.push("Project not built. Run: npm run build");
}
// Try to test pcbnew import (quick validation)
if (existsSync(pythonExe) && existsSync(this.kicadScriptPath)) {
logger.info('Validating pcbnew module access...');
logger.info("Validating pcbnew module access...");
const testCommand = `"${pythonExe}" -c "import pcbnew; print('OK')"`;
try {
const { stdout, stderr } = await new Promise<{stdout: string, stderr: string}>((resolve, reject) => {
exec(testCommand, {
const { stdout, stderr } = await new Promise<{
stdout: string;
stderr: string;
}>((resolve, reject) => {
exec(
testCommand,
{
timeout: 5000,
env: { ...process.env }
}, (error: any, stdout: string, stderr: string) => {
env: { ...process.env },
},
(error: any, stdout: string, stderr: string) => {
if (error) {
reject(error);
} else {
resolve({ stdout, stderr });
}
});
},
);
});
if (!stdout.includes('OK')) {
errors.push('pcbnew module import test failed');
if (!stdout.includes("OK")) {
errors.push("pcbnew module import test failed");
errors.push(`Output: ${stdout}`);
errors.push(`Errors: ${stderr}`);
if (isWindows) {
errors.push('');
errors.push('Windows troubleshooting:');
errors.push('1. Set PYTHONPATH=C:\\Program Files\\KiCad\\9.0\\lib\\python3\\dist-packages');
errors.push('2. Test: "C:\\Program Files\\KiCad\\9.0\\bin\\python.exe" -c "import pcbnew"');
errors.push('3. Run: .\\setup-windows.ps1 for automatic fix');
errors.push('4. See: docs/WINDOWS_TROUBLESHOOTING.md');
errors.push("");
errors.push("Windows troubleshooting:");
errors.push(
"1. Set PYTHONPATH=C:\\Program Files\\KiCad\\9.0\\lib\\python3\\dist-packages",
);
errors.push(
'2. Test: "C:\\Program Files\\KiCad\\9.0\\bin\\python.exe" -c "import pcbnew"',
);
errors.push("3. Run: .\\setup-windows.ps1 for automatic fix");
errors.push("4. See: docs/WINDOWS_TROUBLESHOOTING.md");
}
} else {
logger.info('✓ pcbnew module validated successfully');
logger.info("✓ pcbnew module validated successfully");
}
} catch (error: any) {
errors.push(`pcbnew validation failed: ${error.message}`);
if (isWindows) {
errors.push('');
errors.push('This usually means:');
errors.push('- KiCAD is not installed');
errors.push('- PYTHONPATH is incorrect');
errors.push('- Python cannot find pcbnew module');
errors.push('');
errors.push('Quick fix: Run .\\setup-windows.ps1');
errors.push("");
errors.push("This usually means:");
errors.push("- KiCAD is not installed");
errors.push("- PYTHONPATH is incorrect");
errors.push("- Python cannot find pcbnew module");
errors.push("");
errors.push("Quick fix: Run .\\setup-windows.ps1");
}
}
}
// Log all errors
if (errors.length > 0) {
logger.error('='.repeat(70));
logger.error('STARTUP VALIDATION FAILED');
logger.error('='.repeat(70));
errors.forEach(err => logger.error(err));
logger.error('='.repeat(70));
logger.error("=".repeat(70));
logger.error("STARTUP VALIDATION FAILED");
logger.error("=".repeat(70));
errors.forEach((err) => logger.error(err));
logger.error("=".repeat(70));
// Also write to stderr for Claude Desktop to capture
process.stderr.write('\n' + '='.repeat(70) + '\n');
process.stderr.write('KiCAD MCP Server - Startup Validation Failed\n');
process.stderr.write('='.repeat(70) + '\n');
errors.forEach(err => process.stderr.write(err + '\n'));
process.stderr.write('='.repeat(70) + '\n\n');
process.stderr.write("\n" + "=".repeat(70) + "\n");
process.stderr.write("KiCAD MCP Server - Startup Validation Failed\n");
process.stderr.write("=".repeat(70) + "\n");
errors.forEach((err) => process.stderr.write(err + "\n"));
process.stderr.write("=".repeat(70) + "\n\n");
return false;
}
@@ -379,10 +443,12 @@ export class KiCADMcpServer {
*/
async start(): Promise<void> {
try {
logger.info('Starting KiCAD MCP server...');
logger.info("Starting KiCAD MCP server...");
// Start the Python process for KiCAD scripting
logger.info(`Starting Python process with script: ${this.kicadScriptPath}`);
logger.info(
`Starting Python process with script: ${this.kicadScriptPath}`,
);
const pythonExe = findPythonExecutable(this.kicadScriptPath);
logger.info(`Using Python executable: ${pythonExe}`);
@@ -390,55 +456,61 @@ export class KiCADMcpServer {
// Validate prerequisites
const isValid = await this.validatePrerequisites(pythonExe);
if (!isValid) {
throw new Error('Prerequisites validation failed. See logs above for details.');
throw new Error(
"Prerequisites validation failed. See logs above for details.",
);
}
this.pythonProcess = spawn(pythonExe, [this.kicadScriptPath], {
stdio: ['pipe', 'pipe', 'pipe'],
stdio: ["pipe", "pipe", "pipe"],
env: {
...process.env,
PYTHONPATH: process.env.PYTHONPATH || 'C:/Program Files/KiCad/9.0/lib/python3/dist-packages'
}
PYTHONPATH:
process.env.PYTHONPATH ||
"C:/Program Files/KiCad/9.0/lib/python3/dist-packages",
},
});
// Listen for process exit
this.pythonProcess.on('exit', (code, signal) => {
logger.warn(`Python process exited with code ${code} and signal ${signal}`);
this.pythonProcess.on("exit", (code, signal) => {
logger.warn(
`Python process exited with code ${code} and signal ${signal}`,
);
this.pythonProcess = null;
});
// Listen for process errors
this.pythonProcess.on('error', (err) => {
this.pythonProcess.on("error", (err) => {
logger.error(`Python process error: ${err.message}`);
});
// Set up error logging for stderr
if (this.pythonProcess.stderr) {
this.pythonProcess.stderr.on('data', (data: Buffer) => {
this.pythonProcess.stderr.on("data", (data: Buffer) => {
logger.error(`Python stderr: ${data.toString()}`);
});
}
// Set up persistent stdout handler (instead of adding/removing per request)
if (this.pythonProcess.stdout) {
this.pythonProcess.stdout.on('data', (data: Buffer) => {
this.pythonProcess.stdout.on("data", (data: Buffer) => {
this.handlePythonResponse(data);
});
}
// Connect server to STDIO transport
logger.info('Connecting MCP server to STDIO transport...');
logger.info("Connecting MCP server to STDIO transport...");
try {
await this.server.connect(this.stdioTransport);
logger.info('Successfully connected to STDIO transport');
logger.info("Successfully connected to STDIO transport");
} catch (error) {
logger.error(`Failed to connect to STDIO transport: ${error}`);
throw error;
}
// Write a ready message to stderr (for debugging)
process.stderr.write('KiCAD MCP SERVER READY\n');
process.stderr.write("KiCAD MCP SERVER READY\n");
logger.info('KiCAD MCP server started and ready');
logger.info("KiCAD MCP server started and ready");
} catch (error) {
logger.error(`Failed to start KiCAD MCP server: ${error}`);
throw error;
@@ -449,7 +521,7 @@ export class KiCADMcpServer {
* Stop the MCP server and clean up resources
*/
async stop(): Promise<void> {
logger.info('Stopping KiCAD MCP server...');
logger.info("Stopping KiCAD MCP server...");
// Kill the Python process if it's running
if (this.pythonProcess) {
@@ -457,7 +529,7 @@ export class KiCADMcpServer {
this.pythonProcess = null;
}
logger.info('KiCAD MCP server stopped');
logger.info("KiCAD MCP server stopped");
}
/**
@@ -471,7 +543,7 @@ export class KiCADMcpServer {
return new Promise((resolve, reject) => {
// Check if Python process is running
if (!this.pythonProcess) {
logger.error('Python process is not running');
logger.error("Python process is not running");
reject(new Error("Python process for KiCAD scripting is not running"));
return;
}
@@ -479,17 +551,24 @@ export class KiCADMcpServer {
// Determine timeout based on command type
// DRC and export operations need longer timeouts for large boards
let commandTimeout = 30000; // Default 30 seconds
const longRunningCommands = ['run_drc', 'export_gerber', 'export_pdf', 'export_3d'];
const longRunningCommands = [
"run_drc",
"export_gerber",
"export_pdf",
"export_3d",
];
if (longRunningCommands.includes(command)) {
commandTimeout = 600000; // 10 minutes for long operations
logger.info(`Using extended timeout (${commandTimeout/1000}s) for command: ${command}`);
logger.info(
`Using extended timeout (${commandTimeout / 1000}s) for command: ${command}`,
);
}
// Add request to queue with timeout info
this.requestQueue.push({
request: { command, params, timeout: commandTimeout },
resolve,
reject
reject,
});
// Process the queue if not already processing
@@ -519,8 +598,10 @@ export class KiCADMcpServer {
if (!this.currentRequestHandler) {
// No pending request, clear buffer if it has data (shouldn't happen)
if (this.responseBuffer.trim()) {
logger.warn(`Received data with no pending request: ${this.responseBuffer.substring(0, 100)}...`);
this.responseBuffer = '';
logger.warn(
`Received data with no pending request: ${this.responseBuffer.substring(0, 100)}...`,
);
this.responseBuffer = "";
}
return;
}
@@ -530,7 +611,9 @@ export class KiCADMcpServer {
const result = JSON.parse(this.responseBuffer);
// If we get here, we have a valid JSON response
logger.debug(`Completed KiCAD command with result: ${result.success ? 'success' : 'failure'}`);
logger.debug(
`Completed KiCAD command with result: ${result.success ? "success" : "failure"}`,
);
// Clear the timeout since we got a response
if (this.currentRequestHandler.timeoutHandle) {
@@ -541,7 +624,7 @@ export class KiCADMcpServer {
const handler = this.currentRequestHandler;
// Clear state
this.responseBuffer = '';
this.responseBuffer = "";
this.currentRequestHandler = null;
this.processingRequest = false;
@@ -550,7 +633,6 @@ export class KiCADMcpServer {
// Process next request if any
setTimeout(() => this.processNextRequest(), 0);
} catch (e) {
// Not a complete JSON yet, keep collecting data
// This is normal for large responses that come in chunks
@@ -579,21 +661,29 @@ export class KiCADMcpServer {
const requestStr = JSON.stringify(request);
// Clear response buffer for new request
this.responseBuffer = '';
this.responseBuffer = "";
// Set a timeout (use command-specific timeout or default)
const timeoutDuration = request.timeout || 30000;
const timeoutHandle = setTimeout(() => {
logger.error(`Command timeout after ${timeoutDuration/1000}s: ${request.command}`);
logger.error(`Buffer contents: ${this.responseBuffer.substring(0, 200)}...`);
logger.error(
`Command timeout after ${timeoutDuration / 1000}s: ${request.command}`,
);
logger.error(
`Buffer contents: ${this.responseBuffer.substring(0, 200)}...`,
);
// Clear state
this.responseBuffer = '';
this.responseBuffer = "";
this.currentRequestHandler = null;
this.processingRequest = false;
// Reject the promise
reject(new Error(`Command timeout after ${timeoutDuration/1000}s: ${request.command}`));
reject(
new Error(
`Command timeout after ${timeoutDuration / 1000}s: ${request.command}`,
),
);
// Process next request
setTimeout(() => this.processNextRequest(), 0);
@@ -604,7 +694,7 @@ export class KiCADMcpServer {
// Write the request to the Python process
logger.debug(`Sending request: ${requestStr}`);
this.pythonProcess?.stdin?.write(requestStr + '\n');
this.pythonProcess?.stdin?.write(requestStr + "\n");
} catch (error) {
logger.error(`Error processing request: ${error}`);