#!/usr/bin/env python3 """ Conveyor Maven MCP Server - Monetized Version Model Context Protocol server for Claude to access conveyor calculators Requires API key from https://conveyormaven.com/api-keys """ import asyncio import json import httpx import os from typing import Any, Dict, List from mcp.server import Server from mcp.types import Tool, TextContent, Resource from mcp.server.stdio import stdio_server # Conveyor API configuration CONVEYOR_API_URL = "https://3l0evu9ree.execute-api.us-east-1.amazonaws.com/prod" # Get API key from environment variable API_KEY = os.getenv('CONVEYOR_API_KEY') if not API_KEY: print("\n" + "="*60) print("ERROR: CONVEYOR_API_KEY environment variable not set!") print("="*60) print("\nTo use the Conveyor Maven calculators, you need an API key.") print("\n1. Get your FREE API key at:") print(" https://conveyormaven.com/api-keys") print("\n2. Set the environment variable:") print(" export CONVEYOR_API_KEY='your-api-key-here'") print("\n3. Or add it to your Claude Desktop config:") print(' "env": {"CONVEYOR_API_KEY": "your-api-key-here"}') print("\nFree tier includes:") print(" • 100 calculations per month") print(" • All 14 calculators") print(" • Community support") print("\nUpgrade for unlimited access!") print("="*60 + "\n") raise Exception("CONVEYOR_API_KEY not configured. Get your key at https://conveyormaven.com/api-keys") app = Server("conveyor-maven") # Tier-based calculator access TIER_CALCULATORS = { 'free': [ 'belt-speed', 'tonnage-calculator', 'power', 'belt-safety-factor', 'conveyor-lift' ], 'professional': 'all', # All calculators 'enterprise': 'all' # All calculators + future custom ones } # Calculator definitions with their schemas ALL_CALCULATORS = { "belt_feeder": { "description": "Calculate power requirements and tonnage for belt feeders. Returns shear resistance, skirt resistance, power consumption (both running and starting), and material tonnage.", "calculator_type": "belt-feeder", "tier_required": "professional", "input_schema": { "type": "object", "required": ["beltWidth", "archAngle", "matFricAngle", "skirtFricAngle", "slotWidthIn", "slotWidthOut", "slotLength", "skirtLengthOutHopper", "matDepthInSkirt", "beltSpeed", "bulkDensity", "hopperHeight"], "properties": { "beltWidth": {"type": "number", "description": "Belt width in millimeters"}, "archAngle": {"type": "number", "description": "Arch angle in radians (0 to π/2, default 1.396)"}, "matFricAngle": {"type": "number", "description": "Material friction angle in radians (default 0.698)"}, "skirtFricAngle": {"type": "number", "description": "Skirt friction angle in radians (default 0.524)"}, "slotWidthIn": {"type": "number", "description": "Slot width at inlet in meters"}, "slotWidthOut": {"type": "number", "description": "Slot width at outlet in meters"}, "slotLength": {"type": "number", "description": "Slot length in meters"}, "skirtLengthOutHopper": {"type": "number", "description": "Skirt length outside hopper in meters"}, "matDepthInSkirt": {"type": "number", "description": "Material depth in skirt in meters"}, "beltSpeed": {"type": "number", "description": "Belt speed in meters per second (default 0.6)"}, "bulkDensity": {"type": "number", "description": "Bulk density in kg/m³ (default 2002.0)"}, "hopperHeight": {"type": "number", "description": "Hopper height in meters"} } } }, "belt_speed": { "description": "Calculate belt speed from motor parameters including motor speed, pulley diameter, and gear ratio.", "calculator_type": "belt-speed", "tier_required": "free", "input_schema": { "type": "object", "required": ["motorSpeed", "motorPulleyDiameter", "gearRatio"], "properties": { "motorSpeed": {"type": "number", "description": "Motor speed in RPM"}, "motorPulleyDiameter": {"type": "number", "description": "Motor pulley diameter in mm"}, "gearRatio": {"type": "number", "description": "Gear reduction ratio"} } } }, "tonnage_calculator": { "description": "Calculate material flow rate (tonnage per hour) based on belt speed, width, material density, and surcharge angle.", "calculator_type": "tonnage-calculator", "tier_required": "free", "input_schema": { "type": "object", "required": ["beltSpeed", "beltWidth", "materialDensity", "surchargeAngle"], "properties": { "beltSpeed": {"type": "number", "description": "Belt speed in meters per minute"}, "beltWidth": {"type": "number", "description": "Belt width in mm"}, "materialDensity": {"type": "number", "description": "Material density in tonnes per cubic meter"}, "surchargeAngle": {"type": "number", "description": "Surcharge angle in degrees (0-90)"} } } }, "power": { "description": "Calculate minimum power required for a conveyor system based on tonnage, lift, length, and belt speed.", "calculator_type": "power", "tier_required": "free", "input_schema": { "type": "object", "required": ["tonnagePerHour", "lift", "length", "beltSpeed"], "properties": { "tonnagePerHour": {"type": "number", "description": "Material tonnage in tonnes per hour"}, "lift": {"type": "number", "description": "Vertical lift in meters"}, "length": {"type": "number", "description": "Conveyor length in meters"}, "beltSpeed": {"type": "number", "description": "Belt speed in meters per second"} } } }, "tension_estimate": { "description": "Calculate belt tensions using the Euler-Eytelwein equation. Determines tight side tension from slack side tension, wrap angle, and friction coefficient.", "calculator_type": "tension-estimate", "tier_required": "professional", "input_schema": { "type": "object", "required": ["slackSideTension", "wrapAngle", "frictionCoefficient"], "properties": { "slackSideTension": {"type": "number", "description": "Slack side tension in kN"}, "wrapAngle": {"type": "number", "description": "Wrap angle in degrees (0-360)"}, "frictionCoefficient": {"type": "number", "description": "Coefficient of friction (0-1, typical 0.35)"} } } }, "belt_safety_factor": { "description": "Calculate belt safety factor from belt rating and maximum tension to ensure safe operation.", "calculator_type": "belt-safety-factor", "tier_required": "free", "input_schema": { "type": "object", "required": ["beltRating", "maximumTension"], "properties": { "beltRating": {"type": "number", "description": "Belt rating in kN/m"}, "maximumTension": {"type": "number", "description": "Maximum belt tension in kN"} } } }, "belt_revolution_time": { "description": "Calculate the time required for one complete belt revolution based on belt length and speed.", "calculator_type": "belt-revolution-time", "tier_required": "professional", "input_schema": { "type": "object", "required": ["beltLength", "beltSpeed"], "properties": { "beltLength": {"type": "number", "description": "Total belt length in meters"}, "beltSpeed": {"type": "number", "description": "Belt speed in meters per second"} } } }, "loop_spacing": { "description": "Calculate optimal loop spacing for conveyor belts based on belt width and trough angle.", "calculator_type": "loop-spacing", "tier_required": "professional", "input_schema": { "type": "object", "required": ["beltWidth", "troughAngle"], "properties": { "beltWidth": {"type": "number", "description": "Belt width in mm"}, "troughAngle": {"type": "number", "description": "Trough angle in degrees (0-45)"} } } }, "conveyor_lift": { "description": "Calculate vertical lift from conveyor length and incline angle.", "calculator_type": "conveyor-lift", "tier_required": "free", "input_schema": { "type": "object", "required": ["conveyorLength", "inclineAngle"], "properties": { "conveyorLength": {"type": "number", "description": "Conveyor length in meters"}, "inclineAngle": {"type": "number", "description": "Incline angle in degrees (-90 to 90)"} } } }, "plugged_chute": { "description": "Calculate the force required to pull out material from a plugged chute.", "calculator_type": "plugged-chute", "tier_required": "professional", "input_schema": { "type": "object", "required": ["chuteArea", "chuteLength", "materialDensity", "frictionCoefficient"], "properties": { "chuteArea": {"type": "number", "description": "Chute cross-sectional area in m²"}, "chuteLength": {"type": "number", "description": "Chute length in meters"}, "materialDensity": {"type": "number", "description": "Material bulk density in kg/m³"}, "frictionCoefficient": {"type": "number", "description": "Material friction coefficient (0-1)"} } } }, "hub_outer_diameter": { "description": "Calculate required hub outer diameter for pulleys based on shaft diameter, torque, and material yield strength.", "calculator_type": "hub-outer-diameter", "tier_required": "professional", "input_schema": { "type": "object", "required": ["shaftDiameter", "torque", "yieldStrength"], "properties": { "shaftDiameter": {"type": "number", "description": "Shaft diameter in mm"}, "torque": {"type": "number", "description": "Applied torque in kN⋅m"}, "yieldStrength": {"type": "number", "description": "Material yield strength in MPa"} } } }, "stockpile_volume": { "description": "Calculate stockpile volume and tonnage from base dimensions, height, and angle of repose.", "calculator_type": "stockpile-volume", "tier_required": "professional", "input_schema": { "type": "object", "required": ["baseLength", "baseWidth", "height", "reposeAngle"], "properties": { "baseLength": {"type": "number", "description": "Stockpile base length in meters"}, "baseWidth": {"type": "number", "description": "Stockpile base width in meters"}, "height": {"type": "number", "description": "Stockpile height in meters"}, "reposeAngle": {"type": "number", "description": "Angle of repose in degrees (0-90)"} } } }, "simple_winder": { "description": "Calculate winding torque and safety factors for belt winder systems.", "calculator_type": "simple-winder", "tier_required": "professional", "input_schema": { "type": "object", "required": ["beltTension", "drumDiameter", "safetyFactor"], "properties": { "beltTension": {"type": "number", "description": "Belt tension in kN"}, "drumDiameter": {"type": "number", "description": "Drum diameter in mm"}, "safetyFactor": {"type": "number", "description": "Desired safety factor (minimum 1)"} } } }, "flywheel": { "description": "Calculate flywheel stress and performance including kinetic energy and centrifugal stress.", "calculator_type": "flywheel", "tier_required": "professional", "input_schema": { "type": "object", "required": ["mass", "radius", "angularVelocity"], "properties": { "mass": {"type": "number", "description": "Flywheel mass in kg"}, "radius": {"type": "number", "description": "Flywheel radius in meters"}, "angularVelocity": {"type": "number", "description": "Angular velocity in rad/s"} } } } } async def get_user_tier() -> str: """Check user's tier from API""" try: async with httpx.AsyncClient() as client: # Check with API Gateway authorizer what tier this key has # For now, we'll infer from usage patterns or default to free # In production, you'd call: GET /v1/user/tier # Demo: Use demo key as free, others check from DynamoDB if API_KEY == "demo-key-12345-test": return "free" # TODO: Implement actual tier checking from your DynamoDB # For now, default to free for safety return "free" except Exception as e: print(f"Warning: Could not verify tier, defaulting to free: {e}") return "free" def get_available_calculators(tier: str) -> List[str]: """Get list of calculators available for this tier""" if tier == 'free': return TIER_CALCULATORS['free'] else: # professional or enterprise return list(ALL_CALCULATORS.keys()) @app.list_tools() async def list_tools() -> List[Tool]: """List all available conveyor calculator tools based on user tier.""" tier = await get_user_tier() available_calc_names = get_available_calculators(tier) tools = [] locked_count = 0 for calc_name, calc_info in ALL_CALCULATORS.items(): # Check if user has access to this calculator if tier == 'free' and calc_info.get('tier_required') != 'free': locked_count += 1 continue tools.append(Tool( name=f"conveyor_{calc_name}", description=calc_info["description"] + (f" [FREE tier]" if calc_info.get('tier_required') == 'free' else f" [Requires {calc_info.get('tier_required', 'professional')} tier]"), inputSchema=calc_info["input_schema"] )) # Add Ask Maven knowledge base tool tools.append(Tool( name="ask_maven", description="Query the Conveyor Maven AI knowledge base for expert answers about conveyor engineering. FREE: 20 questions/month | PROFESSIONAL: 100 questions/month", inputSchema={ "type": "object", "required": ["question"], "properties": { "question": { "type": "string", "description": "Your conveyor engineering question" } } } )) # Add info message about locked calculators if locked_count > 0: print(f"\n💡 {locked_count} additional calculators available with Professional tier ($29/month)") print(f" Upgrade at: https://conveyormaven.com/pricing\n") return tools @app.call_tool() async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]: """Execute a conveyor calculator tool or query knowledge base.""" # Handle Ask Maven knowledge base queries if name == "ask_maven": question = arguments.get("question", "").strip() if not question: return [TextContent( type="text", text="Please provide a question." )] try: async with httpx.AsyncClient() as client: response = await client.post( f"{CONVEYOR_API_URL}/v1/ask", headers={ 'X-API-Key': API_KEY, 'Content-Type': 'application/json' }, json={"question": question}, timeout=60.0 ) # Handle rate limits if response.status_code == 429: tier = await get_user_tier() limit = "10 questions/month" if tier == "free" else "100 questions/month" return [TextContent( type="text", text=f""" ⚠️ Knowledge base query limit exceeded! Your {tier} tier includes {limit}. Upgrade options: • Professional: $29/month - 100 questions/month + 1,000 calculations • Enterprise: $299/month - Unlimited access Upgrade at: https://conveyormaven.com/pricing """ )] if response.status_code != 200: error_data = response.json() if response.content else {} error_msg = error_data.get('error', f'HTTP {response.status_code}') return [TextContent( type="text", text=f"Knowledge base error: {error_msg}" )] result = response.json() answer = result.get('answer', 'No answer available') citations = result.get('citations', []) # Format response output = f"## Maven's Answer\n\n{answer}\n\n" if citations: output += "### Sources\n" for i, citation in enumerate(citations[:3], 1): source = citation.get('source', 'Unknown') output += f"{i}. {source}\n" tier = result.get('tier', 'free') output += f"\n---\n*Powered by Conveyor Maven AI | Tier: {tier.title()}*" return [TextContent(type="text", text=output)] except httpx.TimeoutException: return [TextContent( type="text", text="⏱️ Knowledge base query timed out. Please try a simpler question." )] except Exception as e: return [TextContent( type="text", text=f"Error querying knowledge base: {str(e)}" )] # Handle calculator tools # Remove 'conveyor_' prefix to get calculator name calc_name = name.replace("conveyor_", "") if calc_name not in ALL_CALCULATORS: return [TextContent( type="text", text=f"Unknown calculator: {calc_name}" )] calc_info = ALL_CALCULATORS[calc_name] calc_type = calc_info["calculator_type"] # Check tier access tier = await get_user_tier() tier_required = calc_info.get("tier_required", "free") if tier == "free" and tier_required != "free": return [TextContent( type="text", text=f""" 🔒 This calculator requires a Professional tier subscription. **{calc_name.replace('_', ' ').title()}** is available in: • Professional: $29/month - All 14 calculators + 1000 calls/month • Enterprise: $299/month - Unlimited + Priority support Upgrade at: https://conveyormaven.com/pricing Your current tier: Free (5 calculators, 100 calls/month) """ )] try: # Call the Conveyor API with user's API key async with httpx.AsyncClient() as client: response = await client.post( f"{CONVEYOR_API_URL}/v1/calculate", headers={ 'X-API-Key': API_KEY, 'Content-Type': 'application/json' }, json={ "calculatorType": calc_type, "input": arguments }, timeout=30.0 ) # Check for rate limit or auth errors if response.status_code == 429: return [TextContent( type="text", text=""" ⚠️ Rate limit exceeded! Your free tier includes 100 calculations per month. You've reached your limit. Upgrade options: • Professional: $29/month - 1,000 calls/month • Enterprise: $299/month - Unlimited calls Upgrade at: https://conveyormaven.com/pricing """ )] if response.status_code == 401: return [TextContent( type="text", text=""" 🔑 Invalid API key! Your API key may have expired or been deactivated. Get a new key at: https://conveyormaven.com/api-keys """ )] result = response.json() if "error" in result: return [TextContent( type="text", text=f"Calculation error: {result['error']}" )] # Format the output nicely output_text = f"## {calc_type.replace('-', ' ').title()} - Results\n\n" for key, value in result.get("output", {}).items(): # Convert camelCase to Title Case with spaces formatted_key = ''.join([' ' + c if c.isupper() else c for c in key]).strip().title() if isinstance(value, float): output_text += f"**{formatted_key}**: {value:.2f}\n" else: output_text += f"**{formatted_key}**: {value}\n" # Add footer with tier info output_text += f"\n---\n*Calculated using Conveyor Maven API | Tier: {tier.title()}*\n" if tier == "free": output_text += "*Upgrade for unlimited access: https://conveyormaven.com/pricing*" return [TextContent( type="text", text=output_text )] except httpx.TimeoutException: return [TextContent( type="text", text="⏱️ Calculation timed out. Please try again." )] except Exception as e: return [TextContent( type="text", text=f"Error calling calculator API: {str(e)}\n\nIf this persists, contact support@conveyormaven.com" )] @app.list_resources() async def list_resources() -> List[Resource]: """List available resources""" return [ Resource( uri="conveyor://pricing", name="Pricing Information", description="API pricing tiers and features", mimeType="text/plain" ), Resource( uri="conveyor://docs", name="Documentation", description="Conveyor calculator documentation", mimeType="text/plain" ) ] @app.read_resource() async def read_resource(uri: str) -> str: """Read resource content""" if uri == "conveyor://pricing": return """ # Conveyor Maven API Pricing ## Free Tier - 5 calculators (belt-speed, tonnage, power, safety-factor, lift) - 100 calculations per month - 20 knowledge base questions per month - Community support - Perfect for: Individual engineers, students ## Professional - $29/month - All 14 calculators - 1,000 calculations per month - 100 knowledge base questions per month - Email support - Commercial use allowed - Perfect for: Engineering consultants, small teams ## Enterprise - $299/month - Unlimited calculations - Unlimited knowledge base access - Priority support - SLA guarantees - Custom calculators available - White-label options - Perfect for: Companies, software integrations Upgrade at: https://conveyormaven.com/pricing """ elif uri == "conveyor://docs": return """ # Conveyor Maven Calculator Documentation Available calculators and their uses: 1. Belt Speed - Calculate belt speed from motor parameters 2. Tonnage Calculator - Material flow rate calculations 3. Power - Minimum power requirements 4. Belt Safety Factor - Safety factor calculations 5. Conveyor Lift - Vertical lift calculations 6. Belt Feeder - Power and tonnage for belt feeders (Pro) 7. Tension Estimate - Belt tension calculations (Pro) 8. Belt Revolution Time - Revolution time (Pro) 9. Loop Spacing - Optimal loop spacing (Pro) 10. Plugged Chute - Pullout force calculations (Pro) 11. Hub Outer Diameter - Pulley hub sizing (Pro) 12. Stockpile Volume - Volume and tonnage (Pro) 13. Simple Winder - Winding torque (Pro) 14. Flywheel - Stress and energy calculations (Pro) Full documentation: https://conveyormaven.com/api-developers.html """ else: raise ValueError(f"Unknown resource: {uri}") async def main(): """Run the MCP server.""" print(f"\n✓ Conveyor Maven MCP Server starting...") print(f"✓ API Key configured: {API_KEY[:10]}...{API_KEY[-4:]}") print(f"✓ Connecting to: {CONVEYOR_API_URL}") tier = await get_user_tier() print(f"✓ Your tier: {tier.upper()}") available = get_available_calculators(tier) print(f"✓ Available calculators: {len(available)}") if tier == "free": print(f"\n💡 Upgrade to Professional for all 14 calculators!") print(f" https://conveyormaven.com/pricing\n") async with stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, app.create_initialization_options() ) if __name__ == "__main__": asyncio.run(main())