from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from ..client.saleor_client import SaleorClient
from ..utils.logger import logger
from ..utils.exceptions import SaleorMCPError
class BaseTool(ABC):
"""Base class for all tools"""
def __init__(self, client: SaleorClient):
self.client = client
self.logger = logger
@property
@abstractmethod
def name(self) -> str:
"""Tool name for MCP registration"""
pass
@property
@abstractmethod
def description(self) -> str:
"""Tool description"""
pass
@abstractmethod
async def execute(self, **kwargs) -> Any:
"""Execute the tool with given parameters"""
pass
def _validate_limit(
self, limit: Optional[int], max_limit: int, defualt_limit: int
) -> int:
"""Validate and normalize limit parameter"""
if limit is None:
return defualt_limit
if limit <= 0:
raise SaleorMCPError(f"Limit must be positive, got {limit}")
if limit > max_limit:
self.logger.warning(
f"Limit {limit} exceeds maximum {max_limit}, using {max_limit}"
)
return max_limit
return limit
def _handle_errors(self, result: Dict[str, Any]) -> None:
"""Handle GraphQL errors in API responses."""
if "errors" in result:
error_messages = [
error.get("message", str(error)) for error in result["errors"]
]
raise SaleorMCPError(f"GraphQL errors: {'; '.join(error_messages)}")
class ProductTool(BaseTool):
"""Base class for product-related tools"""
def _extract_products(self, result: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Extract products from GraphQL result"""
self._handle_errors(result)
products_data = result.get("data", {}).get("products", {})
if not products_data:
return []
edges = products_data.get("edges", [])
return [edge["node"] for edge in edges if "node" in edge]
class OrderTool(BaseTool):
"""Base class for order-related tools."""
def _extract_orders(self, result: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Extract orders from GraphQL result."""
self._handle_errors(result)
orders_data = result.get("data", {}).get("orders", {})
if not orders_data:
return []
edges = orders_data.get("edges", [])
return [edge["node"] for edge in edges if "node" in edge]
