Turbify Store MCP Server

Instructions

Implementation Reference

• src/turbify_mcp/tools/catalog_tools.py:277-333 (handler)

  The MCP tool handler for 'create_items'. It validates input, calls the Turbify API client to create the items, and returns a JSON-formatted response with status, messages, errors, and items processed.
  def create_items(items: List[CatalogItem]) -> str: """ Create items in Turbify Merchant Solutions catalog using the Catalog API. HTML included in the request values must be contained within CDATA tags. Args: items: List of item dictionaries to create. Each item must have: - id (str): Unique item ID - name (str): Item name - price (float): Item price - orderable (bool): Whether item is orderable - taxable (bool): Whether item is taxable - table_id (str): Table ID where item belongs - custom_data (list, optional): List of {"name": str, "value": str} custom attributes Returns: JSON string with creation results including success status and any error messages """ # Debugging: Check if items is None or contains None values if items is None: logger.error("create_items called with None items") return json.dumps({ "status": "error", "success": False, "errors": ["items parameter is None"], "items_processed": 0 }, indent=2) # Check for None values in the list none_items = [i for i in items if i is None] if none_items: logger.error(f"create_items called with None items in list: {none_items}") return json.dumps({ "status": "error", "success": False, "errors": [f"items list contains {len(none_items)} None values"], "items_processed": 0 }, indent=2) try: response = client.create_items(items) return json.dumps({ "status": response.status, "success": response.is_success, "messages": response.success_messages, "errors": response.error_messages, "items_processed": len(items) }, indent=2) except APIError as e: return json.dumps({ "status": "error", "success": False, "errors": [str(e)], "items_processed": 0 }, indent=2)
• src/turbify_mcp/models/models.py:66-142 (schema)

  Pydantic BaseModel defining the structure and validation for CatalogItem, which is the input type for the create_items tool (List[CatalogItem]). Includes required fields like id, name, price, etc., and many optional fields.
  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.)")
• src/turbify_mcp/server.py:33-33 (registration)

  The call to register_catalog_tools(mcp) in the main server setup, which registers the create_items tool (and others) with the MCP server.
  register_catalog_tools(mcp)
• src/turbify_mcp/client/turbify_api_client.py:557-587 (helper)

  Helper method in TurbifyStoreAPIClient that handles the actual API request to create items, converting CatalogItem models to the required XML structure and making the HTTP call.
  def create_items(self, items_data: List[CatalogItem]) -> APIResponse: """Create multiple items in a single API call.""" self.logger.debug(f"Batch create: {len(items_data)} items") if len(items_data) > self.config.max_items_per_call: raise ValueError(f"Cannot create more than {self.config.max_items_per_call} items at once") # Convert all ItemData to ItemType item_types = [self._item_data_to_item_type(item_data) for item_data in items_data] # Extract item IDs from input data item_ids = [item_data.id for item_data in items_data if hasattr(item_data, 'id') and item_data.id] # Build catalog structure item_list = ItemListType(item=item_types) catalog = CatalogType(item_list=item_list) resource_list = RequestResourceListType(catalog=catalog) # Create request request = self._create_base_request("create") request.resource_list = resource_list response_dict = self._make_api_call("Catalog", request) # Return APIResponse with item IDs return APIResponse( status="success" if response_dict.get('success') else "error", item_ids=item_ids, errors=response_dict.get('errors', []), warnings=response_dict.get('warnings', []), messages=response_dict.get('info', []) )
• src/turbify_mcp/tools/catalog_tools.py:15-396 (registration)

  The registration function that defines and registers the create_items tool using @mcp.tool() decorator inside it.
  def register_catalog_tools(mcp): """Register catalog management tools with the MCP server.""" # Initialize client (will be reused across tool calls) client = TurbifyStoreAPIClient() @mcp.tool() def get_catalog_items(item_ids: List[str]) -> str: """ Get details for multiple catalog items. Args: item_ids: List of item IDs to retrieve Returns: JSON string with item details """ try: response = client.get_item_details(item_ids) # Convert items to dictionaries if they're not already items = response.items if response.items else [] serializable_items = [] for item in items: if item and not isinstance(item, dict): # If it's a Pydantic model, convert to dict if hasattr(item, 'dict'): serializable_items.append(item.dict()) # If it's a dataclass, convert to dict elif hasattr(item, '__dict__'): serializable_items.append(item.__dict__) # Otherwise, try to serialize as is else: serializable_items.append(dict(item)) else: serializable_items.append(item) return json.dumps({ "status": response.status, "success": response.is_success, "messages": response.success_messages, "errors": response.error_messages, "item_count": len(serializable_items), "items": serializable_items }, indent=2) except APIError as e: return json.dumps({ "status": "error", "success": False, "errors": [str(e)], "item_ids": item_ids }, indent=2) @mcp.tool() def search_catalog_items( keyword: str, start_index: int = 1, end_index: int = 100 ) -> str: """ Search for items in the Turbify Store catalog. Note: This simple search only matches the keyword against item ID, name, or code fields. For more advanced search capabilities against other fields, use advanced_search_catalog_items. Args: keyword: Search keyword (matches against ID, name, or code fields only) start_index: Starting index for pagination (default: 1) end_index: Ending index for pagination (default: 100, max: 1000) Returns: JSON string with search results """ if end_index > 1000: end_index = 1000 try: response = client.simple_search(keyword, start_index, end_index) # Extract item IDs from the response item_ids = response.item_ids if response.item_ids else [] return json.dumps({ "status": response.status, "success": response.is_success, "messages": response.success_messages, "errors": response.error_messages, "keyword": keyword, "start_index": start_index, "end_index": end_index, "total_items": len(item_ids), "item_ids": item_ids }, indent=2) except APIError as e: return json.dumps({ "status": "error", "success": False, "errors": [str(e)], "keyword": keyword }, indent=2) @mcp.tool() def advanced_search_catalog_items( criteria_list: List[dict], match_type: str = "all", table_id: Optional[str] = None, start_index: int = 1, end_index: int = 100 ) -> str: """ Advanced search for items in the Turbify Store catalog with multiple criteria. Args: criteria_list: List of criteria dictionaries, each with 'attribute', 'operator', and 'value' Example: [{"attribute": "price", "operator": "gt", "value": "10.00"}] match_type: How to match criteria - "all" (AND) or "any" (OR) (default: "all") table_id: Optional table ID to search within start_index: Starting index for pagination (default: 1) end_index: Ending index for pagination (default: 100, max: 1000) Returns: JSON string with search results """ if end_index > 1000: end_index = 1000 # Validate match_type if match_type not in ["all", "any"]: return json.dumps({ "status": "error", "success": False, "errors": ["match_type must be 'all' or 'any'"], "criteria_list": criteria_list }, indent=2) # Validate criteria_list if not criteria_list: return json.dumps({ "status": "error", "success": False, "errors": ["At least one search criterion is required"], "criteria_list": criteria_list }, indent=2) # Define attribute types and supported operators based on API documentation string_attributes = { "id", "name", "code", "options", "headline", "caption", "abstract", "label", "product url", "manufacturer", "brand", "gender", "upc", "manufacturer-part-number", "model-number", "isbn", "merchant-category", "color", "size", "age-range", "promo-text", "style-number", "style", "ypath", "yahoo-shopping-category", "ean" } numeric_attributes = { "price", "sale-price", "ship-weight", "msrp", "personalization-charge" } boolean_attributes = { "orderable", "taxable", "gift-certificate", "need-bill", "need-payment", "need-ship", "in-yshopping" } enumeration_attributes = { "age-group", "availability", "classification", "condition", "gender", "medium" } all_valid_attributes = string_attributes | numeric_attributes | boolean_attributes | enumeration_attributes valid_operators = {"lt", "gt", "cn", "eq"} # Validate each criterion for i, criteria in enumerate(criteria_list): if not all(key in criteria for key in ["attribute", "operator", "value"]): return json.dumps({ "status": "error", "success": False, "errors": [f"Criterion {i} must have 'attribute', 'operator', and 'value' fields"], "criteria_list": criteria_list }, indent=2) attribute = criteria["attribute"] operator = criteria["operator"].lower() # Operators are case-sensitive and must be lowercase value = str(criteria["value"]) # Validate operator if operator not in valid_operators: return json.dumps({ "status": "error", "success": False, "errors": [f"Criterion {i}: Invalid operator '{operator}'. Valid operators are: {', '.join(valid_operators)}"], "criteria_list": criteria_list }, indent=2) # Validate attribute if attribute not in all_valid_attributes: # Check if it might be a custom field (we'll allow it but warn in documentation) pass # Custom fields are allowed # Validate operator against attribute type if attribute in boolean_attributes: if operator != "eq": return json.dumps({ "status": "error", "success": False, "errors": [f"Criterion {i}: Boolean attributes only support 'eq' operator"], "criteria_list": criteria_list }, indent=2) if value not in ["0", "1"]: return json.dumps({ "status": "error", "success": False, "errors": [f"Criterion {i}: Boolean attributes only accept values '0' or '1'"], "criteria_list": criteria_list }, indent=2) elif attribute in numeric_attributes: if operator in ["lt", "gt", "eq"]: try: float(value) except ValueError: return json.dumps({ "status": "error", "success": False, "errors": [f"Criterion {i}: Numeric attributes require numeric values"], "criteria_list": criteria_list }, indent=2) elif attribute in enumeration_attributes: if operator != "eq": return json.dumps({ "status": "error", "success": False, "errors": [f"Criterion {i}: Enumeration attributes only support 'eq' operator"], "criteria_list": criteria_list }, indent=2) try: response = client.advanced_search(criteria_list, match_type, table_id, start_index, end_index) # Extract item IDs from the response item_ids = response.item_ids if response.item_ids else [] return json.dumps({ "status": response.status, "success": response.is_success, "messages": response.success_messages, "errors": response.error_messages, "criteria_list": criteria_list, "match_type": match_type, "table_id": table_id, "start_index": start_index, "end_index": end_index, "total_items": len(item_ids), "item_ids": item_ids }, indent=2) except APIError as e: return json.dumps({ "status": "error", "success": False, "errors": [str(e)], "criteria_list": criteria_list }, indent=2) @mcp.tool() def create_items(items: List[CatalogItem]) -> str: """ Create items in Turbify Merchant Solutions catalog using the Catalog API. HTML included in the request values must be contained within CDATA tags. Args: items: List of item dictionaries to create. Each item must have: - id (str): Unique item ID - name (str): Item name - price (float): Item price - orderable (bool): Whether item is orderable - taxable (bool): Whether item is taxable - table_id (str): Table ID where item belongs - custom_data (list, optional): List of {"name": str, "value": str} custom attributes Returns: JSON string with creation results including success status and any error messages """ # Debugging: Check if items is None or contains None values if items is None: logger.error("create_items called with None items") return json.dumps({ "status": "error", "success": False, "errors": ["items parameter is None"], "items_processed": 0 }, indent=2) # Check for None values in the list none_items = [i for i in items if i is None] if none_items: logger.error(f"create_items called with None items in list: {none_items}") return json.dumps({ "status": "error", "success": False, "errors": [f"items list contains {len(none_items)} None values"], "items_processed": 0 }, indent=2) try: response = client.create_items(items) return json.dumps({ "status": response.status, "success": response.is_success, "messages": response.success_messages, "errors": response.error_messages, "items_processed": len(items) }, indent=2) except APIError as e: return json.dumps({ "status": "error", "success": False, "errors": [str(e)], "items_processed": 0 }, indent=2) @mcp.tool() def update_items(items: List[CatalogItem]) -> str: """ Update items in Turbify Merchant Solutions catalog using the Catalog API. TableId and Id cannot be updated. HTML included in the request values must be contained within CDATA tags. Args: items: List of item dictionaries to update. Each item must have: - id (str): Unique item ID, this cannot be updated Returns: JSON string with creation results including success status and any error messages """ try: response = client.update_items(items) return json.dumps({ "status": response.status, "success": response.is_success, "messages": response.success_messages, "errors": response.error_messages, "items_processed": len(items) }, indent=2) except APIError as e: return json.dumps({ "status": "error", "success": False, "errors": [str(e)], "items_processed": 0 }, indent=2) @mcp.tool() def delete_items(item_ids: List[str]) -> str: """ Delete multiple items in a single operation. Args: item_ids: List of item IDs to delete Returns: JSON string with operation result """ try: response = client.delete_items(item_ids) return json.dumps({ "status": response.status, "success": response.is_success, "messages": response.success_messages, "errors": response.error_messages, "items_processed": len(item_ids) }, indent=2) except APIError as e: return json.dumps({ "status": "error", "success": False, "errors": [str(e)], "items_processed": 0 }, indent=2)