Turbify Store MCP Server

"""Data models for Turbify Store MCP Server.""" from typing import List, Optional, Any from pydantic import BaseModel, Field, ConfigDict, field_validator, model_validator from enum import Enum from typing_extensions import Self class OrderableStatus(str, Enum): """Item orderable status.""" YES = "yes" NO = "no" class TaxableStatus(str, Enum): """Item taxable status.""" YES = "yes" NO = "no" class APIStatus(str, Enum): """API response status.""" SUCCESS = "success" ERROR = "error" # Pydantic models for FastMCP schema generation class CustomData(BaseModel): name: str = Field(description="Name of the custom attribute") value: str = Field(description="Value of the custom attribute") @model_validator(mode='after') def validate_custom_data(self) -> Self: """Validate custom data fields""" # Ensure name is not empty or just whitespace if not self.name.strip(): raise ValueError("Custom data name cannot be empty") # Value can be empty, but if provided should not be just whitespace # (empty values might be valid for some use cases) return self class ItemOption(BaseModel): name: str = Field(description="Name of the option (e.g., 'Size', 'Color')") value_list: List[str] = Field(description="List of possible values for this option") @model_validator(mode='after') def validate_option(self) -> Self: """Validate option fields""" # Ensure option name is not empty if not self.name.strip(): raise ValueError("Option name cannot be empty") # Ensure all values are non-empty and unique if not self.value_list: raise ValueError("Option must have at least one value") cleaned_values = [v.strip() for v in self.value_list if v.strip()] if len(cleaned_values) != len(self.value_list): raise ValueError("Option values cannot be empty or just whitespace") if len(cleaned_values) != len(set(cleaned_values)): raise ValueError("Option values must be unique") return self class CatalogItem(BaseModel): # Required fields id: str = Field(description="Unique item ID") name: str = Field(description="Item name") price: float = Field(description="Item price", gt=0) orderable: bool = Field(OrderableStatus.YES, description="Whether item is orderable") taxable: bool = Field(TaxableStatus.NO, description="Whether item is taxable") table_id: str = Field(description="Table ID where item belongs") # Optional pricing fields sale_price: Optional[float] = Field(None, description="Sale price of the item") ship_weight: Optional[float] = Field(None, description="Shipping weight of the item") # Optional descriptive fields headline: Optional[str] = Field(None, description="Item headline (used instead of name on item page)") caption: Optional[str] = Field(None, description="Item description/caption") abstract: Optional[str] = Field(None, description="Text used for description on other pages") label: Optional[str] = Field(None, description="Text used when item appears as special on home page") # Optional product information manufacturer: Optional[str] = Field(None, description="Item manufacturer") brand: Optional[str] = Field(None, description="Item brand") gender: Optional[str] = Field(None, description="Gender (men/women/unisex) - for apparel") color: Optional[str] = Field(None, description="Color of the item") size: Optional[str] = Field(None, description="Size of the item") # Optional product codes and identifiers upc: Optional[str] = Field(None, description="Universal Product Code (12-digit)") manufacturer_part_number: Optional[str] = Field(None, description="Manufacturer's part number") model_number: Optional[str] = Field(None, description="Model number") isbn: Optional[str] = Field(None, description="International Standard Book Number") ean: Optional[str] = Field(None, description="European Article Number (13-digit)") # Optional categorization classification: Optional[str] = Field(None, description="Item classification", pattern="^(new|overstock|damaged|returned|refurbished|open box|liquidation|used)$") condition: Optional[str] = Field(None, description="Item condition", pattern="^(New|Like new|Very good|Good|Acceptable|Refurbished|Used)$") merchant_category: Optional[str] = Field(None, description="Merchant category for Yahoo Shopping") # Optional availability and shipping availability: Optional[str] = Field(None, description="Availability status", pattern="^(SAME_DAY|NEXT_DAY|2_3_DAY|3_4_DAY|5_7_DAY|1_2_WEEKS|2_3_WEEKS|4_6_WEEKS|6_8_WEEKS|CONTACT_US|IN_STOCK|AVAILABLE|OUT_OF_STOCK|PRE_ORDER)$") # Optional checkout requirements need_bill: Optional[bool] = Field(None, description="Whether billing address is required") need_payment: Optional[bool] = Field(None, description="Whether payment information is required") need_pay_ship: Optional[bool] = Field(None, description="Whether shipping address is required") # Optional pricing and promotions personalization_charge: Optional[float] = Field(None, description="Charge for monogram/inscription") msrp: Optional[str] = Field(None, description="Manufacturer's suggested retail price") # Optional web and shopping integration product_url: Optional[str] = Field(None, description="Product URL") inyshopping: Optional[bool] = Field(None, description="Include in Yahoo Shopping") yahoo_shopping_category: Optional[str] = Field(None, description="Yahoo Shopping category") # Optional age and media information age_group: Optional[str] = Field(None, description="Age group", pattern="^(infant|toddler|child|pre-teen|teen|adult)$") age_range: Optional[str] = Field(None, description="Age range (for toys)") medium: Optional[str] = Field(None, description="Media type for music/video", pattern="^(CD|Casette|MiniDisc|LPd|EP|45|VHS|Beta|8mm|Laser Disc|DVD|VCD)$") # Optional style information style_number: Optional[str] = Field(None, description="Style number (for apparel/home & garden)") style: Optional[str] = Field(None, description="Style description (e.g., 'chino', 'denim')") promo_text: Optional[str] = Field(None, description="Promotional text (up to 50 chars)", max_length=50) # Optional flags gift_cert: Optional[bool] = Field(None, description="Whether item is a gift certificate") # Custom data and options custom_data: List[CustomData] = Field(default=[], description="List of custom attributes") options: List[ItemOption] = Field(default=[], description="List of item options (size, color, etc.)") # @model_validator(mode='after') # def validate_catalog_item(self) -> Self: # """Validate catalog item according to Yahoo API requirements""" # # Validate sale price is not higher than regular price # if self.sale_price is not None and self.sale_price > self.price: # raise ValueError("Sale price cannot be higher than regular price") # # Validate UPC format (12 digits) # if self.upc is not None and not (self.upc.isdigit() and len(self.upc) == 12): # raise ValueError("UPC must be exactly 12 digits") # # Validate EAN format (13 digits) # if self.ean is not None and not (self.ean.isdigit() and len(self.ean) == 13): # raise ValueError("EAN must be exactly 13 digits") # # Validate ISBN format (10 or 13 digits) # if self.isbn is not None: # isbn_clean = self.isbn.replace('-', '').replace(' ', '') # if not (isbn_clean.isdigit() and len(isbn_clean) in [10, 13]): # raise ValueError("ISBN must be 10 or 13 digits (hyphens/spaces allowed)") # # Validate weight is positive # if self.ship_weight is not None and self.ship_weight <= 0: # raise ValueError("Ship weight must be positive") # # Validate personalization charge is not negative # if self.personalization_charge is not None and self.personalization_charge < 0: # raise ValueError("Personalization charge cannot be negative") # # Validate URL format # if self.product_url is not None: # if not (self.product_url.startswith('http://') or self.product_url.startswith('https://')): # raise ValueError("Product URL must start with http:// or https://") # # Validate custom data doesn't exceed limit (100 per API spec) # if len(self.custom_data) > 100: # raise ValueError("Cannot have more than 100 custom data attributes") # # Validate custom data names are unique # custom_names = [cd.name for cd in self.custom_data] # if len(custom_names) != len(set(custom_names)): # raise ValueError("Custom data attribute names must be unique") # # Validate option names are unique # option_names = [opt.name for opt in self.options] # if len(option_names) != len(set(option_names)): # raise ValueError("Option names must be unique") # # Validate each option has at least one value # for option in self.options: # if not option.value_list: # raise ValueError(f"Option '{option.name}' must have at least one value") # # Business logic validations # if not self.orderable and self.price > 0: # # Warning: might want to log this rather than raise # pass # Non-orderable items with prices might be valid for display # # Validate required fields for specific classifications # if self.classification == "used" and self.condition is None: # raise ValueError("Condition is required when classification is 'used'") # # Validate gender field for apparel # if self.gender is not None and self.gender not in ["men", "women", "unisex"]: # raise ValueError("Gender must be 'men', 'women', or 'unisex'") class SearchQuery(BaseModel): """Model for catalog search queries.""" keyword: str = Field(..., min_length=1, description="Search keyword") start_index: int = Field(1, ge=1, description="Start index for pagination") end_index: int = Field(100, ge=1, le=1000, description="End index for pagination") @model_validator(mode='after') def validate_indices(self) -> 'SearchQuery': """Ensure end_index is greater than start_index.""" if self.end_index <= self.start_index: raise ValueError('end_index must be greater than start_index') return self # Response Models class APIError(BaseModel): """API error information.""" code: str = Field(..., description="Error code") message: str = Field(..., description="Error message") class APIMessage(BaseModel): """API success message information.""" code: str = Field(..., description="Message code") message: str = Field(..., description="Message text") class Item(BaseModel): """Catalog item information.""" id: Optional[str] = Field(None, description="Item ID", alias="ID") name: Optional[str] = Field(None, description="Item name", alias="Name") code: Optional[str] = Field(None, description="Item code", alias="Code") price: Optional[float] = Field(None, description="Item price", alias="Price") sale_price: Optional[float] = Field(None, description="Sale price", alias="SalePrice") ship_weight: Optional[float] = Field(None, description="Shipping weight", alias="ShipWeight") orderable: Optional[bool] = Field(None, description="Orderable status", alias="Orderable") taxable: Optional[bool] = Field(None, description="Taxable status", alias="Taxable") # Fixed: Changed parameter name for Pydantic v2 model_config = ConfigDict(extra="allow", populate_by_name=True) @field_validator('price', 'sale_price', 'ship_weight', mode='before') @classmethod def convert_numeric_fields(cls, value: Any) -> Optional[float]: """Convert string values to floats, handling empty strings and None""" if value in (None, ""): return None try: return float(value) except (TypeError, ValueError): return None class APIResponse(BaseModel): """Base API response model.""" status: APIStatus = Field(..., description="Response status") errors: Optional[List[APIError]] = Field(None, description="List of errors if any") messages: Optional[List[APIMessage]] = Field(None, description="List of success messages") items: Optional[List[Item]] = Field(None, description="List of items (for search/get operations)") item_ids: Optional[List[str]] = Field(None, description="List of item IDs (for some operations)") @property def is_success(self) -> bool: """Check if the response indicates success.""" return self.status == APIStatus.SUCCESS @property def error_messages(self) -> List[str]: """Get list of error messages.""" if not self.errors: return [] return [error.message for error in self.errors] @property def success_messages(self) -> List[str]: """Get list of success messages.""" if not self.messages: return [] return [message.message for message in self.messages] class StoreConfig(BaseModel): """Store configuration information.""" store_id: str = Field(..., description="Turbify Store ID") api_base: str = Field(..., description="API base URL") max_items_per_call: int = Field(..., description="Maximum items per API call") contract_token_configured: bool = Field(..., description="Whether contract token is configured") api_version: str = Field(..., description="API version") request_timeout: int = Field(..., description="Request timeout in seconds") # Utility types XMLElement = Any # xml.etree.ElementTree.Element, but avoiding import here