Build a Warehouse MCP Server for Agentic Inventory & Fulfillment

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

In 2026 warehouse operators stopped asking "can AI agents run my WMS?" and started asking "which tools do they get?" The Model Context Protocol turned that question into an integration problem instead of a strategy debate. A warehouse management system (WMS) is a perfect MCP target: it is full of small, well-defined, high-frequency operations — look up an item, check a bin, reserve stock, raise a restock — that consume analyst hours when done by hand and break on typos when done by script. Wrap those operations as governed MCP tools and a Claude-class agent can answer inventory questions, plan a pick run, and flag stockouts in one session.

This dispatch builds warehouse-mcp: a production-ready FastMCP (Python) server exposing six tools — get_item, get_stock_level, pick_order, trigger_restock, get_supplier_status, and low_stock_alerts — that let an agent query live stock and drive fulfillment through your existing WMS API. We cover the full stack: runnable server code, the inputSchema the model actually sees, the mcpServers config for Claude Desktop and other clients, OAuth 2.0 / API-key security for the WMS backend, and the retry, timeout, and idempotency rules warehouse systems demand. For the landscape first, the MCP Directory is the map; this guide is the build.

Why agents belong on the warehouse floor

Warehouse operations are the most error-prone part of e-commerce and the most visible: stockouts kill conversions, double-picks kill margins, and manual cycle counts never match the system of record. Agentic inventory and fulfillment operations fix this by putting the system of record in front of the model. Instead of a human copying SKUs from a spreadsheet into a picking UI, an agent:

• reads live stock levels and answers "can we ship order X today?" without a dashboard lookup;
• builds a pick plan and reserves units in one atomic call;
• raises restock requests the moment available stock crosses a threshold;
• checks supplier POs and ETAs before promising a customer a delivery date.

None of this requires the WMS to be "AI-ready." It requires an MCP surface over the API you already have. That is the whole point of the protocol: the model stays model, the warehouse stays warehouse, and the tool layer carries the business rules.

The six tools

Two design rules keep this surface safe. First, read tools are free and write tools are gated: pick_order and trigger_restock are the only tools that mutate warehouse state, so they carry scopes, confirmation prompts in the client, and idempotency keys. Second, every write must be repeatable: warehouse networks drop packets and operators re-scan, so a re-sent call must never double-pick or double-restock. That is why pick_order takes a picking_key — the same key returns the same result, no matter how many times the agent retries.

Project setup

mkdir warehouse-mcp && cd warehouse-mcp
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

requirements.txt:

fastmcp>=2.0.0
httpx>=0.27.0
pydantic>=2.7.0
python-dotenv>=1.0.0

.env:

WMS_BASE_URL=https://wms.example.com/api/v1
WMS_API_KEY=wms_live_9f2b6c1d0a4e
REQUEST_TIMEOUT=8
MAX_RETRIES=3

Keep the API key out of git — it is a secret, and the section on security explains the full story.

server.py — the complete FastMCP server

# server.py
import os
import time
from typing import Optional

import httpx
from dotenv import load_dotenv
from fastmcp import FastMCP
from pydantic import BaseModel, Field

load_dotenv()

WMS_BASE_URL = os.getenv("WMS_BASE_URL", "https://wms.example.com/api/v1")
WMS_API_KEY = os.getenv("WMS_API_KEY", "")
REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "8.0"))
MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3"))

mcp = FastMCP("warehouse-mcp", version="1.0.0")


class PickItem(BaseModel):
    sku: str = Field(description="SKU of the item to pick")
    quantity: int = Field(ge=1, description="Number of units to pick")
    bin: Optional[str] = Field(default=None, description="Target bin/location")


class _WMSClient:
    """Hardened client for the WMS backend: timeout, retry, circuit breaker."""

    def __init__(self):
        self._breaker_open_until = 0.0
        self._consecutive_failures = 0

    def _headers(self):
        return {
            "Authorization": f"Bearer {WMS_API_KEY}",
            "Content-Type": "application/json",
        }

    def _circuit_open(self) -> bool:
        return time.time() < self._breaker_open_until

    def _record_failure(self):
        self._consecutive_failures += 1
        if self._consecutive_failures >= 5:
            self._breaker_open_until = time.time() + 30.0  # open circuit 30s

    def _record_success(self):
        self._consecutive_failures = 0

    def _request(self, method, path, *, json=None, params=None, idempotency_key=None):
        if self._circuit_open():
            raise RuntimeError("WMS circuit is open; retry later")
        headers = self._headers()
        if idempotency_key:
            headers["Idempotency-Key"] = idempotency_key
        delay = 0.5
        for attempt in range(MAX_RETRIES + 1):
            try:
                resp = httpx.request(
                    method, f"{WMS_BASE_URL}{path}",
                    json=json, params=params, headers=headers,
                    timeout=REQUEST_TIMEOUT,
                )
                if resp.status_code == 429:
                    time.sleep(float(resp.headers.get("Retry-After", delay)))
                    continue
                resp.raise_for_status()
                self._record_success()
                return resp.json()
            except (httpx.HTTPStatusError, httpx.TransportError, httpx.TimeoutException):
                if attempt >= MAX_RETRIES:
                    self._record_failure()
                    raise
                time.sleep(delay)
                delay = min(delay * 2, 8.0)  # exponential backoff


wms = _WMSClient()


@mcp.tool()
def get_item(sku: str) -> dict:
    """Look up a single SKU: name, category, unit, weight, and primary bin."""
    return wms._request("GET", f"/items/{sku}")


@mcp.tool()
def get_stock_level(sku: str) -> dict:
    """Return live stock for a SKU: on-hand, reserved, available, reorder point."""
    return wms._request("GET", f"/items/{sku}/stock")


@mcp.tool()
def pick_order(order_id: str, picking_key: str, items: list[PickItem]) -> dict:
    """Reserve and pick items for an order. Idempotent: re-calling with the
    same picking_key returns the same result and never double-picks."""
    return wms._request(
        "POST",
        f"/orders/{order_id}/pick",
        json={"items": [i.model_dump() for i in items]},
        idempotency_key=picking_key,
    )


@mcp.tool()
def trigger_restock(sku: str, quantity: int, note: str = "") -> dict:
    """Raise a restock/replenishment request for a SKU. Write tool: scope restock.request."""
    return wms._request(
        "POST", "/restocks",
        json={"sku": sku, "quantity": quantity, "note": note},
    )


@mcp.tool()
def get_supplier_status(supplier_id: str) -> dict:
    """Return open purchase orders and ETA for a supplier."""
    return wms._request("GET", f"/suppliers/{supplier_id}/purchase-orders")


@mcp.tool()
def low_stock_alerts(threshold: int = 5, warehouse: str = "main") -> list:
    """List SKUs in a warehouse whose available stock is at or below threshold."""
    return wms._request(
        "GET", "/alerts/low-stock",
        params={"threshold": threshold, "warehouse": warehouse},
    )


if __name__ == "__main__":
    mcp.run(transport="stdio")

FastMCP derives the JSON Schema from the Pydantic model and type hints automatically. This is the inputSchema the model sees for the write path:

{
  "name": "pick_order",
  "description": "Reserve and pick items for an order. Idempotent: re-calling with the same picking_key returns the same result and never double-picks.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "order_id": { "type": "string" },
      "picking_key": { "type": "string", "description": "Idempotency key; same key = same result" },
      "items": {
        "type": "array",
        "items": {
          "type": "object",
          "properties": {
            "sku": { "type": "string" },
            "quantity": { "type": "integer", "minimum": 1 },
            "bin": { "type": "string" }
          },
          "required": ["sku", "quantity"]
        }
      }
    },
    "required": ["order_id", "picking_key", "items"]
  }
}

Registering the server in Claude Desktop / clients

Add the server to claude_desktop_config.json (or your client's mcpServers block):

{
  "mcpServers": {
    "warehouse": {
      "command": "python",
      "args": ["server.py"],
      "cwd": "/opt/warehouse-mcp",
      "env": {
        "WMS_BASE_URL": "https://wms.example.com/api/v1",
        "WMS_API_KEY": "${WAREHOUSE_MCP_API_KEY}"
      }
    }
  }
}

Use a client-side environment variable substitution or a secret manager reference so the key never lives in a plaintext config file. Every client that speaks MCP — Claude Desktop, Cursor, VS Code, n8n — can consume the same stdio server, so your entire agent fleet shares one inventory tool surface.

Security: OAuth 2.0 and API keys for the WMS backend

The MCP server is a lateral-movement point, not a thin wrapper. If the agent is compromised, the damage is bounded by the token it holds. Treat warehouse tokens like production database credentials:

• Least-privilege scopes. Issue a separate MCP service principal for the server with scopes inventory:read, inventory:write, picking:execute, restock:request. Do not reuse the full-admin WMS account. get_item, get_stock_level, get_supplier_status, and low_stock_alerts only need inventory:read; only pick_order and trigger_restock touch the write scopes.
• OAuth 2.0 client-credentials flow. If your WMS supports OAuth 2.0, use the client-credentials grant: the server exchanges client_id + client_secret for a short-lived access token and refreshes it automatically. If it only takes API keys (most older WMS/ERP backends), use a dedicated key, not a human's key.
• Token storage and rotation. Store secrets in the OS keychain, HashiCorp Vault, or AWS Secrets Manager — never in server.py or env files committed to git. Rotate API keys every 90 days and on any suspected leak; rotate OAuth client secrets on the same cadence. One leaked read-only key should not grant picking rights, and one leaked write key should expire before it is useful.
• Never echo secrets to the model. Strip Authorization headers and token values from any tool result or error before it reaches the LLM. A model that can see a token can repeat it into a log, a prompt injection, or a user-visible tool result.
• Audit every write. Log picking_key, order_id, caller, and timestamp for every pick_order and trigger_restock. The idempotency key doubles as the audit correlation ID.

Retries, timeouts, and idempotency for warehouse systems

Warehouse backends are the least reliable systems you will integrate: hand scanners drop network, PLCs lag, and ERPs return 500s for no reproducible reason. The server above bakes in the three rules that keep agents safe:

1. Exponential backoff with jitter. Retry with 0.5s → 1s → 2s → 4s and cap at 8s. Warehouses have bursty peaks at shift changes; a naive retry storm makes the outage worse. Respect Retry-After on 429s.
2. Circuit breaker. After five consecutive failures, open the circuit for 30 seconds and fail fast instead of hammering a dead backend. The agent sees a clean error ("WMS circuit is open; retry later") and can defer the task instead of looping.
3. Idempotency keys on writes. Every pick_order call carries a picking_key. The backend stores the key with the first successful result and returns it for duplicates. This is the difference between "agent retried a dropped request" and "agent double-picked an order."

Timeouts matter as much as retries. An 8-second REQUEST_TIMEOUT means an agent never sits on a stalled pick request for a minute; it fails, reports, and either retries or escalates. On a slow Friday night in India peak season, that behavior difference is the difference between a 99.9% fulfillment SLA and a pile of stuck orders.

Observability and going live

Before production, run fastmcp dev server.py and the MCP Inspector to verify every tool's schema and result shape. Instrument the server with request IDs per call, log tool name + latency + result hash, and hook low_stock_alerts into the same Slack/WhatsApp channel your ops team already watches. Start with read-only tools for a week, then enable pick_order scoped to a single test warehouse, then roll out write scopes warehouse by warehouse. Agents are useful precisely because they act — but warehouse state is the last place you want a surprise. Build the surface, gate the writes, and let the model earn trust one SKU at a time. Then connect the same server to your Workflows hub and watch restock triggers fire end to end.

FAQ

Which SDK should I use, FastMCP Python or the TypeScript SDK? Both are production-ready. Python (FastMCP) is fastest to ship if your WMS integration stack is Python or you already use pandas/pydantic for inventory math; the TypeScript @modelcontextprotocol/sdk is the right call if your agent fleet and tooling are Node-based. The protocol surface, inputSchema, and mcpServers config are identical either way.

Does this replace my existing WMS, ERP, or OMS? No. warehouse-mcp is a read/write tool surface in front of your existing WMS API. The WMS stays the system of record; the server just makes it addressable by AI agents with scoped, idempotent, auditable operations.

How do I prevent an agent from over-picking or creating duplicate restocks? Two mechanisms: picking_key idempotency on pick_order, and scoped OAuth/API keys that grant restock:request only to approved servers. Add client-side confirmation prompts in Claude Desktop for every write tool as a second line of defense.

What is the safest way to test this without touching live inventory? Stand up a sandbox WMS (a test warehouse in your WMS, or a mock HTTP server) and run read-only tools first. Use the MCP Inspector to validate schema, then enable write scopes on the test warehouse only.

How should I handle WMS rate limits at peak hours? Honor Retry-After on 429s, cap concurrency at the server level, and batch low-stock checks into a single scheduled call instead of per-SKU calls. The circuit breaker also prevents the agent from amplifying an outage.