skulabs_client.py•5.96 kB
"""
Skulabs API Client for MCP Server
Handles authentication and API communication with Skulabs
"""
import httpx
import structlog
from typing import Dict, Any, Optional, List
from pydantic import BaseModel
import os
logger = structlog.get_logger()
class SkulabsAPIError(Exception):
"""Custom exception for Skulabs API errors"""
pass
class SkulabsClient:
"""Client for interacting with Skulabs API"""
def __init__(self, api_key: str, base_url: str = "https://app.skulabs.com"):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.client = httpx.AsyncClient(
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"Accept": "application/json"
},
timeout=30.0
)
logger.info("Skulabs client initialized", base_url=base_url)
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.client.aclose()
async def _make_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
"""Make HTTP request to Skulabs API"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
try:
logger.debug("Making API request", method=method, url=url)
response = await self.client.request(method, url, **kwargs)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
logger.error("API request failed", status_code=e.response.status_code,
response=e.response.text)
raise SkulabsAPIError(f"API request failed: {e.response.status_code} - {e.response.text}")
except httpx.RequestError as e:
logger.error("Request error", error=str(e))
raise SkulabsAPIError(f"Request failed: {str(e)}")
# Inventory Methods
async def get_inventory(self, sku: Optional[str] = None, limit: int = 100, offset: int = 0) -> Dict[str, Any]:
"""Get inventory items"""
params = {"limit": limit, "offset": offset}
if sku:
params["sku"] = sku
return await self._make_request("GET", "/inventory", params=params)
async def update_inventory(self, sku: str, quantity: int, location: Optional[str] = None) -> Dict[str, Any]:
"""Update inventory quantity"""
data = {"quantity": quantity}
if location:
data["location"] = location
return await self._make_request("PUT", f"/inventory/{sku}", json=data)
async def get_inventory_by_location(self, location: str, limit: int = 100) -> Dict[str, Any]:
"""Get inventory by location"""
params = {"location": location, "limit": limit}
return await self._make_request("GET", "/inventory", params=params)
# Product Methods
async def get_products(self, sku: Optional[str] = None, limit: int = 100, offset: int = 0) -> Dict[str, Any]:
"""Get products"""
params = {"limit": limit, "offset": offset}
if sku:
params["sku"] = sku
return await self._make_request("GET", "/products", params=params)
async def get_product(self, sku: str) -> Dict[str, Any]:
"""Get specific product by SKU"""
return await self._make_request("GET", f"/products/{sku}")
async def create_product(self, product_data: Dict[str, Any]) -> Dict[str, Any]:
"""Create new product"""
return await self._make_request("POST", "/products", json=product_data)
# Order Methods
async def get_orders(self, status: Optional[str] = None, limit: int = 100, offset: int = 0) -> Dict[str, Any]:
"""Get orders"""
params = {"limit": limit, "offset": offset}
if status:
params["status"] = status
return await self._make_request("GET", "/orders", params=params)
async def get_order(self, order_id: str) -> Dict[str, Any]:
"""Get specific order"""
return await self._make_request("GET", f"/orders/{order_id}")
async def create_order(self, order_data: Dict[str, Any]) -> Dict[str, Any]:
"""Create new order"""
return await self._make_request("POST", "/orders", json=order_data)
async def update_order_status(self, order_id: str, status: str) -> Dict[str, Any]:
"""Update order status"""
return await self._make_request("PUT", f"/orders/{order_id}/status", json={"status": status})
# Customer Methods
async def get_customers(self, email: Optional[str] = None, limit: int = 100, offset: int = 0) -> Dict[str, Any]:
"""Get customers"""
params = {"limit": limit, "offset": offset}
if email:
params["email"] = email
return await self._make_request("GET", "/customers", params=params)
async def get_customer(self, customer_id: str) -> Dict[str, Any]:
"""Get specific customer"""
return await self._make_request("GET", f"/customers/{customer_id}")
async def create_customer(self, customer_data: Dict[str, Any]) -> Dict[str, Any]:
"""Create new customer"""
return await self._make_request("POST", "/customers", json=customer_data)
# Analytics Methods
async def get_sales_summary(self, start_date: Optional[str] = None, end_date: Optional[str] = None) -> Dict[str, Any]:
"""Get sales summary"""
params = {}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
return await self._make_request("GET", "/analytics/sales", params=params)
async def get_inventory_summary(self) -> Dict[str, Any]:
"""Get inventory summary"""
return await self._make_request("GET", "/analytics/inventory")