add_items_to_cart
Add grocery items to your Kroger cart for pickup or delivery by specifying product IDs and quantities.
Instructions
Add a single item to the user's Kroger cart and track it locally.
If the user doesn't specifically indicate a preference for pickup or delivery,
you should ask them which modality they prefer before calling this tool.
Args:
product_id: The product ID or UPC to add to cart
quantity: Quantity to add (default: 1)
modality: Fulfillment method - PICKUP or DELIVERY
Returns:
Dictionary confirming the item was added to cart
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| product_id | Yes | ||
| quantity | No | ||
| modality | No | PICKUP |
Implementation Reference
- Core handler for the 'add_items_to_cart' tool. Handles API call to Kroger cart.add_to_cart and local tracking update. Includes input validation via types and comprehensive error handling.
@mcp.tool() async def add_items_to_cart( product_id: str, quantity: int = 1, modality: str = "PICKUP", ctx: Context = None ) -> Dict[str, Any]: """ Add a single item to the user's Kroger cart and track it locally. If the user doesn't specifically indicate a preference for pickup or delivery, you should ask them which modality they prefer before calling this tool. Args: product_id: The product ID or UPC to add to cart quantity: Quantity to add (default: 1) modality: Fulfillment method - PICKUP or DELIVERY Returns: Dictionary confirming the item was added to cart """ try: if ctx: await ctx.info(f"Adding {quantity}x {product_id} to cart with {modality} modality") # Get authenticated client client = get_authenticated_client() # Format the item for the API cart_item = { "upc": product_id, "quantity": quantity, "modality": modality } if ctx: await ctx.info(f"Calling Kroger API to add item: {cart_item}") # Add the item to the actual Kroger cart # Note: add_to_cart returns None on success, raises exception on failure client.cart.add_to_cart([cart_item]) if ctx: await ctx.info("Successfully added item to Kroger cart") # Add to local cart tracking _add_item_to_local_cart(product_id, quantity, modality) if ctx: await ctx.info("Item added to local cart tracking") return { "success": True, "message": f"Successfully added {quantity}x {product_id} to cart", "product_id": product_id, "quantity": quantity, "modality": modality, "timestamp": datetime.now().isoformat() } except Exception as e: if ctx: await ctx.error(f"Failed to add item to cart: {str(e)}") # Provide helpful error message for authentication issues error_message = str(e) if "401" in error_message or "Unauthorized" in error_message: return { "success": False, "error": "Authentication failed. Please run force_reauthenticate and try again.", "details": error_message } elif "400" in error_message or "Bad Request" in error_message: return { "success": False, "error": f"Invalid request. Please check the product ID and try again.", "details": error_message } else: return { "success": False, "error": f"Failed to add item to cart: {error_message}", "product_id": product_id, "quantity": quantity, "modality": modality } - src/kroger_mcp/server.py:71-78 (registration)Registration point where cart_tools.register_tools(mcp) is called, which defines and registers the add_items_to_cart tool using @mcp.tool() decorator.
# Register all tools from the modules location_tools.register_tools(mcp) product_tools.register_tools(mcp) cart_tools.register_tools(mcp) info_tools.register_tools(mcp) profile_tools.register_tools(mcp) utility_tools.register_tools(mcp) auth_tools.register_tools(mcp) - Helper function called by the handler to persist cart items to local JSON file (kroger_cart.json), merging quantities for duplicate product/modality.
def _add_item_to_local_cart(product_id: str, quantity: int, modality: str, product_details: Dict[str, Any] = None) -> None: """Add an item to the local cart tracking""" cart_data = _load_cart_data() current_cart = cart_data.get("current_cart", []) # Check if item already exists in cart existing_item = None for item in current_cart: if item.get("product_id") == product_id and item.get("modality") == modality: existing_item = item break if existing_item: # Update existing item quantity existing_item["quantity"] = existing_item.get("quantity", 0) + quantity existing_item["last_updated"] = datetime.now().isoformat() else: # Add new item new_item = { "product_id": product_id, "quantity": quantity, "modality": modality, "added_at": datetime.now().isoformat(), "last_updated": datetime.now().isoformat() } # Add product details if provided if product_details: new_item.update(product_details) current_cart.append(new_item) cart_data["current_cart"] = current_cart cart_data["last_updated"] = datetime.now().isoformat() _save_cart_data(cart_data) - Input schema from function signature and docstring: product_id (str, required), quantity (int, default 1), modality (str, default PICKUP), returns success dict.
async def add_items_to_cart( product_id: str, quantity: int = 1, modality: str = "PICKUP", ctx: Context = None