Kroger MCP Server

Instructions

Implementation Reference

• src/kroger_mcp/tools/product_tools.py:280-396 (handler)

  Core implementation of the get_product_details tool: async function decorated with @mcp.tool() that retrieves comprehensive product information (description, pricing, images, aisle locations, inventory) from the Kroger API for a given product_id and location.
  @mcp.tool() async def get_product_details( product_id: str, location_id: Optional[str] = None, ctx: Context = None ) -> Dict[str, Any]: """ Get detailed information about a specific product. Args: product_id: The unique product identifier location_id: Store location ID for pricing/availability (uses preferred if not provided) Returns: Dictionary containing detailed product information """ # 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"Getting details for product {product_id} at location {location_id}") client = get_client_credentials_client() try: product_details = client.product.get_product( product_id=product_id, location_id=location_id ) if not product_details or "data" not in product_details: return { "success": False, "message": f"Product {product_id} not found" } product = product_details["data"] # Format the detailed product information result = { "success": True, "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", {}), "location_id": location_id } # Add detailed item information if "items" in product and product["items"]: item = product["items"][0] result["item_details"] = { "size": item.get("size"), "sold_by": item.get("soldBy"), "inventory": item.get("inventory", {}), "fulfillment": item.get("fulfillment", {}) } # Add detailed pricing if "price" in item: price = item["price"] result["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')), "savings": price.get("regular", 0) - price.get("promo", price.get("regular", 0)) if price.get("promo") else 0 } # Add aisle locations if "aisleLocations" in product: result["aisle_locations"] = [ { "description": aisle.get("description"), "aisle_number": aisle.get("number"), "side": aisle.get("side"), "shelf_number": aisle.get("shelfNumber") } for aisle in product["aisleLocations"] ] # Add images if "images" in product and product["images"]: result["images"] = [ { "perspective": img.get("perspective"), "sizes": [ { "size": size.get("size"), "url": size.get("url") } for size in img.get("sizes", []) ] } for img in product["images"] ] return result except Exception as e: if ctx: await ctx.error(f"Error getting product details: {str(e)}") return { "success": False, "error": str(e) }
• src/kroger_mcp/server.py:73-73 (registration)

  Registers the product_tools module (containing get_product_details) by calling its register_tools function during server initialization.
  product_tools.register_tools(mcp)
• src/kroger_mcp/tools/product_tools.py:18-18 (registration)

  The register_tools function in product_tools.py where get_product_details is defined and registered via @mcp.tool() decorator.
  def register_tools(mcp):
• src/kroger_mcp/tools/product_tools.py:281-285 (schema)

  Input schema defined by function parameters (product_id required str, location_id optional str) and output Dict[str, Any].
  async def get_product_details( product_id: str, location_id: Optional[str] = None, ctx: Context = None ) -> Dict[str, Any]: