L'électron rare
08277791be
- Enhanced README.md with sponsor iframe and improved formatting. - Added README_FR.md for French localization of the project. - Refactored type hints in kicad_symbol_sqlite.py and nexar.py for consistency. - Simplified exception handling in __init__.py. - Cleaned up imports and removed unused variables in various plugin files. - Introduced CI workflow for repository sanity checks. - Added smoke tests documentation in tests/README.md.
641 lines
26 KiB
Python
641 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Nexar API MCP Server (formerly Octopart)
|
|
Provides comprehensive component pricing from multiple distributors via Nexar GraphQL API
|
|
Much simpler than individual distributor APIs!
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import asyncio
|
|
import os
|
|
import logging
|
|
from typing import Dict, Any
|
|
|
|
try:
|
|
import requests
|
|
REQUESTS_AVAILABLE = True
|
|
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') 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:
|
|
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"""
|
|
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": "2025-03-26",
|
|
"capabilities": {
|
|
"tools": {}
|
|
},
|
|
"serverInfo": {
|
|
"name": "nexar-api",
|
|
"version": "1.0.0"
|
|
}
|
|
}
|
|
}
|
|
|
|
elif method == "tools/list":
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"result": {
|
|
"tools": [
|
|
{
|
|
"name": "search_parts",
|
|
"description": "Search for parts across all distributors",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string", "description": "Part number, manufacturer, or description"},
|
|
"category": {"type": "string", "description": "Component category"},
|
|
"manufacturer": {"type": "string", "description": "Manufacturer name"},
|
|
"limit": {"type": "integer", "description": "Max results", "default": 10}
|
|
},
|
|
"required": ["query"]
|
|
}
|
|
},
|
|
{
|
|
"name": "get_part_pricing",
|
|
"description": "Get comprehensive pricing from all distributors",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"mpn": {"type": "string", "description": "Manufacturer part number"},
|
|
"manufacturer": {"type": "string", "description": "Manufacturer name"},
|
|
"quantity": {"type": "integer", "description": "Desired quantity", "default": 1}
|
|
},
|
|
"required": ["mpn"]
|
|
}
|
|
},
|
|
{
|
|
"name": "get_best_price",
|
|
"description": "Find best price for a component across all distributors",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"parts": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"mpn": {"type": "string"},
|
|
"manufacturer": {"type": "string"},
|
|
"quantity": {"type": "integer", "default": 1}
|
|
}
|
|
}
|
|
}
|
|
},
|
|
"required": ["parts"]
|
|
}
|
|
},
|
|
{
|
|
"name": "get_alternatives",
|
|
"description": "Find alternative parts with similar specs",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"mpn": {"type": "string", "description": "Original part number"},
|
|
"manufacturer": {"type": "string", "description": "Original manufacturer"},
|
|
"specs": {"type": "object", "description": "Required specifications"}
|
|
},
|
|
"required": ["mpn"]
|
|
}
|
|
}
|
|
]
|
|
}
|
|
}
|
|
|
|
elif method == "tools/call":
|
|
tool_name = params.get("name")
|
|
arguments = params.get("arguments", {})
|
|
|
|
if tool_name == "search_parts":
|
|
result = await self._search_parts(arguments)
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"result": result
|
|
}
|
|
|
|
elif tool_name == "get_part_pricing":
|
|
result = await self._get_part_pricing(arguments)
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"result": result
|
|
}
|
|
|
|
elif tool_name == "get_best_price":
|
|
result = await self._get_best_price(arguments)
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"result": result
|
|
}
|
|
|
|
elif tool_name == "get_alternatives":
|
|
result = await self._get_alternatives(arguments)
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"result": result
|
|
}
|
|
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"error": {"code": -32601, "message": "Method not found"}
|
|
}
|
|
|
|
async def _search_parts(self, args: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Search for parts using Nexar API"""
|
|
try:
|
|
query = args.get('query', '')
|
|
|
|
# If we have API key, try real API call
|
|
if self.api_token and REQUESTS_AVAILABLE:
|
|
try:
|
|
return await self._search_parts_api(args)
|
|
except Exception as e:
|
|
logger.warning("Nexar API call failed, falling back to demo mode: %s", e)
|
|
|
|
# Create demo responses based on search query
|
|
demo_parts = []
|
|
|
|
if 'resistor' in query.lower() or 'ohm' in query.lower():
|
|
# Parse resistor value
|
|
value = "10K"
|
|
if "220" in query:
|
|
value = "220"
|
|
elif "1k" in query.lower() or "1000" in query:
|
|
value = "1K"
|
|
elif "2k" in query.lower():
|
|
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",
|
|
"category": "Resistors",
|
|
"distributors": {
|
|
"Digi-Key": {
|
|
"part_number": f"311-{value}CRCT-ND",
|
|
"unit_price": 0.02,
|
|
"stock": 50000,
|
|
"minimum_order": 1,
|
|
"url": "https://www.digikey.com"
|
|
},
|
|
"Mouser": {
|
|
"part_number": f"603-RC0805FR-07{value}L",
|
|
"unit_price": 0.018,
|
|
"stock": 75000,
|
|
"minimum_order": 1,
|
|
"url": "https://www.mouser.com"
|
|
},
|
|
"Farnell": {
|
|
"part_number": "9239111",
|
|
"unit_price": 0.025,
|
|
"stock": 25000,
|
|
"minimum_order": 1,
|
|
"url": "https://www.farnell.com"
|
|
},
|
|
"Newark": {
|
|
"part_number": "52AC3050",
|
|
"unit_price": 0.022,
|
|
"stock": 30000,
|
|
"minimum_order": 10,
|
|
"url": "https://www.newark.com"
|
|
}
|
|
},
|
|
"best_price": {
|
|
"distributor": "Mouser",
|
|
"price": 0.018,
|
|
"stock": 75000,
|
|
"minimum_order": 1
|
|
},
|
|
"specifications": {
|
|
"resistance": value + " Ohms",
|
|
"tolerance": "±1%",
|
|
"power": "0.125W",
|
|
"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",
|
|
"category": "Capacitors",
|
|
"distributors": {
|
|
"Digi-Key": {
|
|
"part_number": "CL21B104KBCNNNC-ND",
|
|
"unit_price": 0.01,
|
|
"stock": 100000,
|
|
"minimum_order": 1
|
|
},
|
|
"Mouser": {
|
|
"part_number": "187-CL21B104KBCNNNC",
|
|
"unit_price": 0.009,
|
|
"stock": 85000,
|
|
"minimum_order": 1
|
|
},
|
|
"Arrow": {
|
|
"part_number": "CL21B104KBCNNNC",
|
|
"unit_price": 0.008,
|
|
"stock": 60000,
|
|
"minimum_order": 1
|
|
}
|
|
},
|
|
"best_price": {
|
|
"distributor": "Arrow",
|
|
"price": 0.008,
|
|
"stock": 60000,
|
|
"minimum_order": 1
|
|
},
|
|
"specifications": {
|
|
"capacitance": "100nF",
|
|
"voltage": "50V",
|
|
"tolerance": "±10%",
|
|
"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}",
|
|
"category": "Demo Components",
|
|
"distributors": {
|
|
"Digi-Key": {"unit_price": 0.05, "stock": 10000, "minimum_order": 1},
|
|
"Mouser": {"unit_price": 0.048, "stock": 8000, "minimum_order": 1}
|
|
},
|
|
"best_price": {
|
|
"distributor": "Mouser",
|
|
"price": 0.048,
|
|
"stock": 8000,
|
|
"minimum_order": 1
|
|
}
|
|
})
|
|
]
|
|
|
|
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)}"}
|
|
|
|
async def _search_parts_api(self, args: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Make actual API call to Nexar"""
|
|
query = args.get('query', '')
|
|
limit = args.get('limit', 10)
|
|
|
|
endpoint = f"{self.base_url}/parts/search"
|
|
|
|
headers = {
|
|
'Authorization': f'Token {self.api_token}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
payload = {
|
|
"q": query,
|
|
"limit": limit,
|
|
"include": ["specs", "category", "offers"]
|
|
}
|
|
|
|
try:
|
|
response = requests.post(endpoint, headers=headers, json=payload, timeout=15)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
parts = []
|
|
|
|
for result in data.get('results', []):
|
|
part = result.get('item', {})
|
|
|
|
# Process distributors and pricing
|
|
distributors = {}
|
|
best_price = None
|
|
|
|
for offer in part.get('offers', []):
|
|
seller = offer.get('seller', {})
|
|
distributor_name = seller.get('name', 'Unknown')
|
|
|
|
if offer.get('prices'):
|
|
price_break = offer['prices'][0] # First price break
|
|
price = float(price_break[1]) # Price in USD
|
|
|
|
distributors[distributor_name] = {
|
|
"part_number": offer.get('sku', ''),
|
|
"unit_price": price,
|
|
"stock": offer.get('in_stock_quantity', 0),
|
|
"minimum_order": offer.get('moq', 1),
|
|
"url": offer.get('click_url', '')
|
|
}
|
|
|
|
if best_price is None or price < best_price['price']:
|
|
best_price = {
|
|
"distributor": distributor_name,
|
|
"price": price,
|
|
"stock": offer.get('in_stock_quantity', 0),
|
|
"minimum_order": offer.get('moq', 1)
|
|
}
|
|
|
|
part_info = {
|
|
"mpn": part.get('mpn', ''),
|
|
"manufacturer": part.get('manufacturer', {}).get('name', ''),
|
|
"description": part.get('short_description', ''),
|
|
"category": part.get('category', {}).get('name', ''),
|
|
"distributors": distributors,
|
|
"best_price": best_price,
|
|
"specifications": {spec.get('attribute', {}).get('name', ''): spec.get('value', '')
|
|
for spec in part.get('specs', [])}
|
|
}
|
|
parts.append(self._part_with_client_aliases(part_info))
|
|
|
|
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}")
|
|
|
|
except Exception as e:
|
|
raise Exception(f"Nexar API call failed: {str(e)}")
|
|
|
|
async def _get_part_pricing(self, args: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Get comprehensive pricing for a specific part"""
|
|
try:
|
|
mpn = args.get('mpn', '')
|
|
manufacturer = args.get('manufacturer', '')
|
|
quantity = args.get('quantity', 1)
|
|
|
|
# Demo pricing with quantity breaks
|
|
demo_pricing = {
|
|
"mpn": mpn,
|
|
"manufacturer": manufacturer,
|
|
"quantity_requested": quantity,
|
|
"distributors": {
|
|
"Digi-Key": {
|
|
"availability": 50000,
|
|
"pricing_tiers": [
|
|
{"quantity": 1, "unit_price": 0.025, "total": 0.025},
|
|
{"quantity": 10, "unit_price": 0.020, "total": 0.20},
|
|
{"quantity": 100, "unit_price": 0.015, "total": 1.50},
|
|
{"quantity": 1000, "unit_price": 0.010, "total": 10.00},
|
|
{"quantity": 5000, "unit_price": 0.008, "total": 40.00}
|
|
],
|
|
"recommended_price": 0.025 if quantity < 10 else 0.020 if quantity < 100 else 0.015,
|
|
"lead_time_weeks": 0
|
|
},
|
|
"Mouser": {
|
|
"availability": 75000,
|
|
"pricing_tiers": [
|
|
{"quantity": 1, "unit_price": 0.023, "total": 0.023},
|
|
{"quantity": 25, "unit_price": 0.018, "total": 0.45},
|
|
{"quantity": 100, "unit_price": 0.014, "total": 1.40},
|
|
{"quantity": 1000, "unit_price": 0.009, "total": 9.00}
|
|
],
|
|
"recommended_price": 0.023 if quantity < 25 else 0.018,
|
|
"lead_time_weeks": 0
|
|
},
|
|
"Arrow": {
|
|
"availability": 25000,
|
|
"pricing_tiers": [
|
|
{"quantity": 1, "unit_price": 0.021, "total": 0.021},
|
|
{"quantity": 50, "unit_price": 0.016, "total": 0.80},
|
|
{"quantity": 500, "unit_price": 0.012, "total": 6.00}
|
|
],
|
|
"recommended_price": 0.021 if quantity < 50 else 0.016,
|
|
"lead_time_weeks": 2
|
|
}
|
|
},
|
|
"best_option": {
|
|
"distributor": "Arrow",
|
|
"unit_price": 0.021,
|
|
"total_price": 0.021 * quantity,
|
|
"availability": 25000,
|
|
"lead_time": "2 weeks"
|
|
},
|
|
"total_market_availability": 150000,
|
|
"demo_mode": True
|
|
}
|
|
|
|
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)}"}
|
|
|
|
async def _get_best_price(self, args: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Find best prices for multiple parts"""
|
|
try:
|
|
parts = args.get('parts', [])
|
|
|
|
best_prices = []
|
|
total_cost = 0
|
|
|
|
for part in parts:
|
|
mpn = part.get('mpn', '')
|
|
quantity = part.get('quantity', 1)
|
|
|
|
# Demo best price calculation
|
|
unit_price = 0.02 # Base price
|
|
if 'resistor' in mpn.lower() or any(x in mpn.lower() for x in ['ohm', 'res']):
|
|
unit_price = 0.015
|
|
elif 'capacitor' in mpn.lower() or 'cap' in mpn.lower():
|
|
unit_price = 0.008
|
|
|
|
# Quantity discounts
|
|
if quantity >= 1000:
|
|
unit_price *= 0.4
|
|
elif quantity >= 100:
|
|
unit_price *= 0.6
|
|
elif quantity >= 10:
|
|
unit_price *= 0.8
|
|
|
|
part_total = unit_price * quantity
|
|
total_cost += part_total
|
|
|
|
best_prices.append({
|
|
"mpn": mpn,
|
|
"quantity": quantity,
|
|
"best_distributor": "Mouser",
|
|
"unit_price": round(unit_price, 4),
|
|
"total_price": round(part_total, 2),
|
|
"availability": 50000,
|
|
"lead_time": "In Stock"
|
|
})
|
|
|
|
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)}"}
|
|
|
|
async def _get_alternatives(self, args: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Find alternative parts"""
|
|
try:
|
|
mpn = args.get('mpn', '')
|
|
|
|
alternatives = [
|
|
{
|
|
"mpn": "Alternative-001",
|
|
"manufacturer": "Alternative Mfg",
|
|
"description": f"Alternative to {mpn}",
|
|
"compatibility_score": 0.95,
|
|
"price_difference": -0.002,
|
|
"availability": 75000,
|
|
"advantages": ["Lower cost", "Better availability"],
|
|
"considerations": ["Different package marking"]
|
|
},
|
|
{
|
|
"mpn": "Alternative-002",
|
|
"manufacturer": "Another Mfg",
|
|
"description": f"Drop-in replacement for {mpn}",
|
|
"compatibility_score": 0.98,
|
|
"price_difference": 0.001,
|
|
"availability": 40000,
|
|
"advantages": ["Exact specifications", "Same footprint"],
|
|
"considerations": ["Slightly higher cost"]
|
|
}
|
|
]
|
|
|
|
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)}"}
|
|
|
|
|
|
async def main():
|
|
"""Main MCP server loop"""
|
|
server = NexarServer()
|
|
|
|
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:
|
|
error_response = {
|
|
"jsonrpc": "2.0",
|
|
"id": None,
|
|
"error": {"code": -32603, "message": f"Internal error: {str(e)}"}
|
|
}
|
|
print(json.dumps(error_response))
|
|
sys.stdout.flush()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|