coles-woolworths

Instructions

Implementation Reference

• main.py:16-59 (handler)

  The primary handler function for the 'get_coles_products' tool. It is registered via @mcp.tool() decorator and implements the tool logic by invoking helper functions for searching and extracting Coles products, applying limits, and formatting the output.
  @mcp.tool() async def get_coles_products(query: str, store_id: str = COLES_DEFAULT_STORE_ID, limit: int = 10) -> str: """Search for products at Coles. Args: query: The product search query. store_id: The Coles store ID to search in. limit: Maximum number of products to return. """ try: # The limit parameter in coles_search_products's signature is not used in its API call. # We will fetch all available (up to API's own limit) and then slice. # Also, the original coles_search_products signature includes a limit, but it's not used. # Forcing keyword arguments for clarity with asyncio.to_thread search_results = await asyncio.to_thread( coles_search_products, query=query, store_id=store_id # limit parameter is not passed here as the underlying coles_search_products doesn't use it in API call ) if search_results.get("status") == "error": return f"Error fetching Coles products: {search_results.get('message', 'Unknown error')}\nResponse: {search_results.get('response_text', '')}" # extract_products is CPU-bound/quick, but run in thread for consistency with I/O products = await asyncio.to_thread(coles_extract_products, search_results) # Apply limit after extraction products = products[: min(limit, len(products))] if not products: return f"No products found at Coles 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_coles_products: {str(e)}"
• src/supermarkets/coles.py:38-95 (helper)

  Core helper function (aliased as coles_search_products via __init__.py) that performs the HTTP request to Coles API for product search.
  def search_products(query: str, store_id: str = DEFAULT_STORE_ID, limit: int = 10) -> Dict[str, Any]: """ Search for products using the Coles API. Args: query (str): The search query store_id (str, optional): The store ID to search in. Defaults to DEFAULT_STORE_ID. limit (int, optional): The maximum number of results to return. Defaults to 10. Returns: Dict[str, Any]: The search results """ try: # Use the exact URL format provided by the user search_url = f"{API_URL}/search" # Set up the API request parameters for search using the exact format provided search_params = { "storeId": store_id, "searchTerm": query, "start": "0", "sortBy": "salesDescending", "excludeAds": "true", "authenticated": "false", "subscription-key": API_KEY } # 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(search_url, params=search_params, 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() return { "status": "success", "query": query, "store_id": store_id, "response_data": response_data } except Exception as e: return { "status": "error", "message": str(e) }
• src/supermarkets/coles.py:113-198 (helper)

  Core helper function (aliased as coles_extract_products via __init__.py) that parses the API response to extract structured product data including name, price, unit, and store.
  def extract_products(search_results: Dict[str, Any]) -> List[Dict[str, Any]]: """ Extract product information from search results. Args: search_results (Dict[str, Any]): The search results from the Coles API Returns: List[Dict[str, Any]]: The extracted product information """ products = [] if search_results["status"] != "success": return products response_data = search_results["response_data"] # Check if the response contains products if "results" in response_data and response_data["results"]: api_products = response_data["results"] # Process the products for product in api_products: try: # Extract product information name = product.get("name", "") # Extract price information pricing = product.get("pricing", {}) now_price = pricing.get("now", None) was_price = pricing.get("was", None) # Use the current price if available, otherwise use the regular price price = now_price if now_price is not None else was_price # Extract unit information - check multiple fields unit = "" # Check package_size field package_size = product.get("packageSize", "") # Check quantity field quantity = product.get("quantity", "") # Check size field size = product.get("size", "") # Check description field for unit information description = product.get("description", "") # Try to extract unit information from available fields unit_fields = [package_size, quantity, size, description] for field in unit_fields: if not field: continue field = field.lower() if "kg" in field: unit = "kg" break elif "g" in field and not "mg" in field: unit = "g" break elif "l" in field and not "ml" in field: unit = "L" break elif "ml" in field: unit = "ml" break elif "each" in field: unit = "each" break elif "pack" in field: unit = "pack" break products.append({ "name": name, "price": float(price) if price is not None else None, "unit": unit, "store": STORE_NAME }) except Exception as e: continue return products
• main.py:16-16 (registration)

  The @mcp.tool() decorator registers the get_coles_products function as an MCP tool.
  @mcp.tool()
• src/supermarkets/coles.py:21-21 (helper)

  Default store ID constant used as COLES_DEFAULT_STORE_ID in the tool signature.
  DEFAULT_STORE_ID = "0584"