AI Agent for Warehouse Operations: Inventory Queries, Pick-Pack, and Receiving

AI in the Warehouse

Modern warehouses process thousands of SKUs across receiving, put-away, picking, packing, and shipping. Warehouse associates regularly need to check stock levels, locate items, confirm receipts, and report discrepancies. Traditional WMS interfaces require navigating complex menus and scanning sequences.

An AI warehouse agent provides a natural language interface to the WMS. Associates can ask "where is SKU-4421?" or "did we receive the PO from Acme today?" and get immediate answers. The agent can also guide pick-pack workflows, validate quantities, and escalate exceptions to supervisors.

Warehouse Data Models

from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class InventoryItem:
    sku: str
    name: str
    description: str
    quantity_on_hand: int
    quantity_reserved: int
    location_bin: str
    zone: str
    reorder_point: int
    unit_cost: float

@dataclass
class PurchaseOrder:
    po_number: str
    vendor: str
    expected_date: str
    status: str
    lines: list[dict]

@dataclass
class PickTask:
    task_id: str
    order_id: str
    sku: str
    quantity: int
    bin_location: str
    status: str = "pending"

INVENTORY = {
    "SKU-4421": InventoryItem(
        "SKU-4421", "Wireless Mouse", "Ergonomic wireless mouse 2.4GHz",
        342, 28, "A-12-03", "Zone A", 100, 8.50),
    "SKU-4422": InventoryItem(
        "SKU-4422", "USB-C Hub", "7-port USB-C docking station",
        87, 15, "A-14-01", "Zone A", 50, 22.00),
    "SKU-5510": InventoryItem(
        "SKU-5510", "Laptop Stand", "Adjustable aluminum laptop stand",
        156, 0, "B-03-02", "Zone B", 75, 15.00),
    "SKU-5511": InventoryItem(
        "SKU-5511", "Monitor Arm", "Single monitor desk mount 27 inch",
        23, 10, "B-05-04", "Zone B", 30, 35.00),
    "SKU-6001": InventoryItem(
        "SKU-6001", "Keyboard", "Mechanical keyboard RGB backlit",
        410, 52, "C-01-01", "Zone C", 150, 12.00),
}

Inventory Query Tool

The inventory tool supports lookups by SKU, name search, zone filtering, and low-stock alerts:

sequenceDiagram
    autonumber
    participant Caller as Caller
    participant Agent as CallSphere Agent
    participant API as CRM API
    participant DB as CRM Database
    participant Webhook as Webhook Listener
    Caller->>Agent: Inbound call begins
    Agent->>Agent: STT plus intent detection
    Agent->>API: Lookup contact by phone
    API->>DB: Read contact record
    DB-->>API: Contact and history
    API-->>Agent: Personalized context
    Agent->>API: Create call activity
    Agent->>API: Update deal stage
    API->>Webhook: Outbound webhook fires
    Webhook-->>Agent: Confirmed
    Agent->>Caller: Spoken confirmation

from agents import function_tool

@function_tool
def query_inventory(
    sku: Optional[str] = None,
    search_name: Optional[str] = None,
    zone: Optional[str] = None,
    low_stock_only: bool = False,
) -> str:
    """Query warehouse inventory by SKU, name, zone, or low stock status."""
    items = list(INVENTORY.values())

    if sku:
        item = INVENTORY.get(sku.upper())
        if not item:
            return f"SKU {sku} not found in inventory."
        available = item.quantity_on_hand - item.quantity_reserved
        return (
            f"{item.sku}: {item.name}\n"
            f"On Hand: {item.quantity_on_hand} | Reserved: {item.quantity_reserved} | "
            f"Available: {available}\n"
            f"Location: {item.location_bin} ({item.zone})\n"
            f"Unit Cost: ${item.unit_cost:.2f} | "
            f"Reorder Point: {item.reorder_point}"
        )

    if search_name:
        items = [i for i in items
                 if search_name.lower() in i.name.lower()]
    if zone:
        items = [i for i in items
                 if i.zone.lower() == zone.lower()]
    if low_stock_only:
        items = [i for i in items
                 if (i.quantity_on_hand - i.quantity_reserved) <= i.reorder_point]

    if not items:
        return "No items match your criteria."

    lines = []
    for i in items:
        avail = i.quantity_on_hand - i.quantity_reserved
        flag = " [LOW STOCK]" if avail <= i.reorder_point else ""
        lines.append(
            f"{i.sku}: {i.name} | Avail: {avail} | "
            f"Bin: {i.location_bin}{flag}"
        )
    return "\n".join(lines)

Receiving Tool

When shipments arrive, the agent helps process purchase order receipts:

PURCHASE_ORDERS = {
    "PO-8001": PurchaseOrder(
        "PO-8001", "Acme Electronics", "2026-03-17", "in_transit",
        [
            {"sku": "SKU-4421", "expected_qty": 200, "received_qty": 0},
            {"sku": "SKU-4422", "expected_qty": 100, "received_qty": 0},
        ],
    ),
    "PO-8002": PurchaseOrder(
        "PO-8002", "TechParts Inc", "2026-03-18", "pending",
        [
            {"sku": "SKU-5511", "expected_qty": 50, "received_qty": 0},
        ],
    ),
}

@function_tool
def receive_purchase_order(
    po_number: str,
    sku: str,
    received_quantity: int,
) -> str:
    """Process receiving for a purchase order line item."""
    po = PURCHASE_ORDERS.get(po_number.upper())
    if not po:
        return f"Purchase order {po_number} not found."

    line = next((l for l in po.lines if l["sku"] == sku.upper()), None)
    if not line:
        return f"SKU {sku} not found on {po_number}."

    line["received_qty"] += received_quantity
    variance = line["received_qty"] - line["expected_qty"]

    # Update inventory
    item = INVENTORY.get(sku.upper())
    if item:
        item.quantity_on_hand += received_quantity

    status = "complete" if variance == 0 else ("over" if variance > 0 else "short")
    result = (
        f"Received {received_quantity} units of {sku} on {po_number}\n"
        f"Expected: {line['expected_qty']} | Total Received: {line['received_qty']}\n"
    )
    if variance != 0:
        result += f"VARIANCE: {'+' if variance > 0 else ''}{variance} units ({status})\n"
        result += "Exception reported to supervisor."
    else:
        result += "Receipt complete. No variance."

    return result

Pick Task Management Tool

PICK_TASKS = [
    PickTask("PT-001", "SO-3001", "SKU-4421", 5, "A-12-03"),
    PickTask("PT-002", "SO-3001", "SKU-6001", 3, "C-01-01"),
    PickTask("PT-003", "SO-3002", "SKU-5510", 2, "B-03-02"),
    PickTask("PT-004", "SO-3003", "SKU-4422", 1, "A-14-01"),
]

@function_tool
def get_pick_tasks(order_id: Optional[str] = None, zone: Optional[str] = None) -> str:
    """Get pending pick tasks, optionally filtered by order or zone."""
    tasks = [t for t in PICK_TASKS if t.status == "pending"]

    if order_id:
        tasks = [t for t in tasks if t.order_id == order_id]
    if zone:
        tasks = [t for t in tasks if t.bin_location.startswith(zone[0].upper())]

    if not tasks:
        return "No pending pick tasks match your criteria."

    lines = [f"Pending Pick Tasks ({len(tasks)} total):"]
    for t in tasks:
        item = INVENTORY.get(t.sku)
        name = item.name if item else t.sku
        lines.append(
            f"  {t.task_id} | Order: {t.order_id} | {name} x{t.quantity} | "
            f"Bin: {t.bin_location}"
        )
    return "\n".join(lines)

@function_tool
def confirm_pick(task_id: str, picked_quantity: int) -> str:
    """Confirm a pick task with actual quantity picked."""
    task = next((t for t in PICK_TASKS if t.task_id == task_id), None)
    if not task:
        return f"Pick task {task_id} not found."

    if picked_quantity == task.quantity:
        task.status = "completed"
        return f"Pick {task_id} confirmed: {picked_quantity} units of {task.sku} from {task.bin_location}."

    if picked_quantity < task.quantity:
        short = task.quantity - picked_quantity
        task.status = "short_pick"
        return (
            f"Short pick on {task_id}: expected {task.quantity}, "
            f"got {picked_quantity} (short {short}). "
            f"Exception flagged for supervisor review."
        )

    return f"Picked quantity ({picked_quantity}) exceeds expected ({task.quantity}). Please verify."

Assembling the Warehouse Agent

from agents import Agent, Runner

warehouse_agent = Agent(
    name="Warehouse Assistant",
    instructions="""You are a warehouse operations assistant. Help associates:
    1. Check inventory levels, locations, and low-stock alerts
    2. Process purchase order receipts and flag variances
    3. Manage pick tasks and confirm quantities
    Always report variances and short picks clearly.""",
    tools=[query_inventory, receive_purchase_order, get_pick_tasks, confirm_pick],
)

result = Runner.run_sync(
    warehouse_agent,
    "Show me all low stock items and any pending pick tasks for Zone A."
)
print(result.final_output)

FAQ

How do I integrate with a real WMS like Manhattan, Blue Yonder, or SAP EWM?

Most enterprise WMS platforms expose REST or SOAP APIs for inventory queries, receipt processing, and task management. Replace the in-memory data structures with API calls. Use service accounts with read/write permissions scoped to the operations the agent performs. Implement retry logic for transient API failures.

Can the agent work with barcode scanners?

Yes. Build a thin interface layer that accepts barcode scan input (typically via HTTP POST from a mobile scanner app) and passes the scanned value as a parameter to the appropriate tool. The agent can then confirm the scan matches the expected SKU or bin location and proceed with the workflow.

How do I handle cycle counts and inventory adjustments?

Add a cycle count tool that generates count tasks for specific bins or SKUs. The associate reports the physical count, and the tool compares it against the system quantity. If there is a variance beyond a configurable threshold, the tool creates an adjustment record and flags it for approval.

---

#WarehouseManagement #WMSIntegration #InventoryAI #PickAndPack #Python #AgenticAI #LearnAI #AIEngineering