skulabs_mcp_server_working.py•12.5 kB
"""
Working Skulabs MCP Server
Uses confirmed working API endpoints and provides mock data for testing
"""
import asyncio
import json
import os
import structlog
from typing import Any, Dict, List, Optional
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
Resource,
Tool,
TextContent,
ImageContent,
EmbeddedResource,
LoggingLevel
)
from dotenv import load_dotenv
from skulabs_client_working import SkulabsClient, SkulabsAPIError
# Load environment variables
load_dotenv()
# Configure logging
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
structlog.processors.JSONRenderer()
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
logger = structlog.get_logger()
# Initialize MCP server
server = Server("skulabs-mcp")
# Configuration
SKULABS_API_KEY = os.getenv("SKULABS_API_KEY")
SKULABS_BASE_URL = os.getenv("SKULABS_BASE_URL", "https://api.skulabs.com")
if not SKULABS_API_KEY:
raise ValueError("SKULABS_API_KEY environment variable is required")
@server.list_tools()
async def list_tools() -> List[Tool]:
"""List all available Skulabs MCP tools"""
return [
# Account/Profile Tools (CONFIRMED WORKING)
Tool(
name="get_account_info",
description="Get account information and user profile (CONFIRMED WORKING)",
inputSchema={
"type": "object",
"properties": {}
}
),
# Inventory Tools
Tool(
name="get_inventory",
description="Get inventory items by SKU or all inventory (MOCK DATA - API endpoint needs confirmation)",
inputSchema={
"type": "object",
"properties": {
"sku": {
"type": "string",
"description": "Specific SKU to retrieve (optional)"
},
"limit": {
"type": "integer",
"description": "Maximum number of items to return (default: 100)",
"default": 100
},
"offset": {
"type": "integer",
"description": "Number of items to skip (default: 0)",
"default": 0
}
}
}
),
Tool(
name="update_inventory",
description="Update inventory quantity for a specific SKU (MOCK DATA - API endpoint needs confirmation)",
inputSchema={
"type": "object",
"properties": {
"sku": {
"type": "string",
"description": "SKU to update"
},
"quantity": {
"type": "integer",
"description": "New quantity"
},
"location": {
"type": "string",
"description": "Location to update (optional)"
}
},
"required": ["sku", "quantity"]
}
),
# Product Tools
Tool(
name="get_products",
description="Get products by SKU or all products (MOCK DATA - API endpoint needs confirmation)",
inputSchema={
"type": "object",
"properties": {
"sku": {
"type": "string",
"description": "Specific SKU to retrieve (optional)"
},
"limit": {
"type": "integer",
"description": "Maximum number of products to return (default: 100)",
"default": 100
},
"offset": {
"type": "integer",
"description": "Number of products to skip (default: 0)",
"default": 0
}
}
}
),
Tool(
name="get_product",
description="Get detailed information about a specific product (MOCK DATA - API endpoint needs confirmation)",
inputSchema={
"type": "object",
"properties": {
"sku": {
"type": "string",
"description": "Product SKU"
}
},
"required": ["sku"]
}
),
# Order Tools
Tool(
name="get_orders",
description="Get orders with optional status filtering (MOCK DATA - API endpoint needs confirmation)",
inputSchema={
"type": "object",
"properties": {
"status": {
"type": "string",
"description": "Filter by order status (optional)"
},
"limit": {
"type": "integer",
"description": "Maximum number of orders to return (default: 100)",
"default": 100
},
"offset": {
"type": "integer",
"description": "Number of orders to skip (default: 0)",
"default": 0
}
}
}
),
Tool(
name="get_order",
description="Get detailed information about a specific order (MOCK DATA - API endpoint needs confirmation)",
inputSchema={
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "Order ID"
}
},
"required": ["order_id"]
}
),
# Customer Tools
Tool(
name="get_customers",
description="Get customers with optional email filtering (MOCK DATA - API endpoint needs confirmation)",
inputSchema={
"type": "object",
"properties": {
"email": {
"type": "string",
"description": "Filter by customer email (optional)"
},
"limit": {
"type": "integer",
"description": "Maximum number of customers to return (default: 100)",
"default": 100
},
"offset": {
"type": "integer",
"description": "Number of customers to skip (default: 0)",
"default": 0
}
}
}
),
# Analytics Tools
Tool(
name="get_sales_summary",
description="Get sales summary for a date range (MOCK DATA - API endpoint needs confirmation)",
inputSchema={
"type": "object",
"properties": {
"start_date": {
"type": "string",
"description": "Start date (YYYY-MM-DD format, optional)"
},
"end_date": {
"type": "string",
"description": "End date (YYYY-MM-DD format, optional)"
}
}
}
),
Tool(
name="get_inventory_summary",
description="Get inventory summary statistics (MOCK DATA - API endpoint needs confirmation)",
inputSchema={
"type": "object",
"properties": {}
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle tool calls to Skulabs API"""
logger.info("Tool called", tool_name=name, arguments=arguments)
try:
async with SkulabsClient(SKULABS_API_KEY, SKULABS_BASE_URL) as client:
result = await execute_tool(client, name, arguments)
# Add helpful context about data status
if isinstance(result, dict) and result.get("status") == "mock_data":
result["note"] = "⚠️ This is mock data. Contact Skulabs support (api-support@skulabs.com) to get the correct API endpoints for production use."
return [TextContent(type="text", text=json.dumps(result, indent=2))]
except SkulabsAPIError as e:
error_msg = f"Skulabs API Error: {str(e)}"
logger.error("API error", error=error_msg)
return [TextContent(type="text", text=error_msg)]
except Exception as e:
error_msg = f"Unexpected error: {str(e)}"
logger.error("Unexpected error", error=error_msg)
return [TextContent(type="text", text=error_msg)]
async def execute_tool(client: SkulabsClient, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the specified tool with given arguments"""
# Account Tools (CONFIRMED WORKING)
if name == "get_account_info":
return await client.get_account_info()
# Inventory Tools
elif name == "get_inventory":
return await client.get_inventory(
arguments.get("sku"),
arguments.get("limit", 100),
arguments.get("offset", 0)
)
elif name == "update_inventory":
return await client.update_inventory(
arguments["sku"],
arguments["quantity"],
arguments.get("location")
)
# Product Tools
elif name == "get_products":
return await client.get_products(
arguments.get("sku"),
arguments.get("limit", 100),
arguments.get("offset", 0)
)
elif name == "get_product":
return await client.get_product(arguments["sku"])
# Order Tools
elif name == "get_orders":
return await client.get_orders(
arguments.get("status"),
arguments.get("limit", 100),
arguments.get("offset", 0)
)
elif name == "get_order":
return await client.get_order(arguments["order_id"])
# Customer Tools
elif name == "get_customers":
return await client.get_customers(
arguments.get("email"),
arguments.get("limit", 100),
arguments.get("offset", 0)
)
# Analytics Tools
elif name == "get_sales_summary":
return await client.get_sales_summary(
arguments.get("start_date"),
arguments.get("end_date")
)
elif name == "get_inventory_summary":
return await client.get_inventory_summary()
else:
raise ValueError(f"Unknown tool: {name}")
@server.list_resources()
async def list_resources() -> List[Resource]:
"""List available resources (none for this server)"""
return []
async def main():
"""Main entry point for the MCP server"""
logger.info("Starting Working Skulabs MCP Server")
# Test the connection first
try:
async with SkulabsClient(SKULABS_API_KEY, SKULABS_BASE_URL) as client:
account_info = await client.get_account_info()
logger.info("API connection successful", account_id=account_info.get("account_id"))
except Exception as e:
logger.error("Failed to connect to Skulabs API", error=str(e))
raise
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())