skulabs_mcp_server.py•18.5 kB
"""
Skulabs MCP Server
Exposes Skulabs API functionality as MCP tools for AI agents
"""
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 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 [
# Inventory Tools
Tool(
name="get_inventory",
description="Get inventory items by SKU or all inventory with optional filtering",
inputSchema={
"type": "object",
"properties": {
"sku": {
"type": "string",
"description": "Specific SKU to retrieve (optional)"
},
"location": {
"type": "string",
"description": "Filter by location (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",
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"]
}
),
Tool(
name="get_inventory_by_location",
description="Get all inventory items for a specific location",
inputSchema={
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "Location name"
},
"limit": {
"type": "integer",
"description": "Maximum number of items to return (default: 100)",
"default": 100
}
},
"required": ["location"]
}
),
# Product Tools
Tool(
name="get_products",
description="Get products by SKU or all products with optional filtering",
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",
inputSchema={
"type": "object",
"properties": {
"sku": {
"type": "string",
"description": "Product SKU"
}
},
"required": ["sku"]
}
),
Tool(
name="create_product",
description="Create a new product in Skulabs",
inputSchema={
"type": "object",
"properties": {
"sku": {
"type": "string",
"description": "Product SKU"
},
"name": {
"type": "string",
"description": "Product name"
},
"description": {
"type": "string",
"description": "Product description"
},
"price": {
"type": "number",
"description": "Product price"
},
"cost": {
"type": "number",
"description": "Product cost"
},
"weight": {
"type": "number",
"description": "Product weight"
},
"dimensions": {
"type": "object",
"properties": {
"length": {"type": "number"},
"width": {"type": "number"},
"height": {"type": "number"}
}
}
},
"required": ["sku", "name"]
}
),
# Order Tools
Tool(
name="get_orders",
description="Get orders with optional status filtering",
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",
inputSchema={
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "Order ID"
}
},
"required": ["order_id"]
}
),
Tool(
name="create_order",
description="Create a new order in Skulabs",
inputSchema={
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "Customer ID"
},
"items": {
"type": "array",
"items": {
"type": "object",
"properties": {
"sku": {"type": "string"},
"quantity": {"type": "integer"},
"price": {"type": "number"}
},
"required": ["sku", "quantity"]
}
},
"shipping_address": {
"type": "object",
"properties": {
"name": {"type": "string"},
"address1": {"type": "string"},
"city": {"type": "string"},
"state": {"type": "string"},
"zip": {"type": "string"},
"country": {"type": "string"}
}
},
"notes": {
"type": "string",
"description": "Order notes"
}
},
"required": ["customer_id", "items"]
}
),
Tool(
name="update_order_status",
description="Update the status of an existing order",
inputSchema={
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "Order ID"
},
"status": {
"type": "string",
"description": "New order status",
"enum": ["pending", "processing", "shipped", "delivered", "cancelled"]
}
},
"required": ["order_id", "status"]
}
),
# Customer Tools
Tool(
name="get_customers",
description="Get customers with optional email filtering",
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
}
}
}
),
Tool(
name="get_customer",
description="Get detailed information about a specific customer",
inputSchema={
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "Customer ID"
}
},
"required": ["customer_id"]
}
),
Tool(
name="create_customer",
description="Create a new customer in Skulabs",
inputSchema={
"type": "object",
"properties": {
"first_name": {
"type": "string",
"description": "Customer first name"
},
"last_name": {
"type": "string",
"description": "Customer last name"
},
"email": {
"type": "string",
"description": "Customer email"
},
"phone": {
"type": "string",
"description": "Customer phone number"
},
"address": {
"type": "object",
"properties": {
"address1": {"type": "string"},
"city": {"type": "string"},
"state": {"type": "string"},
"zip": {"type": "string"},
"country": {"type": "string"}
}
}
},
"required": ["first_name", "last_name", "email"]
}
),
# Analytics Tools
Tool(
name="get_sales_summary",
description="Get sales summary for a date range",
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",
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)
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"""
# Inventory Tools
if name == "get_inventory":
sku = arguments.get("sku")
location = arguments.get("location")
limit = arguments.get("limit", 100)
offset = arguments.get("offset", 0)
if location:
return await client.get_inventory_by_location(location, limit)
else:
return await client.get_inventory(sku, limit, offset)
elif name == "update_inventory":
return await client.update_inventory(
arguments["sku"],
arguments["quantity"],
arguments.get("location")
)
elif name == "get_inventory_by_location":
return await client.get_inventory_by_location(
arguments["location"],
arguments.get("limit", 100)
)
# 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"])
elif name == "create_product":
return await client.create_product(arguments)
# 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"])
elif name == "create_order":
return await client.create_order(arguments)
elif name == "update_order_status":
return await client.update_order_status(
arguments["order_id"],
arguments["status"]
)
# Customer Tools
elif name == "get_customers":
return await client.get_customers(
arguments.get("email"),
arguments.get("limit", 100),
arguments.get("offset", 0)
)
elif name == "get_customer":
return await client.get_customer(arguments["customer_id"])
elif name == "create_customer":
return await client.create_customer(arguments)
# 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 Skulabs MCP Server")
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())