electron-rare/KiC-AI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote changes, keeping our configuration system implementation

Browse Source
This commit is contained in:
Jochem
2025-08-01 21:23:15 +02:00
parent 3c06ec6a89 6e50db1446
commit b5ff50dffb
13 changed files with 2534 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1 -1
View File
@@ -22,7 +22,7 @@ Thumbs.db
# Release files
*.zip
!kic-ai-assistant-v*.zip
!kicad-ai-assistant*.zip
# IDE
.vscode/
CHANGELOG.md
+36 -1
View File
@@ -1,4 +1,39 @@
# Chang## [1.2.1] - 2025-07-30
# Changelog
## [1.5.0-MCP] - 2025-08-01
### 🚀 Major New Features - MCP Integration
- **Model Context Protocol (MCP) Support**: Revolutionary integration enabling AI to access external tools and databases
- **Component Database Integration**: Real-time component pricing, availability, and specifications
- **Enhanced AI Analysis**: PCB analysis now includes cost estimation and stock availability
- **Smart Component Suggestions**: AI can suggest alternatives based on pricing, availability, and specifications
- **MCP Server Framework**: Extensible architecture for adding new data sources and tools
### 🔧 Technical Improvements
- **MCP Client**: Full MCP protocol implementation for external tool integration
- **Async Processing**: Non-blocking component database queries
- **Safety Controls**: User confirmation required for all automated suggestions
- **Error Handling**: Graceful fallback when MCP services are unavailable
- **Configuration System**: Easy setup and management of MCP servers
### 📊 Enhanced Capabilities
- **Cost Analysis**: "Your PCB costs €12.34 with 45 components"
- **Availability Checking**: Real-time stock level verification
- **Alternative Components**: Smart suggestions for better/cheaper parts
- **Supply Chain Intelligence**: Integration ready for Digi-Key, Mouser APIs
### 🛡️ Safety & Reliability
- **Safety Mode**: All MCP modifications require user approval
- **Backward Compatibility**: Plugin works normally without MCP
- **Graceful Degradation**: Automatic fallback to basic mode if MCP fails
### 📁 New Files
- `plugins/mcp_client.py` - MCP protocol client implementation
- `plugins/mcp_config.json` - MCP server configuration
- `mcp_servers/component_db.py` - Sample component database server
- `MCP_SETUP.md` - Installation and configuration guide
## [1.4.5] - 2025-07-31
### Documentation & UX Improvements
- **Clarified Usage Instructions**: Updated README to clearly explain plugin is accessed via PCB Editor toolbar
NEXAR_SETUP.md
+139
View File
@@ -0,0 +1,139 @@
# Nexar API Setup Guide for KiCad AI Assistant
## Quick Start - Free Tier Available!
Nexar API (formerly Octopart) provides comprehensive component pricing from multiple distributors in a single API call. **Much simpler than managing individual distributor APIs!**
### Free Tier: Welcome 1K Plan
- **1,000 API calls per month** - completely free
- Access to all major distributors (Digi-Key, Mouser, Farnell, Newark, Arrow, etc.)
- Multi-distributor pricing comparison in single request
- No credit card required for free tier
## Setup Steps
### 1. Create Nexar Account
1. Visit: https://portal.nexar.com
2. Sign up with your email
3. Verify your email (check spam folder if needed)
4. Log into the Nexar portal
### 2. Create Application
1. In Nexar portal, click "Create Application"
2. Choose application type: "Supply Chain"
3. Give it a name like "KiCad AI Assistant"
4. Description: "Component pricing for PCB design"
### 3. Get API Token
1. After creating the app, you'll get an API token
2. Copy this token - you'll need it for the environment variable
### 4. Configure KiCad AI Assistant
Set the environment variable:
```bash
export NEXAR_TOKEN="your_token_here"
```
Or add to your shell profile (`~/.zshrc` for zsh):
```bash
echo 'export NEXAR_TOKEN="your_token_here"' >> ~/.zshrc
source ~/.zshrc
```
## Testing the Setup
Run the test script to verify everything works:
```bash
cd /Users/jochem/Project/KIC-AI
python test_octopart_setup.py
```
## What You Get
### Multi-Distributor Pricing
One API call returns pricing from:
- Digi-Key
- Mouser Electronics
- Farnell/element14
- Newark
- Arrow Electronics
- Future Electronics
- And many more...
### Example Response
For a 10kΩ resistor query, you'll get:
```json
{
"parts": [
{
"mpn": "CF14JT10K0",
"manufacturer": "Stackpole Electronics Inc",
"pricing": {
"digikey": {"price": 0.10, "stock": 50000},
"mouser": {"price": 0.09, "stock": 25000},
"farnell": {"price": 0.11, "stock": 15000}
},
"best_price": {"distributor": "mouser", "price": 0.09}
}
]
}
```
### KiCad Integration
The AI assistant can now:
- Find real component pricing for your BOM
- Compare prices across all distributors
- Suggest cost optimizations
- Find alternative parts with better pricing
- Calculate total BOM costs with real data
## No More Complex API Setup!
Unlike Digi-Key V4 API which requires:
- ❌ Client ID + Client Secret
- ❌ OAuth flow management
- ❌ Sandbox vs Production URLs
- ❌ Complex authentication
Nexar API only needs:
- ✅ Single API token
- ✅ Simple GraphQL queries
- ✅ All distributors in one call
- ✅ Free tier with 1K calls/month
## API Limits & Costs
### Free Tier (Welcome 1K)
- 1,000 calls/month
- All distributor access
- Real-time pricing
- No credit card required
### Paid Tiers (if you need more)
- Contact Nexar for enterprise pricing
- Higher call limits available
## Demo Mode
Even without an API token, the MCP server works in demo mode with realistic pricing data, so you can test the integration immediately!
## Troubleshooting
### No Email Confirmation?
- Check spam/junk folder
- Try different email address
- Contact Nexar support if needed
### Token Not Working?
- Verify token is set correctly: `echo $NEXAR_TOKEN`
- Check token hasn't expired in portal
- Ensure no extra spaces in environment variable
### Rate Limiting?
- Free tier: 1,000 calls/month
- Check usage in Nexar portal
- Consider caching results for development
## Support
- Nexar Documentation: https://docs.nexar.com
- Support: support@nexar.com
- KiCad AI Assistant: Check GitHub issues
kicad-ai-assistant.zip
BIN
View File
Binary file not shown.
mcp_servers/__init__.py
+1
View File
@@ -0,0 +1 @@
# MCP Servers for KIC-AI Plugin
mcp_servers/component_db.py
+205
View File
@@ -0,0 +1,205 @@
#!/usr/bin/env python3
"""
Simple MCP Server for component database functionality
This demonstrates how to create an MCP server for KiCad integration
"""
import json
import sys
import asyncio
from typing import Dict, List, Any
class ComponentDatabaseServer:
"""Simple component database MCP server"""
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"]
}
}
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": {}
},
"serverInfo": {
"name": "component-database",
"version": "1.0.0"
}
}
}
elif method == "tools/list":
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"tools": [
{
"name": "search_components",
"description": "Search for components by type and specifications",
"inputSchema": {
"type": "object",
"properties": {
"type": {"type": "string"},
"specs": {"type": "object"}
}
}
},
{
"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",
"inputSchema": {
"type": "object",
"properties": {
"part_numbers": {"type": "array", "items": {"type": "string"}}
}
}
}
]
}
}
elif 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)
}
}
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}
}
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"]
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {"availability": availability}
}
# Default error response
return {
"jsonrpc": "2.0",
"id": request_id,
"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:
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())
mcp_servers/kicad_tools.py
+499
View File
@@ -0,0 +1,499 @@
#!/usr/bin/env python3
"""
KiCad Tools MCP Server
Provides tools for manipulating KiCad designs and schematics
"""
import json
import sys
import asyncio
import os
from typing import Dict, List, Any, Optional
class KiCadToolsServer:
"""MCP server for KiCad design manipulation tools"""
def __init__(self):
self.current_project_path = None
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": {}
},
"serverInfo": {
"name": "kicad-tools",
"version": "1.0.0"
}
}
}
elif method == "tools/list":
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"tools": [
{
"name": "analyze_schematic",
"description": "Analyze schematic for design rule violations and suggestions",
"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"
}
},
"required": ["schematic_path"]
}
},
{
"name": "analyze_pcb",
"description": "Analyze PCB layout for design issues",
"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"
}
},
"required": ["pcb_path"]
}
},
{
"name": "get_component_list",
"description": "Extract component list from schematic",
"inputSchema": {
"type": "object",
"properties": {
"schematic_path": {"type": "string"},
"include_values": {"type": "boolean", "default": true},
"include_footprints": {"type": "boolean", "default": true}
},
"required": ["schematic_path"]
}
},
{
"name": "generate_bom",
"description": "Generate Bill of Materials with pricing if available",
"inputSchema": {
"type": "object",
"properties": {
"schematic_path": {"type": "string"},
"include_pricing": {"type": "boolean", "default": false},
"group_by": {"type": "string", "enum": ["value", "footprint", "manufacturer"], "default": "value"}
},
"required": ["schematic_path"]
}
},
{
"name": "suggest_improvements",
"description": "Suggest design improvements based on analysis",
"inputSchema": {
"type": "object",
"properties": {
"project_path": {"type": "string"},
"focus_areas": {
"type": "array",
"items": {"type": "string"},
"description": "Areas to focus on: cost, performance, reliability, manufacturability"
}
},
"required": ["project_path"]
}
},
{
"name": "validate_footprints",
"description": "Check if all components have valid footprints assigned",
"inputSchema": {
"type": "object",
"properties": {
"schematic_path": {"type": "string"},
"suggest_alternatives": {"type": "boolean", "default": true}
},
"required": ["schematic_path"]
}
}
]
}
}
elif method == "tools/call":
tool_name = params.get("name")
arguments = params.get("arguments", {})
if tool_name == "analyze_schematic":
result = await self._analyze_schematic(arguments)
return {
"jsonrpc": "2.0",
"id": request_id,
"result": result
}
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
}
return {
"jsonrpc": "2.0",
"id": request_id,
"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:
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())
mcp_servers/nexar.py
+585
View File
@@ -0,0 +1,585 @@
#!/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
from typing import Dict, List, Any, Optional
try:
import requests
REQUESTS_AVAILABLE = True
except ImportError:
REQUESTS_AVAILABLE = False
class NexarServer:
"""MCP server for Nexar API integration (formerly Octopart)"""
def __init__(self):
self.api_token = os.getenv('NEXAR_TOKEN') # Nexar uses tokens instead of API keys
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 enhanced demo mode with realistic pricing", file=sys.stderr)
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": {}
},
"serverInfo": {
"name": "octopart-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 Octopart API"""
try:
query = args.get('query', '')
# If we have API key, try real API call
if self.api_key and REQUESTS_AVAILABLE:
try:
return await self._search_parts_api(args)
except Exception as e:
print(f"Octopart API call failed, using demo mode: {e}", file=sys.stderr)
# Demo mode - comprehensive pricing examples
print("Warning: No Octopart API key, using demo data", file=sys.stderr)
# 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 = [
{
"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": f"9239111",
"unit_price": 0.025,
"stock": 25000,
"minimum_order": 1,
"url": "https://www.farnell.com"
},
"Newark": {
"part_number": f"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 = [
{
"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 = [
{
"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 {
"parts": demo_parts,
"total_count": len(demo_parts),
"demo_mode": True,
"message": "Demo data - add OCTOPART_API_KEY for real pricing from all distributors",
"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 Octopart"""
query = args.get('query', '')
limit = args.get('limit', 10)
endpoint = f"{self.base_url}/parts/search"
headers = {
'Authorization': f'Token {self.api_key}',
'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(part_info)
return {
"parts": parts,
"total_count": len(parts),
"demo_mode": False,
"message": f"Found {len(parts)} parts via Octopart 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"Octopart 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 demo_pricing
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"
})
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"
}
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 {
"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())
plugins/config_dialog.py
+669
View File
@@ -0,0 +1,669 @@
# -*- coding: utf-8 -*-
"""
Configuration Dialog for KIC-AI Assistant
Allows users to configure MCP servers, API keys, and other settings
"""
import wx
import json
import os
from typing import Dict, List, Any
class MCPServerConfigPanel(wx.Panel):
"""Panel for configuring a single MCP server"""
def __init__(self, parent, server_name: str, server_config: Dict[str, Any]):
super().__init__(parent)
self.server_name = server_name
self.server_config = server_config.copy()
self.init_ui()
self.load_config()
def init_ui(self):
"""Initialize the UI for this server config"""
main_sizer = wx.BoxSizer(wx.VERTICAL)
# Server header
header_sizer = wx.BoxSizer(wx.HORIZONTAL)
# Enable checkbox
self.enabled_cb = wx.CheckBox(self, label=f"Enable {self.server_name}")
self.enabled_cb.SetFont(wx.Font(12, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_BOLD))
# Delete button
self.delete_btn = wx.Button(self, label="🗑️", size=(30, -1))
self.delete_btn.SetToolTip("Delete this server configuration")
header_sizer.Add(self.enabled_cb, 1, wx.EXPAND | wx.RIGHT, 5)
header_sizer.Add(self.delete_btn, 0, wx.ALIGN_CENTER_VERTICAL)
# Description
self.description_text = wx.StaticText(self, label="")
self.description_text.SetFont(wx.Font(9, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_ITALIC, wx.FONTWEIGHT_NORMAL))
# Configuration grid
config_sizer = wx.FlexGridSizer(0, 2, 5, 10)
config_sizer.AddGrowableCol(1, 1)
# Command
config_sizer.Add(wx.StaticText(self, label="Command:"), 0, wx.ALIGN_CENTER_VERTICAL)
self.command_text = wx.TextCtrl(self, style=wx.TE_READONLY)
config_sizer.Add(self.command_text, 1, wx.EXPAND)
# API Key (if required)
self.api_key_label = wx.StaticText(self, label="API Key:")
self.api_key_text = wx.TextCtrl(self, style=wx.TE_PASSWORD)
self.api_key_text.SetHint("Enter your API key here...")
config_sizer.Add(self.api_key_label, 0, wx.ALIGN_CENTER_VERTICAL)
config_sizer.Add(self.api_key_text, 1, wx.EXPAND)
# Client ID (for some APIs)
self.client_id_label = wx.StaticText(self, label="Client ID:")
self.client_id_text = wx.TextCtrl(self)
self.client_id_text.SetHint("Enter client ID if required...")
config_sizer.Add(self.client_id_label, 0, wx.ALIGN_CENTER_VERTICAL)
config_sizer.Add(self.client_id_text, 1, wx.EXPAND)
# API URL (for custom endpoints)
self.api_url_label = wx.StaticText(self, label="API URL:")
self.api_url_text = wx.TextCtrl(self)
self.api_url_text.SetHint("Custom API endpoint URL...")
config_sizer.Add(self.api_url_label, 0, wx.ALIGN_CENTER_VERTICAL)
config_sizer.Add(self.api_url_text, 1, wx.EXPAND)
# Client Secret (for V4 APIs like Digi-Key)
self.client_secret_label = wx.StaticText(self, label="Client Secret:")
self.client_secret_text = wx.TextCtrl(self, style=wx.TE_PASSWORD)
self.client_secret_text.SetHint("Enter client secret if required...")
config_sizer.Add(self.client_secret_label, 0, wx.ALIGN_CENTER_VERTICAL)
config_sizer.Add(self.client_secret_text, 1, wx.EXPAND)
# Sandbox mode checkbox
self.sandbox_label = wx.StaticText(self, label="Sandbox Mode:")
self.sandbox_checkbox = wx.CheckBox(self, label="Use sandbox/testing endpoint")
config_sizer.Add(self.sandbox_label, 0, wx.ALIGN_CENTER_VERTICAL)
config_sizer.Add(self.sandbox_checkbox, 0, wx.ALIGN_CENTER_VERTICAL)
# Test button
test_sizer = wx.BoxSizer(wx.HORIZONTAL)
self.test_btn = wx.Button(self, label="🧪 Test Connection")
self.test_status = wx.StaticText(self, label="")
test_sizer.Add(self.test_btn, 0, wx.RIGHT, 10)
test_sizer.Add(self.test_status, 1, wx.ALIGN_CENTER_VERTICAL)
# Layout
main_sizer.Add(header_sizer, 0, wx.EXPAND | wx.ALL, 5)
main_sizer.Add(self.description_text, 0, wx.EXPAND | wx.LEFT | wx.RIGHT, 10)
main_sizer.Add(wx.StaticLine(self), 0, wx.EXPAND | wx.ALL, 5)
main_sizer.Add(config_sizer, 0, wx.EXPAND | wx.ALL, 10)
main_sizer.Add(test_sizer, 0, wx.EXPAND | wx.ALL, 10)
main_sizer.Add(wx.StaticLine(self), 0, wx.EXPAND | wx.ALL, 5)
self.SetSizer(main_sizer)
# Events
self.enabled_cb.Bind(wx.EVT_CHECKBOX, self.on_enabled_change)
self.test_btn.Bind(wx.EVT_BUTTON, self.on_test_connection)
self.delete_btn.Bind(wx.EVT_BUTTON, self.on_delete)
# Update visibility based on server type
self.update_field_visibility()
def load_config(self):
"""Load configuration values"""
self.enabled_cb.SetValue(self.server_config.get('enabled', False))
self.description_text.SetLabel(self.server_config.get('description', ''))
# Command
command = self.server_config.get('command', [])
if isinstance(command, list):
self.command_text.SetValue(' '.join(command))
else:
self.command_text.SetValue(str(command))
# API credentials (load from environment or config)
if self.server_name == 'digikey_api':
api_key = os.getenv('DIGIKEY_API_KEY', '')
client_id = os.getenv('DIGIKEY_CLIENT_ID', '')
client_secret = os.getenv('DIGIKEY_CLIENT_SECRET', '')
use_sandbox = os.getenv('DIGIKEY_USE_SANDBOX', 'true')
self.api_key_text.SetValue(api_key)
self.client_id_text.SetValue(client_id)
if hasattr(self, 'client_secret_text'):
self.client_secret_text.SetValue(client_secret)
if hasattr(self, 'sandbox_checkbox'):
self.sandbox_checkbox.SetValue(use_sandbox.lower() == 'true')
self.api_url_text.SetValue('https://sandbox-api.digikey.com/products/v4' if use_sandbox.lower() == 'true' else 'https://api.digikey.com/products/v4')
elif self.server_name == 'mouser_api':
api_key = os.getenv('MOUSER_API_KEY', '')
self.api_key_text.SetValue(api_key)
self.api_url_text.SetValue('https://api.mouser.com/api/v1')
self.update_field_visibility()
def update_field_visibility(self):
"""Update field visibility based on server type"""
requires_api_key = self.server_config.get('requires_api_key', False)
# Show/hide API fields based on server type
if self.server_name in ['component_database', 'kicad_tools']:
# Local servers don't need API keys
self.api_key_label.Hide()
self.api_key_text.Hide()
self.client_id_label.Hide()
self.client_id_text.Hide()
self.api_url_label.Hide()
self.api_url_text.Hide()
elif self.server_name == 'digikey_api':
# Digi-Key needs API key and client ID
self.api_key_label.Show()
self.api_key_text.Show()
self.client_id_label.Show()
self.client_id_text.Show()
self.api_url_label.Show()
self.api_url_text.Show()
else:
# Other APIs might need just API key
self.api_key_label.Show()
self.api_key_text.Show()
self.client_id_label.Hide()
self.client_id_text.Hide()
self.api_url_label.Show()
self.api_url_text.Show()
self.Layout()
def on_enabled_change(self, event):
"""Handle enable/disable change"""
enabled = self.enabled_cb.GetValue()
self.server_config['enabled'] = enabled
# Enable/disable all controls
for child in self.GetChildren():
if child != self.enabled_cb and child != self.delete_btn:
child.Enable(enabled)
def on_test_connection(self, event):
"""Test the server connection"""
self.test_status.SetLabel("🔄 Testing...")
wx.CallAfter(self._test_connection_async)
def _test_connection_async(self):
"""Test connection in background"""
try:
# Import test function
import sys
import subprocess
import tempfile
import time
# Create test message
test_message = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"clientInfo": {"name": "config-test", "version": "1.0.0"}
}
}
# Get command
command = self.server_config.get('command', [])
if isinstance(command, str):
command = command.split()
# Start server process
process = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
cwd=os.path.dirname(os.path.dirname(__file__))
)
# Send test message
process.stdin.write(json.dumps(test_message) + '\n')
process.stdin.flush()
# Wait for response (with timeout)
import select
ready, _, _ = select.select([process.stdout], [], [], 5.0) # 5 second timeout
if ready:
response_line = process.stdout.readline()
if response_line:
response = json.loads(response_line.strip())
if 'result' in response:
server_name = response.get('result', {}).get('serverInfo', {}).get('name', 'Unknown')
self.test_status.SetLabel(f"✅ Connected: {server_name}")
self.test_status.SetForegroundColour(wx.Colour(0, 128, 0))
else:
self.test_status.SetLabel("❌ Server error")
self.test_status.SetForegroundColour(wx.Colour(255, 0, 0))
else:
self.test_status.SetLabel("❌ No response")
self.test_status.SetForegroundColour(wx.Colour(255, 0, 0))
else:
self.test_status.SetLabel("❌ Timeout")
self.test_status.SetForegroundColour(wx.Colour(255, 0, 0))
process.terminate()
except Exception as e:
self.test_status.SetLabel(f"❌ Error: {str(e)}")
self.test_status.SetForegroundColour(wx.Colour(255, 0, 0))
def on_delete(self, event):
"""Handle delete server"""
dlg = wx.MessageDialog(
self,
f"Are you sure you want to delete the '{self.server_name}' server configuration?",
"Confirm Delete",
wx.YES_NO | wx.ICON_QUESTION
)
if dlg.ShowModal() == wx.ID_YES:
# Notify parent to remove this panel
event = wx.PyCommandEvent(wx.wxEVT_COMMAND_BUTTON_CLICKED, self.GetId())
event.SetEventObject(self)
event.server_name = self.server_name
wx.PostEvent(self.GetParent(), event)
dlg.Destroy()
def get_config(self) -> Dict[str, Any]:
"""Get the current configuration"""
config = self.server_config.copy()
config['enabled'] = self.enabled_cb.GetValue()
# Save API credentials to environment variables
if self.api_key_text.IsShown() and self.api_key_text.GetValue():
if self.server_name == 'digikey_api':
os.environ['DIGIKEY_API_KEY'] = self.api_key_text.GetValue()
os.environ['DIGIKEY_CLIENT_ID'] = self.client_id_text.GetValue()
if hasattr(self, 'client_secret_text') and self.client_secret_text.GetValue():
os.environ['DIGIKEY_CLIENT_SECRET'] = self.client_secret_text.GetValue()
if hasattr(self, 'sandbox_checkbox'):
os.environ['DIGIKEY_USE_SANDBOX'] = 'true' if self.sandbox_checkbox.GetValue() else 'false'
elif self.server_name == 'mouser_api':
os.environ['MOUSER_API_KEY'] = self.api_key_text.GetValue()
return config
class AddServerDialog(wx.Dialog):
"""Dialog for adding a new MCP server"""
def __init__(self, parent):
super().__init__(parent, title="Add MCP Server", size=(400, 300))
self.init_ui()
def init_ui(self):
"""Initialize the UI"""
panel = wx.Panel(self)
main_sizer = wx.BoxSizer(wx.VERTICAL)
# Instructions
instruction_text = wx.StaticText(panel, label="Add a new MCP server configuration:")
instruction_text.SetFont(wx.Font(12, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_BOLD))
# Server template selection
template_sizer = wx.StaticBoxSizer(wx.VERTICAL, panel, "Server Template")
self.template_choice = wx.Choice(panel, choices=[
"Custom Server",
"Digi-Key API Server",
"Mouser API Server",
"Local Component Database",
"External Tool Server"
])
self.template_choice.SetSelection(0)
template_desc = wx.StaticText(panel, label="Select a template or create a custom server")
template_desc.SetFont(wx.Font(9, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_ITALIC, wx.FONTWEIGHT_NORMAL))
template_sizer.Add(self.template_choice, 0, wx.EXPAND | wx.ALL, 5)
template_sizer.Add(template_desc, 0, wx.EXPAND | wx.ALL, 5)
# Server details
details_sizer = wx.StaticBoxSizer(wx.VERTICAL, panel, "Server Details")
form_sizer = wx.FlexGridSizer(0, 2, 5, 10)
form_sizer.AddGrowableCol(1, 1)
# Server name
form_sizer.Add(wx.StaticText(panel, label="Server Name:"), 0, wx.ALIGN_CENTER_VERTICAL)
self.name_text = wx.TextCtrl(panel)
self.name_text.SetHint("e.g., my_custom_server")
form_sizer.Add(self.name_text, 1, wx.EXPAND)
# Description
form_sizer.Add(wx.StaticText(panel, label="Description:"), 0, wx.ALIGN_CENTER_VERTICAL)
self.desc_text = wx.TextCtrl(panel)
self.desc_text.SetHint("Brief description of the server")
form_sizer.Add(self.desc_text, 1, wx.EXPAND)
# Command
form_sizer.Add(wx.StaticText(panel, label="Command:"), 0, wx.ALIGN_CENTER_VERTICAL)
self.command_text = wx.TextCtrl(panel)
self.command_text.SetHint("python3 -m my_server")
form_sizer.Add(self.command_text, 1, wx.EXPAND)
details_sizer.Add(form_sizer, 1, wx.EXPAND | wx.ALL, 5)
# Buttons
button_sizer = wx.BoxSizer(wx.HORIZONTAL)
button_sizer.AddStretchSpacer()
cancel_btn = wx.Button(panel, wx.ID_CANCEL, "Cancel")
add_btn = wx.Button(panel, wx.ID_OK, "Add Server")
add_btn.SetDefault()
button_sizer.Add(cancel_btn, 0, wx.RIGHT, 5)
button_sizer.Add(add_btn, 0)
# Layout
main_sizer.Add(instruction_text, 0, wx.EXPAND | wx.ALL, 10)
main_sizer.Add(template_sizer, 0, wx.EXPAND | wx.ALL, 10)
main_sizer.Add(details_sizer, 1, wx.EXPAND | wx.ALL, 10)
main_sizer.Add(button_sizer, 0, wx.EXPAND | wx.ALL, 10)
panel.SetSizer(main_sizer)
# Events
self.template_choice.Bind(wx.EVT_CHOICE, self.on_template_change)
add_btn.Bind(wx.EVT_BUTTON, self.on_add)
cancel_btn.Bind(wx.EVT_BUTTON, self.on_cancel)
self.Bind(wx.EVT_CLOSE, self.on_close)
# Load default template
self.on_template_change(None)
def on_template_change(self, event):
"""Handle template selection change"""
selection = self.template_choice.GetSelection()
if selection == 1: # Digi-Key API Server
self.name_text.SetValue("digikey_api_custom")
self.desc_text.SetValue("Custom Digi-Key API integration")
self.command_text.SetValue("python3 -m mcp_servers.digikey")
elif selection == 2: # Mouser API Server
self.name_text.SetValue("mouser_api")
self.desc_text.SetValue("Mouser Electronics API integration")
self.command_text.SetValue("python3 -m mcp_servers.mouser")
elif selection == 3: # Local Component Database
self.name_text.SetValue("local_database")
self.desc_text.SetValue("Local component database")
self.command_text.SetValue("python3 -m mcp_servers.local_db")
elif selection == 4: # External Tool Server
self.name_text.SetValue("external_tool")
self.desc_text.SetValue("External tool integration")
self.command_text.SetValue("python3 -m mcp_servers.external")
else: # Custom Server
self.name_text.SetValue("")
self.desc_text.SetValue("")
self.command_text.SetValue("")
def get_server_config(self) -> tuple:
"""Get the server configuration"""
name = self.name_text.GetValue().strip()
if not name:
return None, None
config = {
"command": self.command_text.GetValue().strip().split(),
"description": self.desc_text.GetValue().strip(),
"enabled": False, # Start disabled
"requires_api_key": "api" in name.lower()
}
return name, config
def on_add(self, event):
"""Handle add button click"""
name, config = self.get_server_config()
if name and config:
self.EndModal(wx.ID_OK)
else:
wx.MessageBox("Please provide a server name and command!", "Missing Information", wx.OK | wx.ICON_WARNING)
def on_cancel(self, event):
"""Handle cancel button click"""
self.EndModal(wx.ID_CANCEL)
def on_close(self, event):
"""Handle window close (X button)"""
self.EndModal(wx.ID_CANCEL)
class ConfigurationDialog(wx.Dialog):
"""Main configuration dialog for KIC-AI Assistant"""
def __init__(self, parent):
super().__init__(parent, title="KIC-AI Configuration", size=(700, 600))
self.config_file = os.path.join(os.path.dirname(__file__), 'mcp_config.json')
self.server_panels = {}
self.init_ui()
self.load_configuration()
def init_ui(self):
"""Initialize the UI"""
panel = wx.Panel(self)
main_sizer = wx.BoxSizer(wx.VERTICAL)
# Header
header_text = wx.StaticText(panel, label="KIC-AI Assistant Configuration")
header_text.SetFont(wx.Font(16, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_BOLD))
# Notebook for different config sections
self.notebook = wx.Notebook(panel)
# MCP Servers tab
self.mcp_panel = wx.ScrolledWindow(self.notebook)
self.mcp_panel.SetScrollRate(5, 5)
self.mcp_sizer = wx.BoxSizer(wx.VERTICAL)
# Add server button
add_server_btn = wx.Button(self.mcp_panel, label="➕ Add MCP Server")
add_server_btn.Bind(wx.EVT_BUTTON, self.on_add_server)
self.mcp_sizer.Add(add_server_btn, 0, wx.EXPAND | wx.ALL, 10)
self.mcp_panel.SetSizer(self.mcp_sizer)
self.notebook.AddPage(self.mcp_panel, "MCP Servers")
# AI Settings tab
ai_panel = wx.Panel(self.notebook)
ai_sizer = wx.BoxSizer(wx.VERTICAL)
# AI Enhancement settings
ai_group = wx.StaticBoxSizer(wx.VERTICAL, ai_panel, "AI Enhancement Settings")
self.enable_suggestions_cb = wx.CheckBox(ai_panel, label="Enable component suggestions")
self.enable_pricing_cb = wx.CheckBox(ai_panel, label="Enable pricing lookup")
self.enable_auto_mod_cb = wx.CheckBox(ai_panel, label="Enable automated modifications (Advanced)")
self.safety_mode_cb = wx.CheckBox(ai_panel, label="Safety mode (Recommended)")
ai_group.Add(self.enable_suggestions_cb, 0, wx.EXPAND | wx.ALL, 5)
ai_group.Add(self.enable_pricing_cb, 0, wx.EXPAND | wx.ALL, 5)
ai_group.Add(self.enable_auto_mod_cb, 0, wx.EXPAND | wx.ALL, 5)
ai_group.Add(self.safety_mode_cb, 0, wx.EXPAND | wx.ALL, 5)
ai_sizer.Add(ai_group, 0, wx.EXPAND | wx.ALL, 10)
ai_panel.SetSizer(ai_sizer)
self.notebook.AddPage(ai_panel, "AI Settings")
# Buttons
button_sizer = wx.BoxSizer(wx.HORIZONTAL)
button_sizer.AddStretchSpacer()
cancel_btn = wx.Button(panel, wx.ID_CANCEL, "Cancel")
save_btn = wx.Button(panel, wx.ID_OK, "Save Configuration")
save_btn.SetDefault()
button_sizer.Add(cancel_btn, 0, wx.RIGHT, 5)
button_sizer.Add(save_btn, 0)
# Layout
main_sizer.Add(header_text, 0, wx.EXPAND | wx.ALL, 10)
main_sizer.Add(self.notebook, 1, wx.EXPAND | wx.ALL, 10)
main_sizer.Add(button_sizer, 0, wx.EXPAND | wx.ALL, 10)
panel.SetSizer(main_sizer)
# Events
self.Bind(wx.EVT_BUTTON, self.on_server_delete, id=wx.ID_ANY)
save_btn.Bind(wx.EVT_BUTTON, self.on_save)
cancel_btn.Bind(wx.EVT_BUTTON, self.on_cancel)
self.Bind(wx.EVT_CLOSE, self.on_close)
def load_configuration(self):
"""Load existing configuration"""
try:
if os.path.exists(self.config_file):
with open(self.config_file, 'r') as f:
config = json.load(f)
else:
# Default configuration
config = {
"mcp_servers": {
"component_database": {
"command": ["python3", "-m", "mcp_servers.component_db"],
"description": "Component database with pricing and availability",
"enabled": True
},
"digikey_api": {
"command": ["python3", "-m", "mcp_servers.digikey"],
"description": "Digi-Key API integration for real-time data",
"enabled": False,
"requires_api_key": True
},
"kicad_tools": {
"command": ["python3", "-m", "mcp_servers.kicad_tools"],
"description": "KiCad design manipulation tools",
"enabled": True
}
},
"ai_enhancement": {
"enable_component_suggestions": True,
"enable_pricing_lookup": True,
"enable_automated_modifications": False,
"safety_mode": True
}
}
# Load MCP servers
for server_name, server_config in config.get('mcp_servers', {}).items():
self.add_server_panel(server_name, server_config)
# Load AI settings
ai_settings = config.get('ai_enhancement', {})
self.enable_suggestions_cb.SetValue(ai_settings.get('enable_component_suggestions', True))
self.enable_pricing_cb.SetValue(ai_settings.get('enable_pricing_lookup', True))
self.enable_auto_mod_cb.SetValue(ai_settings.get('enable_automated_modifications', False))
self.safety_mode_cb.SetValue(ai_settings.get('safety_mode', True))
except Exception as e:
wx.MessageBox(f"Error loading configuration: {e}", "Configuration Error", wx.OK | wx.ICON_ERROR)
def add_server_panel(self, server_name: str, server_config: Dict[str, Any]):
"""Add a server configuration panel"""
panel = MCPServerConfigPanel(self.mcp_panel, server_name, server_config)
self.server_panels[server_name] = panel
self.mcp_sizer.Insert(self.mcp_sizer.GetItemCount() - 1, panel, 0, wx.EXPAND | wx.ALL, 5)
self.mcp_panel.Layout()
self.mcp_panel.FitInside()
def on_add_server(self, event):
"""Handle add server button"""
dlg = AddServerDialog(self)
if dlg.ShowModal() == wx.ID_OK:
server_name, server_config = dlg.get_server_config()
if server_name and server_config:
if server_name not in self.server_panels:
self.add_server_panel(server_name, server_config)
else:
wx.MessageBox("Server name already exists!", "Duplicate Name", wx.OK | wx.ICON_WARNING)
dlg.Destroy()
def on_server_delete(self, event):
"""Handle server deletion"""
if hasattr(event, 'server_name'):
server_name = event.server_name
if server_name in self.server_panels:
panel = self.server_panels[server_name]
self.mcp_sizer.Detach(panel)
panel.Destroy()
del self.server_panels[server_name]
self.mcp_panel.Layout()
self.mcp_panel.FitInside()
def save_configuration(self):
"""Save the current configuration"""
try:
config = {
"mcp_servers": {},
"ai_enhancement": {
"enable_component_suggestions": self.enable_suggestions_cb.GetValue(),
"enable_pricing_lookup": self.enable_pricing_cb.GetValue(),
"enable_automated_modifications": self.enable_auto_mod_cb.GetValue(),
"safety_mode": self.safety_mode_cb.GetValue()
}
}
# Get server configurations
for server_name, panel in self.server_panels.items():
config["mcp_servers"][server_name] = panel.get_config()
# Save to file
with open(self.config_file, 'w') as f:
json.dump(config, f, indent=2)
return True
except Exception as e:
wx.MessageBox(f"Error saving configuration: {e}", "Save Error", wx.OK | wx.ICON_ERROR)
return False
def on_save(self, event):
"""Handle save button click"""
if self.save_configuration():
self.EndModal(wx.ID_OK)
# If save fails, dialog stays open
def on_cancel(self, event):
"""Handle cancel button click"""
self.EndModal(wx.ID_CANCEL)
def on_close(self, event):
"""Handle window close (X button)"""
self.EndModal(wx.ID_CANCEL)
def show_configuration_dialog(parent=None):
"""Show the configuration dialog"""
dlg = ConfigurationDialog(parent)
result = dlg.ShowModal()
if result == wx.ID_OK:
success = dlg.save_configuration()
if success:
wx.MessageBox("Configuration saved successfully!\nRestart the plugin to apply changes.",
"Configuration Saved", wx.OK | wx.ICON_INFORMATION)
dlg.Destroy()
return result == wx.ID_OK
if __name__ == "__main__":
app = wx.App()
show_configuration_dialog()
app.MainLoop()
plugins/mcp_client.py
+225
View File
@@ -0,0 +1,225 @@
# -*- coding: utf-8 -*-
"""
MCP (Model Context Protocol) Client for KIC-AI Plugin
Enables communication with MCP servers for extended functionality
"""
import json
import subprocess
import threading
import time
from typing import Dict, List, Any, Optional
class MCPClient:
"""MCP Client for communicating with MCP servers"""
def __init__(self):
self.servers = {}
self.available_tools = {}
def connect_server(self, server_name: str, server_config: Dict[str, Any]) -> bool:
"""Connect to an MCP server"""
try:
# Start MCP server process
process = subprocess.Popen(
server_config['command'],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=0
)
self.servers[server_name] = {
'process': process,
'config': server_config
}
# Initialize server
init_message = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
},
"clientInfo": {
"name": "KIC-AI",
"version": "1.4.5"
}
}
}
self._send_message(server_name, init_message)
response = self._receive_message(server_name)
if response and 'result' in response:
# Get available tools
self._load_server_tools(server_name)
return True
except Exception as e:
print(f"Failed to connect to MCP server {server_name}: {e}")
return False
def _send_message(self, server_name: str, message: Dict[str, Any]):
"""Send JSON-RPC message to MCP server"""
if server_name in self.servers:
process = self.servers[server_name]['process']
json_message = json.dumps(message) + '\n'
process.stdin.write(json_message)
process.stdin.flush()
def _receive_message(self, server_name: str) -> Optional[Dict[str, Any]]:
"""Receive JSON-RPC message from MCP server"""
if server_name in self.servers:
process = self.servers[server_name]['process']
try:
line = process.stdout.readline()
if line:
try:
response = json.loads(line.strip())
return response
except json.JSONDecodeError as e:
print(f"JSON decode error from {server_name}: {e}")
print(f"Raw response: {line}")
return None
else:
print(f"Empty response from {server_name}")
return None
except Exception as e:
print(f"Error reading from {server_name}: {e}")
return None
return None
def _load_server_tools(self, server_name: str):
"""Load available tools from MCP server"""
tools_message = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list"
}
self._send_message(server_name, tools_message)
response = self._receive_message(server_name)
if response and 'result' in response:
tools = response['result'].get('tools', [])
for tool in tools:
tool_name = tool['name']
self.available_tools[tool_name] = {
'server': server_name,
'schema': tool
}
def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Call a tool via MCP"""
if tool_name not in self.available_tools:
return {"error": f"Tool {tool_name} not available"}
server_name = self.available_tools[tool_name]['server']
tool_message = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
}
try:
self._send_message(server_name, tool_message)
response = self._receive_message(server_name)
if response:
return response
else:
return {"error": f"No response from server {server_name}"}
except Exception as e:
return {"error": f"Tool call failed: {str(e)}"}
def get_available_tools(self) -> List[str]:
"""Get list of available tools"""
return list(self.available_tools.keys())
def disconnect_all(self):
"""Disconnect all MCP servers"""
for server_name, server_info in self.servers.items():
try:
server_info['process'].terminate()
except:
pass
self.servers.clear()
self.available_tools.clear()
class KiCadMCPTools:
"""KiCad-specific MCP tool implementations"""
def __init__(self, mcp_client: MCPClient):
self.mcp_client = mcp_client
def search_components(self, component_type: str, specifications: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Search for components using MCP component database"""
try:
result = self.mcp_client.call_tool("search_components", {
"type": component_type,
"specs": specifications
})
if result and 'result' in result:
return result['result'].get('components', [])
except Exception as e:
print(f"Component search error: {e}")
return []
def get_component_pricing(self, part_numbers: List[str]) -> Dict[str, Dict[str, Any]]:
"""Get component pricing via MCP"""
try:
result = self.mcp_client.call_tool("get_pricing", {
"part_numbers": part_numbers
})
if result and 'result' in result:
return result['result'].get('pricing', {})
except Exception as e:
print(f"Pricing lookup error: {e}")
return {}
def check_component_availability(self, part_numbers: List[str]) -> Dict[str, int]:
"""Check component availability via MCP"""
try:
result = self.mcp_client.call_tool("check_availability", {
"part_numbers": part_numbers
})
if result and 'result' in result:
return result['result'].get('availability', {})
except Exception as e:
print(f"Availability check error: {e}")
return {}
def suggest_alternatives(self, component_specs: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Get component alternatives via MCP"""
try:
result = self.mcp_client.call_tool("suggest_alternatives", component_specs)
if result and 'result' in result:
return result['result'].get('alternatives', [])
except Exception as e:
print(f"Alternative search error: {e}")
return []
plugins/mcp_config.json
+26
View File
@@ -0,0 +1,26 @@
{
"mcp_servers": {
"component_database": {
"command": ["python3", "-m", "mcp_servers.component_db"],
"description": "Component database with pricing and availability",
"enabled": true
},
"nexar_api": {
"command": ["python3", "-m", "mcp_servers.nexar"],
"description": "Nexar API - Multi-distributor pricing in one call (FREE tier: 1K calls/month)",
"enabled": true,
"requires_api_key": false
},
"kicad_tools": {
"command": ["python3", "-m", "mcp_servers.kicad_tools"],
"description": "KiCad design manipulation tools",
"enabled": true
}
},
"ai_enhancement": {
"enable_component_suggestions": true,
"enable_pricing_lookup": true,
"enable_automated_modifications": false,
"safety_mode": true
}
}
test_component_db_interactive.py
+148
View File
@@ -0,0 +1,148 @@
#!/usr/bin/env python3
"""
Interactive test for component_db MCP server
Shows you exactly how to communicate with the server
"""
import subprocess
import json
import sys
import time
def test_component_db_interactive():
"""Test component_db server interactively"""
print("🧪 Interactive Component DB Test")
print("=" * 40)
# Start the server process
print("Starting component_db server...")
try:
process = subprocess.Popen(
["python3", "-m", "mcp_servers.component_db"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1 # Line buffered
)
def send_request(request_dict):
"""Send a JSON-RPC request and get response"""
request_json = json.dumps(request_dict) + "\n"
print(f"\n📤 Sending: {request_json.strip()}")
process.stdin.write(request_json)
process.stdin.flush()
# Read response
response = process.stdout.readline()
if response:
print(f"📥 Response: {response.strip()}")
return json.loads(response.strip())
else:
print("❌ No response received")
return None
# Test 1: Initialize
print("\n1️⃣ Testing initialize...")
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {}
}
response = send_request(init_request)
if response and "result" in response:
print("✅ Initialize successful!")
server_info = response["result"]["serverInfo"]
print(f" Server: {server_info['name']} v{server_info['version']}")
else:
print("❌ Initialize failed!")
return
# Test 2: List tools
print("\n2️⃣ Testing tools/list...")
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
response = send_request(tools_request)
if response and "result" in response:
tools = response["result"]["tools"]
print(f"✅ Found {len(tools)} tools:")
for tool in tools:
print(f" - {tool['name']}: {tool['description']}")
else:
print("❌ Tools/list failed!")
return
# Test 3: Call a tool
print("\n3️⃣ Testing tools/call - search_components...")
call_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "search_components",
"arguments": {
"type": "resistor",
"specs": {"value": "10k"}
}
}
}
response = send_request(call_request)
if response and "result" in response:
print("✅ Tool call successful!")
print(f" Result: {json.dumps(response['result'], indent=2)}")
else:
print("❌ Tool call failed!")
if response and "error" in response:
print(f" Error: {response['error']}")
# Test 4: Get pricing
print("\n4️⃣ Testing tools/call - get_pricing...")
pricing_request = {
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "get_pricing",
"arguments": {
"part_numbers": ["R1", "C1"]
}
}
}
response = send_request(pricing_request)
if response and "result" in response:
print("✅ Pricing request successful!")
print(f" Result: {json.dumps(response['result'], indent=2)}")
else:
print("❌ Pricing request failed!")
if response and "error" in response:
print(f" Error: {response['error']}")
print("\n🎉 All tests completed!")
except Exception as e:
print(f"❌ Error running test: {e}")
finally:
# Clean up
if 'process' in locals():
process.terminate()
process.wait(timeout=5)
print("\n💡 Manual test instructions:")
print("To test manually, run:")
print("python3 -m mcp_servers.component_db")
print("\nThen send JSON-RPC requests like:")
print('{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}')
print("\n⚠️ Important: Each request must end with a newline!")
if __name__ == "__main__":
test_component_db_interactive()
test_nexar_setup.py
View File