advanced_search_catalog_items
Search Turbify Store catalog items using multiple criteria with flexible matching options and pagination controls.
Instructions
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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| criteria_list | Yes | ||
| match_type | No | all | |
| table_id | No | ||
| start_index | No | ||
| end_index | No |
Implementation Reference
- The core handler function for the 'advanced_search_catalog_items' tool. It validates input criteria, performs type checks on attributes and operators, and calls the Turbify API's advanced_search method to retrieve matching item IDs.
@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) - Input schema defined by function type hints (List[dict] for criteria, strings, ints) and comprehensive docstring explaining parameters, operators, attributes, and validation rules.
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 """ - src/turbify_mcp/server.py:33-33 (registration)Top-level registration of catalog tools (including advanced_search_catalog_items) by calling register_catalog_tools on the MCP server instance during server setup.
register_catalog_tools(mcp) - src/turbify_mcp/tools/catalog_tools.py:15-396 (registration)Function that defines all catalog tools using @mcp.tool() decorators, effectively registering them when called.
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)