search_products
Find grocery items at Kroger stores by searching product names, filtering by brand or fulfillment method, and specifying store location.
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 |
|---|---|---|---|
| search_term | Yes | ||
| location_id | No | ||
| limit | No | Number of results to return (1-50) | |
| fulfillment | No | ||
| brand | No |
Implementation Reference
- The core handler function decorated with @mcp.tool() that implements the search_products tool logic: validates inputs, gets Kroger client, calls API, formats results (products with pricing, inventory, aisles, images), 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": [] } - Pydantic-based input schema defined by function parameters, including search_term (required), optional filters like limit, fulfillment, brand, with validation via Field.
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]: - 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.
product_tools.register_tools(mcp) - src/kroger_mcp/server.py:24-24 (registration)Imports the product_tools module which contains the register_tools function.
from .tools import product_tools - Uses helper get_client_credentials_client() and calls the underlying Kroger SDK client.product.search_products() API method.
client = get_client_credentials_client() try: products = client.product.search_products( term=search_term, location_id=location_id, limit=limit, fulfillment=fulfillment, brand=brand )