search_products
Search for products at Kroger stores using a term, location, and filters like brand or fulfillment method. Returns results to simplify grocery shopping decisions.
Instructions
Search for products at a Kroger store.
Args:
search_term: Product search term (e.g., "milk", "bread", "organic apples")
location_id: Store location ID (uses preferred location if not provided)
limit: Number of results to return (1-50)
fulfillment: Filter by fulfillment method (csp=curbside pickup, delivery, pickup)
brand: Filter by brand name
Returns:
Dictionary containing product search results
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| brand | No | ||
| fulfillment | No | ||
| limit | No | Number of results to return (1-50) | |
| location_id | No | ||
| search_term | Yes |
Implementation Reference
- The core handler function for the 'search_products' MCP tool. It uses the Kroger client to search products based on term, location, filters, formats detailed results (including pricing, images, aisle locations), and handles errors.@mcp.tool() async def search_products( search_term: str, location_id: Optional[str] = None, limit: int = Field(default=10, ge=1, le=50, description="Number of results to return (1-50)"), fulfillment: Optional[Literal["csp", "delivery", "pickup"]] = None, brand: Optional[str] = None, ctx: Context = None ) -> Dict[str, Any]: """ Search for products at a Kroger store. Args: search_term: Product search term (e.g., "milk", "bread", "organic apples") location_id: Store location ID (uses preferred location if not provided) limit: Number of results to return (1-50) fulfillment: Filter by fulfillment method (csp=curbside pickup, delivery, pickup) brand: Filter by brand name Returns: Dictionary containing product search results """ # Use preferred location if none provided if not location_id: location_id = get_preferred_location_id() if not location_id: return { "success": False, "error": "No location_id provided and no preferred location set. Use set_preferred_location first." } if ctx: await ctx.info(f"Searching for '{search_term}' at location {location_id}") client = get_client_credentials_client() try: products = client.product.search_products( term=search_term, location_id=location_id, limit=limit, fulfillment=fulfillment, brand=brand ) if not products or "data" not in products or not products["data"]: return { "success": False, "message": f"No products found matching '{search_term}'", "data": [] } # Format product data formatted_products = [] for product in products["data"]: formatted_product = { "product_id": product.get("productId"), "upc": product.get("upc"), "description": product.get("description"), "brand": product.get("brand"), "categories": product.get("categories", []), "country_origin": product.get("countryOrigin"), "temperature": product.get("temperature", {}) } # Add item information (size, price, etc.) if "items" in product and product["items"]: item = product["items"][0] formatted_product["item"] = { "size": item.get("size"), "sold_by": item.get("soldBy"), "inventory": item.get("inventory", {}), "fulfillment": item.get("fulfillment", {}) } # Add pricing information if "price" in item: price = item["price"] formatted_product["pricing"] = { "regular_price": price.get("regular"), "sale_price": price.get("promo"), "regular_per_unit": price.get("regularPerUnitEstimate"), "formatted_regular": format_currency(price.get("regular")), "formatted_sale": format_currency(price.get("promo")), "on_sale": price.get("promo") is not None and price.get("promo") < price.get("regular", float('inf')) } # Add aisle information if "aisleLocations" in product: formatted_product["aisle_locations"] = [ { "description": aisle.get("description"), "number": aisle.get("number"), "side": aisle.get("side"), "shelf_number": aisle.get("shelfNumber") } for aisle in product["aisleLocations"] ] # Add image information if "images" in product and product["images"]: formatted_product["images"] = [ { "perspective": img.get("perspective"), "url": img["sizes"][0].get("url") if img.get("sizes") else None, "size": img["sizes"][0].get("size") if img.get("sizes") else None } for img in product["images"] if img.get("sizes") ] formatted_products.append(formatted_product) if ctx: await ctx.info(f"Found {len(formatted_products)} products") return { "success": True, "search_params": { "search_term": search_term, "location_id": location_id, "limit": limit, "fulfillment": fulfillment, "brand": brand }, "count": len(formatted_products), "data": formatted_products } except Exception as e: if ctx: await ctx.error(f"Error searching products: {str(e)}") return { "success": False, "error": str(e), "data": [] }
- src/kroger_mcp/server.py:73-73 (registration)Registers all product tools, including 'search_products', by calling the register_tools function from product_tools module on the MCP server instance.product_tools.register_tools(mcp)
- Input schema defined by function parameters using Pydantic Field and type annotations for validation.search_term: str, location_id: Optional[str] = None, limit: int = Field(default=10, ge=1, le=50, description="Number of results to return (1-50)"), fulfillment: Optional[Literal["csp", "delivery", "pickup"]] = None, brand: Optional[str] = None, ctx: Context = None ) -> Dict[str, Any]: