coles-woolworths

Instructions

Implementation Reference

• main.py:61-92 (handler)

  The primary handler for the 'get_woolworths_products' tool. Decorated with @mcp.tool() for registration. Calls the woolworths_search_products helper, processes results, applies limit, formats output, and handles errors.
  @mcp.tool() async def get_woolworths_products(query: str, limit: int = 10) -> str: """Search for products at Woolworths. Args: query: The product search query. limit: Maximum number of products to return. """ try: search_results = await asyncio.to_thread(woolworths_search_products, query=query) if search_results.get("status") == "error": return f"Error fetching Woolworths products: {search_results.get('message', 'Unknown error')}\nResponse: {search_results.get('response_text', '')}" products = search_results.get("products", []) # Apply limit after fetching products = products[: min(limit, len(products))] if not products: return f"No products found at Woolworths for '{query}'." formatted_products = [] for p in products: price_str = f"${p['price']:.2f}" if p['price'] is not None else "N/A" unit_str = p['unit'] if p['unit'] else "N/A" # Ensure unit is not None formatted_products.append( f"Name: {p['name']}\nPrice: {price_str}\nUnit: {unit_str}\nStore: {p['store']}" ) return "\n---\n".join(formatted_products) except Exception as e: return f"An unexpected error occurred in get_woolworths_products: {str(e)}"
• main.py:61-61 (registration)

  The @mcp.tool() decorator registers the get_woolworths_products function as an MCP tool.
  @mcp.tool()
• src/supermarkets/woolworths.py:38-166 (helper)

  Core helper function `search_products` (imported as `woolworths_search_products`) that performs HTTP request to Woolworths search API, parses JSON response, extracts and formats product data including name, price, and unit.
  def search_products(query: str) -> Dict[str, Any]: """ Search for products using the Woolworths API. Args: query (str): The search query Returns: Dict[str, Any]: The search results """ try: # Format the API URL url = format_api_url(query) # Set up the headers headers = { "Accept": "application/json", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" } # Make the API request response = requests.get(url, headers=headers) # Check if the request was successful if response.status_code != 200: return { "status": "error", "message": f"API request failed with status code {response.status_code}", "response_text": response.text } # Parse the JSON response response_data = response.json() # Extract and format the products products = [] # The outer "Products" key in response_data contains a list of product categories/groups. # Each category/group then has its own "Products" key containing a list of actual products. if "Products" in response_data and response_data["Products"]: for product_category in response_data.get("Products", []): # product_category is a dict actual_product_list = product_category.get("Products") if not isinstance(actual_product_list, list): # Handle cases where product_category might be a product itself (e.g. direct search hit or promotion) if isinstance(product_category, dict) and "Stockcode" in product_category and "Products" not in product_category: actual_product_list = [product_category] # Treat the category as a single product list else: # If it's not a list and not a product-like dict, skip it. # print(f"Skipping product_category due to unexpected structure: {product_category.get('Name')}") continue for product in actual_product_list: # This 'product' is the actual product item # Extract product information name = product.get("DisplayName", product.get("Name", "")) # Extract price information price = product.get("Price") if price is None: price = product.get("InstorePrice") # Check InstorePrice as well if price is None: price = product.get("WasPrice") # Fallback to WasPrice # Extract unit information unit = "" package_size_str = product.get("PackageSize", "") cup_string_str = product.get("CupString", "") cup_measure_str = product.get("CupMeasure", "") api_unit_field = product.get("Unit", "") # Priority 1: PackageSize if package_size_str and isinstance(package_size_str, str): ps_lower = package_size_str.lower() if "kg" in ps_lower: unit = "kg" elif "g" in ps_lower and "kg" not in ps_lower: unit = "g" elif "l" in ps_lower and "ml" not in ps_lower: unit = "L" elif "ml" in ps_lower: unit = "ml" elif "each" in ps_lower: unit = "each" elif "pack" in ps_lower or "pk" in ps_lower: unit = "pack" # Priority 2: CupString if not unit and cup_string_str and isinstance(cup_string_str, str): cs_lower = cup_string_str.lower() parts = cs_lower.split('/') target_str_for_unit = parts[-1].strip() if len(parts) > 1 else cs_lower if "kg" in target_str_for_unit: unit = "kg" elif "g" in target_str_for_unit and "kg" not in target_str_for_unit: unit = "g" elif "l" in target_str_for_unit and "ml" not in target_str_for_unit: unit = "L" elif "ml" in target_str_for_unit: unit = "ml" elif "each" in target_str_for_unit or "ea" in target_str_for_unit: unit = "each" elif "pack" in target_str_for_unit or "pk" in target_str_for_unit: unit = "pack" # Priority 3: CupMeasure if not unit and cup_measure_str and isinstance(cup_measure_str, str): cm_lower = cup_measure_str.lower() if "kg" in cm_lower: unit = "kg" elif "g" in cm_lower and "kg" not in cm_lower: unit = "g" elif "l" in cm_lower and "ml" not in cm_lower: unit = "L" elif "ml" in cm_lower: unit = "ml" elif "each" in cm_lower: unit = "each" elif "pack" in cm_lower or "pk" in cm_lower: unit = "pack" # Priority 4: API "Unit" field (often "Each") if not unit and api_unit_field and isinstance(api_unit_field, str): auf_lower = api_unit_field.lower() if "each" == auf_lower: # Exact match for "each" unit = "each" # Can add other specific unit mappings from api_unit_field if needed # Add the product to the list products.append({ "name": name, "price": float(price) if price is not None else None, "unit": unit, "store": STORE_NAME }) return { "status": "success", "query": query, "products": products, "product_count": len(products) } except Exception as e: return { "status": "error", "message": str(e) }
• src/supermarkets/__init__.py:13-15 (helper)

  Package __init__.py exports `search_products` from woolworths.py as `woolworths_search_products` for import in main.py.
  from .woolworths import ( search_products as woolworths_search_products, )