Tutorial 04
Build a low stock reorder bot
Find items running out, ask Stockpilot for purchase order recommendations, and raise the PO. Covers the async task pattern and the heavy rate limit bucket.
Stockpilot already knows your sales velocity, lead times and inbound stock. Rather than reimplementing that maths, you can ask it for recommendations and turn the answer into a purchase order.
Find what is running low
Start with the inventory list. Thresholds are expressed as either units (5u) or weeks of cover (33w), so an item is “low” relative to its own threshold rather than a global number.
def low_stock(client, page_size=1000):
page, out = 1, []
while True:
data = client.get("/inventory", params={"page": page, "page_size": page_size}).json()
out.extend(i for i in data["results"] if i["quantity"] <= threshold_units(i))
if len(data["results"]) < page_size:
return out
page += 1
One paginated call at page_size=1000 costs a single read-list request. Looping \/inventory\/get per SKU would cost hundreds of read-single requests for the same data, so prefer the list.
List your suppliers
GET /purchase-orders/suppliers/listcurl -s https://api.stockpilot.dev/purchase-orders/suppliers/list \
-H "X-CLIENT-ID: $SP_CLIENT_ID" \
-H "X-CLIENT-SECRET: $SP_CLIENT_SECRET"
Recommendations are generated per supplier, because lead time and minimum order quantities are supplier properties.
Ask for recommendations
This is a long running job, so the API splits it into “start” and “poll”.
POST /purchase-orders/recommendationscurl -s -X POST https://api.stockpilot.dev/purchase-orders/recommendations \
-H "X-CLIENT-ID: $SP_CLIENT_ID" \
-H "X-CLIENT-SECRET: $SP_CLIENT_SECRET" \
-H "Content-Type: application/json" \
-d '{
"supplier_id": 312,
"lead_time": 21,
"durability": 60,
"scope_days": 90,
"include_flagged": true,
"include_inbound": true
}'
| Field | Meaning |
|---|---|
lead_time | Days between raising the PO and stock landing |
durability | Days of cover the resulting order should provide |
scope_days | How far back to look when measuring demand |
include_flagged | Include items flagged for review |
include_inbound | Subtract stock already on its way |
You get a task_id back, not results.
Poll the task
GET /purchase-orders/recommendations/status/{task_id}import time
def wait_for_recommendations(client, task_id, timeout=180):
deadline, delay = time.time() + timeout, 2.0
while time.time() < deadline:
data = client.get(f"/purchase-orders/recommendations/status/{task_id}").json()
if data["status"] == "completed":
return data["results"]
if data["status"] == "failed":
raise RuntimeError(data.get("error", "recommendation task failed"))
time.sleep(delay)
delay = min(delay * 1.5, 15.0)
raise TimeoutError(task_id)
Both \/purchase-orders\/recommendations and the analytics endpoints sit in the heavy bucket: 20 requests per minute. Backing off between polls is not politeness, it is the difference between one integration and twenty of them coexisting on the same key. See rate limits.
Raise the purchase order
POST /purchase-orderslines = [
{"product_id": r["product_id"], "quantity": r["recommended_quantity"]}
for r in recommendations
if r["recommended_quantity"] > 0
]
po = client.post("/purchase-orders", json={
"supplier_id": 312,
"expected_date": "2026-10-01",
"items": lines,
}).json()
print("raised PO", po["id"], "with", len(lines), "lines")
Creating a purchase order is not idempotent. If the call times out, do not resend it. Poll GET \/purchase-orders first and check whether the PO already landed, then decide.
Receive the delivery
When the goods arrive, book them in as parcels against the PO:
POST /purchase-orders/{order_id}/parcels/createcurl -s -X POST https://api.stockpilot.dev/purchase-orders/$PO_ID/parcels/create \
-H "X-CLIENT-ID: $SP_CLIENT_ID" \
-H "X-CLIENT-SECRET: $SP_CLIENT_SECRET" \
-H "Content-Type: application/json" \
-d '{"items": [{"product_id": 90211, "quantity": 120}]}'
Booked a parcel by mistake? An unprocessed parcel can be removed with POST /purchase-orders/{order_id}/parcels/{parcel_id}/delete .
Running it on a schedule
The whole thing is a cron job:
def nightly_replenishment(client, suppliers):
for supplier_id in suppliers:
task = client.post("/purchase-orders/recommendations", json={
"supplier_id": supplier_id,
"lead_time": 21,
"durability": 60,
"scope_days": 90,
"include_flagged": True,
"include_inbound": True,
}).json()
recs = wait_for_recommendations(client, task["task_id"])
lines = [
{"product_id": r["product_id"], "quantity": r["recommended_quantity"]}
for r in recs if r["recommended_quantity"] > 0
]
if not lines:
continue
draft = client.post("/purchase-orders", json={
"supplier_id": supplier_id,
"items": lines,
}).json()
notify_buyer(draft)
Send the draft to a human rather than committing it automatically. Recommendations are good; a buyer who knows a supplier is about to change their minimum order quantity is better.
