PCB Parts MCP Server

Instructions

Implementation Reference

• src/pcbparts_mcp/server.py:933-972 (handler)

  The main handler function for the cse_get_kicad tool. It validates that the CSE client is initialized, checks that either query or part_id is provided, validates query length, and delegates to _cse_client.get_kicad(). Returns error dict on failure.

  async def cse_get_kicad(
      query: str | None = None,
      part_id: int | None = None,
  ) -> dict:
      """Get KiCad schematic symbol and PCB footprint for any component.
  
      Downloads from SamacSys ComponentSearchEngine. Works for any manufacturer's part,
      not limited to JLCPCB. Returns the raw .kicad_sym and .kicad_mod file contents
      as text that can be read directly or saved to a KiCad project.
  
      WARNING: This tool is slow (up to 45s response time). Only use when you specifically need
      KiCad symbol/footprint files. For checking if an EasyEDA footprint exists, use jlc_get_part instead.
  
      Args:
          query: MPN to search for (e.g., "LM358P", "STM32F103CBT6", "ESP32-WROOM-32E").
                 Finds the best matching part with an available model.
          part_id: CSE part ID from a previous cse_search result (skips search step).
                 Use this if you already know the exact part.
  
      One of query or part_id must be provided.
  
      Returns:
          kicad_symbol: Raw .kicad_sym file content (pin names, types, graphical symbol)
          kicad_footprint: Raw .kicad_mod file content (pad layout, silkscreen, courtyard)
          part_id: CSE part ID (for future lookups)
          mfr_part_number, manufacturer, description: Part metadata (when searched by query)
      """
      if not _cse_client:
          return {"error": "CSE client not initialized"}
  
      if not query and not part_id:
          return {"error": "Must provide either query or part_id"}
      if query and len(query) > 500:
          return {"error": "Query too long (max 500 characters)"}
  
      try:
          return await _cse_client.get_kicad(query=query, part_id=part_id)
      except Exception as e:
          logger.error(f"CSE get_kicad failed: {type(e).__name__}: {e}")
          return {"error": "CSE get_kicad failed. Check server logs for details."}

• src/pcbparts_mcp/server.py:924-932 (registration)

  Tool registration via @mcp.tool() decorator. Defines tool annotations including title 'Get KiCad Symbol & Footprint', and hints (readOnly, non-destructive, idempotent, openWorld).

  @mcp.tool(
      annotations=ToolAnnotations(
          title="Get KiCad Symbol & Footprint",
          readOnlyHint=True,
          destructiveHint=False,
          idempotentHint=True,
          openWorldHint=True,
      )
  )

• src/pcbparts_mcp/cse.py:174-278 (helper)

  The CSEClient.get_kicad() method contains the actual implementation logic. It resolves part_id from query via search, downloads the KiCad model zip from CSE authenticated endpoint, extracts .kicad_sym and .kicad_mod files, handles zip bomb protection, caches results, and returns the symbol/footprint content.

  async def get_kicad(self, query: str | None = None, part_id: int | None = None) -> dict[str, Any]:
      """Download and extract KiCad symbol + footprint for a part.
  
      Args:
          query: MPN to search for (used to find part_id if not provided)
          part_id: CSE part ID (from a previous cse_search result)
  
      Returns:
          Dict with kicad_symbol, kicad_footprint text content, plus part metadata
      """
      if not CSE_USER or not CSE_PASS:
          return {"error": "CSE credentials not configured. Set CSEARCH_USER and CSEARCH_PASS in environment."}
  
      # Resolve part_id from query if needed
      resolved_part_id = part_id
      part_info: dict[str, Any] = {}
  
      if not resolved_part_id:
          if not query:
              return {"error": "Must provide either query or part_id"}
  
          search_result = await self.search(query)
          results = search_result.get("results", [])
          if not results:
              return {"error": f"No parts found for '{query}'"}
  
          # Find first result with a model available
          for r in results:
              if r.get("has_model") and r.get("cse_part_id"):
                  resolved_part_id = r["cse_part_id"]
                  part_info = r
                  break
  
          if not resolved_part_id:
              return {"error": f"No ECAD model available for '{query}'"}
  
      # Check kicad cache
      cache_key = f"kicad:{resolved_part_id}"
      cached = self._kicad_cache.get(cache_key)
      if cached is not None:
          return {
              "part_id": resolved_part_id,
              **part_info,
              "kicad_symbol": cached.get("kicad_symbol"),
              "kicad_footprint": cached.get("kicad_footprint"),
          }
  
      # Download the model zip (authenticated)
      try:
          async with self._get_semaphore():
              response = await self._get_http().get(
                  _CSE_MODEL_URL,
                  params={"partID": str(resolved_part_id)},
                  auth=(CSE_USER, CSE_PASS),
                  timeout=30,
              )
              await asyncio.sleep(CSE_RATE_LIMIT)
      except httpx.HTTPError:
          return {"error": "CSE model download failed (network/connection error)"}
  
      if response.status_code == 401:
          return {"error": "CSE authentication failed. Check CSEARCH_USER and CSEARCH_PASS."}
      if response.status_code != 200:
          return {"error": f"CSE model download failed (HTTP {response.status_code})"}
  
      # Check response size to prevent zip bombs
      content_length = len(response.content)
      if content_length > _MAX_ZIP_SIZE:
          return {"error": f"CSE model download too large ({content_length} bytes, max {_MAX_ZIP_SIZE})"}
  
      # Extract KiCad files from zip in memory — no temp files
      kicad_files: dict[str, str] = {}
      try:
          with zipfile.ZipFile(io.BytesIO(response.content)) as zf:
              for name in zf.namelist():
                  lower = name.lower()
                  # Only extract from the KiCad directory
                  if "/kicad/" not in lower:
                      continue
                  # Check decompressed size before reading
                  info = zf.getinfo(name)
                  if info.file_size > 10 * 1024 * 1024:  # 10MB per file max
                      logger.warning(f"Skipping oversized file in zip: {name} ({info.file_size} bytes)")
                      continue
                  if any(lower.endswith(ext) for ext in _KICAD_EXTENSIONS):
                      content = zf.read(name).decode("utf-8", errors="replace")
                      if lower.endswith(".kicad_sym"):
                          kicad_files["kicad_symbol"] = content
                      elif lower.endswith(".kicad_mod"):
                          kicad_files["kicad_footprint"] = content
      except zipfile.BadZipFile:
          return {"error": "CSE returned invalid zip file"}
  
      if not kicad_files:
          return {"error": f"No KiCad files found in model archive for part {resolved_part_id}"}
  
      # Cache the extracted text (zip is discarded)
      self._kicad_cache.set(cache_key, kicad_files)
  
      return {
          "part_id": resolved_part_id,
          **part_info,
          "kicad_symbol": kicad_files.get("kicad_symbol"),
          "kicad_footprint": kicad_files.get("kicad_footprint"),
      }

• src/pcbparts_mcp/cse.py:87-109 (helper)

  CSEClient class definition with initialization. Sets up TTL caches for search results and KiCad files, semaphore for rate limiting, and HTTP client configuration.

  class CSEClient:
      """Async client for ComponentSearchEngine (SamacSys) search.
  
      Uses the alligator JSON API on the RS subdomain.
      No API key or authentication required.
      """
  
      def __init__(self):
          self._cache = TTLCache(ttl=CSE_CACHE_TTL)
          self._kicad_cache = TTLCache(ttl=CSE_KICAD_CACHE_TTL, max_size=CSE_KICAD_CACHE_MAX_SIZE)
          self._semaphore: asyncio.Semaphore | None = None
          self._http: httpx.AsyncClient | None = None
  
      def _get_semaphore(self) -> asyncio.Semaphore:
          # Safe in single-threaded asyncio: no await between None check and assignment
          if self._semaphore is None:
              self._semaphore = asyncio.Semaphore(CSE_CONCURRENT_LIMIT)
          return self._semaphore
  
      def _get_http(self) -> httpx.AsyncClient:
          if self._http is None:
              self._http = httpx.AsyncClient(timeout=CSE_REQUEST_TIMEOUT)
          return self._http