bulk_add_to_cart
Add multiple groceries to your Kroger cart in one step. Specify product IDs, quantities, and choose pickup or delivery for streamlined shopping.
Instructions
Add multiple items to the user's Kroger cart in a single operation.
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:
items: List of items to add. Each item should have:
- product_id: The product ID or UPC
- quantity: Quantity to add (default: 1)
- modality: PICKUP or DELIVERY (default: PICKUP)
Returns:
Dictionary with results for each item
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| items | Yes |
Implementation Reference
- The primary handler function for the 'bulk_add_to_cart' tool. Decorated with @mcp.tool(), it handles bulk addition to Kroger cart via API and updates local tracking.@mcp.tool() async def bulk_add_to_cart( items: List[Dict[str, Any]], ctx: Context = None ) -> Dict[str, Any]: """ Add multiple items to the user's Kroger cart in a single operation. 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: items: List of items to add. Each item should have: - product_id: The product ID or UPC - quantity: Quantity to add (default: 1) - modality: PICKUP or DELIVERY (default: PICKUP) Returns: Dictionary with results for each item """ try: if ctx: await ctx.info(f"Adding {len(items)} items to cart in bulk") client = get_authenticated_client() # Format items for the API cart_items = [] for item in items: cart_item = { "upc": item["product_id"], "quantity": item.get("quantity", 1), "modality": item.get("modality", "PICKUP") } cart_items.append(cart_item) if ctx: await ctx.info(f"Calling Kroger API to add {len(cart_items)} items") # Add all items to the actual Kroger cart client.cart.add_to_cart(cart_items) if ctx: await ctx.info("Successfully added all items to Kroger cart") # Add all items to local cart tracking for item in items: _add_item_to_local_cart( item["product_id"], item.get("quantity", 1), item.get("modality", "PICKUP") ) if ctx: await ctx.info("All items added to local cart tracking") return { "success": True, "message": f"Successfully added {len(items)} items to cart", "items_added": len(items), "timestamp": datetime.now().isoformat() } except Exception as e: if ctx: await ctx.error(f"Failed to bulk add items to cart: {str(e)}") 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 } else: return { "success": False, "error": f"Failed to add items to cart: {error_message}", "items_attempted": len(items) }
- src/kroger_mcp/server.py:72-78 (registration)Tool registration block in the MCP server setup. The line 'cart_tools.register_tools(mcp)' triggers the definition and registration of the bulk_add_to_cart handler.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 used by bulk_add_to_cart to update local cart tracking for each item added.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)