coles-woolworths

Instructions

Args:
    query: The product search query.
    store_id: The Coles store ID to search in.
    limit: Maximum number of products to return.

Implementation Reference

• main.py:16-59 (handler)

  The primary handler function for the 'get_coles_products' tool. Decorated with @mcp.tool() for automatic registration in the MCP server. Implements the tool logic by calling helper functions to search and extract products from Coles API, applies limits, formats 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)

  Supporting helper 'search_products' (exposed as coles_search_products via __init__.py) that makes the HTTP request to Coles search API and returns raw response data or error.

  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)

  Supporting helper 'extract_products' (exposed as coles_extract_products) that parses the raw Coles API search response to extract structured product data (name, price, unit).

  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 with schema inferred from signature and docstring.

  @mcp.tool()

• src/supermarkets/coles.py:21-21 (helper)

  Constant COLES_DEFAULT_STORE_ID (aliased from DEFAULT_STORE_ID) used as default store_id parameter in the tool.

  DEFAULT_STORE_ID = "0584"