Inventory Management Tools

Nothing frustrates customers more than ordering a product only to learn it's out of stock. Inventory tools give your AI agent real-time visibility into stock levels and the ability to reserve items during checkout.

Inventory Tool Architecture

┌──────────────────────────────────────────────────────────────────┐
│                     AI AGENT CONVERSATION                        │
│  "Do you have the red jacket in medium?"                        │
└─────────────────────────────────┬────────────────────────────────┘
                                  │
                                  ▼
┌──────────────────────────────────────────────────────────────────┐
│                   INVENTORY MCP TOOLS                            │
├────────────────┬───────────────┬───────────────┬────────────────┤
│ check_inventory│ reserve_stock │ release_stock │ list_warehouses│
│ get_stock_level│ bulk_check    │ low_stock_alert│ update_quantity│
└────────────────┴───────────────┴───────────────┴────────────────┘
                                  │
                                  ▼
┌──────────────────────────────────────────────────────────────────┐
│                   INVENTORY DATABASE                             │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐              │
│  │  US-EAST    │  │  US-WEST    │  │  EU-CENTRAL │              │
│  │  Warehouse  │  │  Warehouse  │  │  Warehouse  │              │
│  └─────────────┘  └─────────────┘  └─────────────┘              │
└──────────────────────────────────────────────────────────────────┘

Core Inventory Tools

1. Check Inventory Tool

The most frequently used tool - check if products are available:

from mcp.server import Server
from mcp import types
from datetime import datetime, timedelta
import json

app = Server("inventory-tools")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="check_inventory",
            description="""Checks real-time inventory levels for one or more products.
            Use when: Customer asks about availability, before adding to cart, or during checkout.
            Returns: Available quantity, reserved quantity, and warehouse locations.
            Note: Supports checking multiple SKUs in one call for efficiency.""",
            inputSchema={
                "type": "object",
                "properties": {
                    "product_ids": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "Product SKUs to check (e.g., ['JACKET-RED-M', 'JACKET-RED-L'])",
                        "minItems": 1,
                        "maxItems": 50
                    },
                    "warehouse_id": {
                        "type": "string",
                        "description": "Specific warehouse to check. Omit for all warehouses.",
                        "enum": ["US-EAST", "US-WEST", "EU-CENTRAL", "APAC"]
                    },
                    "include_incoming": {
                        "type": "boolean",
                        "default": False,
                        "description": "Include expected incoming shipments in response"
                    }
                },
                "required": ["product_ids"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "check_inventory":
        product_ids = arguments["product_ids"]
        warehouse_id = arguments.get("warehouse_id")
        include_incoming = arguments.get("include_incoming", False)

        results = {}
        for product_id in product_ids:
            inventory = await inventory_db.get_stock(
                product_id,
                warehouse_id
            )

            product_result = {
                "product_id": product_id,
                "total_available": 0,
                "warehouses": []
            }

            for warehouse, stock in inventory.items():
                warehouse_data = {
                    "warehouse_id": warehouse,
                    "available": stock["available"],
                    "reserved": stock["reserved"],
                    "on_hand": stock["available"] + stock["reserved"]
                }

                if include_incoming and stock.get("incoming"):
                    warehouse_data["incoming"] = {
                        "quantity": stock["incoming"]["quantity"],
                        "expected_date": stock["incoming"]["date"].isoformat()
                    }

                product_result["warehouses"].append(warehouse_data)
                product_result["total_available"] += stock["available"]

            # Add status for quick reference
            if product_result["total_available"] == 0:
                product_result["status"] = "out_of_stock"
            elif product_result["total_available"] < 5:
                product_result["status"] = "low_stock"
            else:
                product_result["status"] = "in_stock"

            results[product_id] = product_result

        return [types.TextContent(
            type="text",
            text=json.dumps({"inventory": results})
        )]

2. Reserve Stock Tool

Reserve inventory during checkout to prevent overselling:

types.Tool(
    name="reserve_stock",
    description="""Temporarily reserves inventory for a customer during checkout.
    Use when: Customer proceeds to checkout, before payment processing.
    Returns: Reservation ID and expiration time.
    Note: Reservations auto-expire after 15 minutes if not converted to order.""",
    inputSchema={
        "type": "object",
        "properties": {
            "items": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "product_id": {"type": "string"},
                        "quantity": {"type": "integer", "minimum": 1}
                    },
                    "required": ["product_id", "quantity"]
                },
                "description": "Items to reserve",
                "minItems": 1
            },
            "customer_id": {
                "type": "string",
                "description": "Customer ID for the reservation"
            },
            "cart_id": {
                "type": "string",
                "description": "Shopping cart ID to link reservation"
            },
            "ttl_minutes": {
                "type": "integer",
                "default": 15,
                "minimum": 5,
                "maximum": 60,
                "description": "Minutes before reservation expires"
            },
            "warehouse_preference": {
                "type": "string",
                "enum": ["nearest", "fastest", "cheapest"],
                "default": "nearest",
                "description": "How to select fulfillment warehouse"
            }
        },
        "required": ["items", "customer_id"]
    }
)

elif name == "reserve_stock":
    items = arguments["items"]
    customer_id = arguments["customer_id"]
    cart_id = arguments.get("cart_id")
    ttl_minutes = arguments.get("ttl_minutes", 15)

    # Check availability first
    unavailable = []
    for item in items:
        stock = await inventory_db.get_available(item["product_id"])
        if stock < item["quantity"]:
            unavailable.append({
                "product_id": item["product_id"],
                "requested": item["quantity"],
                "available": stock
            })

    if unavailable:
        return [types.TextContent(
            type="text",
            text=json.dumps({
                "success": False,
                "error": "insufficient_stock",
                "unavailable_items": unavailable
            })
        )]

    # Create reservation
    reservation_id = f"res_{uuid.uuid4().hex[:12]}"
    expires_at = datetime.utcnow() + timedelta(minutes=ttl_minutes)

    reserved_items = []
    for item in items:
        # Select optimal warehouse
        warehouse = await select_warehouse(
            item["product_id"],
            item["quantity"],
            arguments.get("warehouse_preference", "nearest")
        )

        # Reserve the stock
        await inventory_db.reserve(
            product_id=item["product_id"],
            quantity=item["quantity"],
            warehouse_id=warehouse,
            reservation_id=reservation_id,
            expires_at=expires_at
        )

        reserved_items.append({
            "product_id": item["product_id"],
            "quantity": item["quantity"],
            "warehouse_id": warehouse
        })

    return [types.TextContent(
        type="text",
        text=json.dumps({
            "success": True,
            "reservation_id": reservation_id,
            "customer_id": customer_id,
            "cart_id": cart_id,
            "items": reserved_items,
            "expires_at": expires_at.isoformat(),
            "ttl_seconds": ttl_minutes * 60
        })
    )]

3. Release Stock Tool

Release reserved stock if checkout is abandoned or fails:

types.Tool(
    name="release_stock",
    description="""Releases previously reserved inventory back to available stock.
    Use when: Payment fails, customer abandons cart, or order is cancelled.
    Returns: Confirmation of released items.
    Note: Partial releases are supported for specific items.""",
    inputSchema={
        "type": "object",
        "properties": {
            "reservation_id": {
                "type": "string",
                "description": "The reservation ID to release"
            },
            "items": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "product_id": {"type": "string"},
                        "quantity": {"type": "integer", "minimum": 1}
                    }
                },
                "description": "Specific items to release. Omit to release entire reservation."
            },
            "reason": {
                "type": "string",
                "enum": ["payment_failed", "cart_abandoned", "customer_cancelled", "order_modified"],
                "description": "Reason for releasing the reservation"
            }
        },
        "required": ["reservation_id"]
    }
)

4. Low Stock Alert Tool

Monitor inventory levels and trigger alerts:

types.Tool(
    name="get_low_stock_products",
    description="""Returns products that are below their reorder threshold.
    Use when: Reviewing inventory status, planning restocking, or checking for potential stockouts.
    Returns: List of products with current levels and reorder recommendations.""",
    inputSchema={
        "type": "object",
        "properties": {
            "warehouse_id": {
                "type": "string",
                "enum": ["US-EAST", "US-WEST", "EU-CENTRAL", "APAC", "ALL"],
                "default": "ALL",
                "description": "Warehouse to check"
            },
            "category": {
                "type": "string",
                "description": "Product category to filter (e.g., 'clothing', 'electronics')"
            },
            "threshold_type": {
                "type": "string",
                "enum": ["critical", "warning", "custom"],
                "default": "warning",
                "description": "Stock level threshold type"
            },
            "custom_threshold": {
                "type": "integer",
                "minimum": 1,
                "description": "Custom threshold quantity (required if threshold_type is 'custom')"
            },
            "limit": {
                "type": "integer",
                "default": 50,
                "maximum": 200,
                "description": "Maximum products to return"
            }
        }
    }
)

TypeScript Implementation

import { Server } from "@modelcontextprotocol/sdk/server";
import { v4 as uuidv4 } from "uuid";

const server = new Server({ name: "inventory-tools", version: "1.0.0" });

interface InventoryItem {
  product_id: string;
  available: number;
  reserved: number;
  warehouse_id: string;
}

interface Reservation {
  id: string;
  items: Array<{
    product_id: string;
    quantity: number;
    warehouse_id: string;
  }>;
  customer_id: string;
  expires_at: Date;
}

const inventoryTools = [
  {
    name: "check_inventory",
    description: "Checks real-time inventory levels for products across warehouses",
    inputSchema: {
      type: "object" as const,
      properties: {
        product_ids: {
          type: "array",
          items: { type: "string" },
          minItems: 1,
          maxItems: 50
        },
        warehouse_id: {
          type: "string",
          enum: ["US-EAST", "US-WEST", "EU-CENTRAL", "APAC"]
        }
      },
      required: ["product_ids"]
    }
  },
  {
    name: "reserve_stock",
    description: "Temporarily reserves inventory during checkout to prevent overselling",
    inputSchema: {
      type: "object" as const,
      properties: {
        items: {
          type: "array",
          items: {
            type: "object",
            properties: {
              product_id: { type: "string" },
              quantity: { type: "integer", minimum: 1 }
            },
            required: ["product_id", "quantity"]
          },
          minItems: 1
        },
        customer_id: { type: "string" },
        ttl_minutes: { type: "integer", default: 15, minimum: 5, maximum: 60 }
      },
      required: ["items", "customer_id"]
    }
  },
  {
    name: "release_stock",
    description: "Releases reserved inventory back to available stock",
    inputSchema: {
      type: "object" as const,
      properties: {
        reservation_id: { type: "string" },
        reason: {
          type: "string",
          enum: ["payment_failed", "cart_abandoned", "customer_cancelled"]
        }
      },
      required: ["reservation_id"]
    }
  }
];

server.setRequestHandler("tools/list", async () => ({ tools: inventoryTools }));

server.setRequestHandler("tools/call", async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case "check_inventory": {
      const results: Record<string, any> = {};

      for (const productId of args.product_ids) {
        const inventory = await getInventory(productId, args.warehouse_id);

        let totalAvailable = 0;
        const warehouses = inventory.map((item: InventoryItem) => {
          totalAvailable += item.available;
          return {
            warehouse_id: item.warehouse_id,
            available: item.available,
            reserved: item.reserved,
            on_hand: item.available + item.reserved
          };
        });

        results[productId] = {
          product_id: productId,
          total_available: totalAvailable,
          status: totalAvailable === 0 ? "out_of_stock" :
                  totalAvailable < 5 ? "low_stock" : "in_stock",
          warehouses
        };
      }

      return {
        content: [{
          type: "text",
          text: JSON.stringify({ inventory: results })
        }]
      };
    }

    case "reserve_stock": {
      // Check availability
      const unavailable: any[] = [];
      for (const item of args.items) {
        const stock = await getAvailableQuantity(item.product_id);
        if (stock < item.quantity) {
          unavailable.push({
            product_id: item.product_id,
            requested: item.quantity,
            available: stock
          });
        }
      }

      if (unavailable.length > 0) {
        return {
          content: [{
            type: "text",
            text: JSON.stringify({
              success: false,
              error: "insufficient_stock",
              unavailable_items: unavailable
            })
          }]
        };
      }

      // Create reservation
      const reservationId = `res_${uuidv4().slice(0, 12)}`;
      const ttlMinutes = args.ttl_minutes || 15;
      const expiresAt = new Date(Date.now() + ttlMinutes * 60 * 1000);

      const reservedItems = [];
      for (const item of args.items) {
        const warehouse = await selectOptimalWarehouse(item.product_id);
        await createReservation(reservationId, item, warehouse, expiresAt);

        reservedItems.push({
          product_id: item.product_id,
          quantity: item.quantity,
          warehouse_id: warehouse
        });
      }

      return {
        content: [{
          type: "text",
          text: JSON.stringify({
            success: true,
            reservation_id: reservationId,
            customer_id: args.customer_id,
            items: reservedItems,
            expires_at: expiresAt.toISOString(),
            ttl_seconds: ttlMinutes * 60
          })
        }]
      };
    }

    case "release_stock": {
      const released = await releaseReservation(
        args.reservation_id,
        args.reason
      );

      return {
        content: [{
          type: "text",
          text: JSON.stringify({
            success: true,
            reservation_id: args.reservation_id,
            released_items: released,
            reason: args.reason
          })
        }]
      };
    }

    default:
      throw new Error(`Unknown tool: ${name}`);
  }
});

Inventory Data Model

-- Products table
CREATE TABLE products (
    id VARCHAR(50) PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    category VARCHAR(100),
    reorder_threshold INT DEFAULT 10,
    reorder_quantity INT DEFAULT 100
);

-- Inventory per warehouse
CREATE TABLE inventory (
    product_id VARCHAR(50) REFERENCES products(id),
    warehouse_id VARCHAR(20) NOT NULL,
    available INT NOT NULL DEFAULT 0,
    reserved INT NOT NULL DEFAULT 0,
    last_updated TIMESTAMP DEFAULT NOW(),
    PRIMARY KEY (product_id, warehouse_id)
);

-- Reservations with TTL
CREATE TABLE reservations (
    id VARCHAR(50) PRIMARY KEY,
    customer_id VARCHAR(50) NOT NULL,
    cart_id VARCHAR(50),
    status VARCHAR(20) DEFAULT 'active',
    created_at TIMESTAMP DEFAULT NOW(),
    expires_at TIMESTAMP NOT NULL
);

-- Reservation items
CREATE TABLE reservation_items (
    reservation_id VARCHAR(50) REFERENCES reservations(id),
    product_id VARCHAR(50),
    warehouse_id VARCHAR(20),
    quantity INT NOT NULL,
    PRIMARY KEY (reservation_id, product_id, warehouse_id)
);

-- Index for expired reservation cleanup
CREATE INDEX idx_reservations_expires ON reservations(expires_at)
WHERE status = 'active';

Handling Race Conditions

Solution: Optimistic Locking with Version Numbers

async def reserve_stock_safely(product_id: str, quantity: int, warehouse_id: str):
    # Use database transaction with row-level locking
    async with db.transaction():
        # Lock the row for update
        row = await db.fetchone("""
            SELECT available, version
            FROM inventory
            WHERE product_id = $1 AND warehouse_id = $2
            FOR UPDATE
        """, product_id, warehouse_id)

        if row["available"] < quantity:
            raise InsufficientStockError(
                f"Only {row['available']} available"
            )

        # Update with version check
        result = await db.execute("""
            UPDATE inventory
            SET available = available - $1,
                reserved = reserved + $1,
                version = version + 1
            WHERE product_id = $2
              AND warehouse_id = $3
              AND version = $4
        """, quantity, product_id, warehouse_id, row["version"])

        if result.rowcount == 0:
            raise ConcurrencyError("Inventory was modified, please retry")

        return True

Real-World Conversation Flow

Customer: "Do you have the wireless headphones in stock?"

AI Agent: [Calls check_inventory]
  check_inventory(product_ids=["HEADPHONES-WL-BLK"])

Tool Response:
{
  "inventory": {
    "HEADPHONES-WL-BLK": {
      "total_available": 45,
      "status": "in_stock",
      "warehouses": [
        {"warehouse_id": "US-EAST", "available": 30},
        {"warehouse_id": "US-WEST", "available": 15}
      ]
    }
  }
}

AI Agent: "Yes! The wireless headphones are in stock. We have 45 units
available - 30 in our East Coast warehouse and 15 on the West Coast.
Would you like to add them to your cart?"

Customer: "Yes, add 2 to my cart"

AI Agent: [Calls reserve_stock]
  reserve_stock(
    items=[{"product_id": "HEADPHONES-WL-BLK", "quantity": 2}],
    customer_id="cus_12345",
    cart_id="cart_67890"
  )

Tool Response:
{
  "success": true,
  "reservation_id": "res_abc123def",
  "items": [{"product_id": "HEADPHONES-WL-BLK", "quantity": 2, "warehouse_id": "US-EAST"}],
  "expires_at": "2025-01-15T10:30:00Z",
  "ttl_seconds": 900
}

AI Agent: "I've added 2 wireless headphones to your cart and reserved them
for the next 15 minutes. Your total is $199.98. Ready to checkout?"

Key Takeaways

Next up: Shipping & Fulfillment Tools - Building tools to calculate shipping rates, generate labels, and track packages.