AI Agent Integration for Shopify Inventory Management Across APAC

1import shopify
2import os
3
4shopify.Session.setup(
5    api_key=os.environ["SHOPIFY_API_KEY"],
6    secret=os.environ["SHOPIFY_API_SECRET"]
7)
8
9session = shopify.Session(
10    os.environ["SHOPIFY_STORE_URL"],
11    "2024-10",
12    os.environ["SHOPIFY_ACCESS_TOKEN"]
13)
14shopify.ShopifyResource.activate_session(session)
15
16locations = shopify.Location.find()
17for loc in locations:
18    print(f"ID: {loc.id} | Name: {loc.name} | Country: {loc.country_code}")

1ID: 61234567890 | Name: HK Central Warehouse | Country: HK
2ID: 61234567891 | Name: SG Jurong 3PL | Country: SG
3ID: 61234567892 | Name: TW Taoyuan Bonded | Country: TW
4ID: 61234567893 | Name: AU Melbourne DC | Country: AU

1# config/apac_locations.yaml
2locations:
3  - shopify_id: 61234567890
4    code: HK_CENTRAL
5    country: HK
6    lead_time_days: 1
7    supports_bonded: false
8    primary_markets: [HK, MO]
9    restock_source: CN_SHENZHEN
10    min_safety_stock: 50
11
12  - shopify_id: 61234567891
13    code: SG_JURONG
14    country: SG
15    lead_time_days: 2
16    supports_bonded: false
17    primary_markets: [SG, MY, ID]
18    restock_source: HK_CENTRAL
19    min_safety_stock: 30
20
21  - shopify_id: 61234567892
22    code: TW_TAOYUAN
23    country: TW
24    lead_time_days: 3
25    supports_bonded: true
26    primary_markets: [TW]
27    restock_source: CN_SHENZHEN
28    min_safety_stock: 20
29
30  - shopify_id: 61234567893
31    code: AU_MELBOURNE
32    country: AU
33    lead_time_days: 5
34    supports_bonded: false
35    primary_markets: [AU, NZ]
36    restock_source: HK_CENTRAL
37    min_safety_stock: 40

1# docker-compose.kafka.yml
2version: '3.8'
3services:
4  zookeeper:
5    image: confluentinc/cp-zookeeper:7.6.0
6    environment:
7      ZOOKEEPER_CLIENT_PORT: 2181
8
9  kafka:
10    image: confluentinc/cp-kafka:7.6.0
11    depends_on: [zookeeper]
12    ports:
13      - "9092:9092"
14    environment:
15      KAFKA_BROKER_ID: 1
16      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
17      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
18      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

1docker exec -it kafka kafka-topics --create \
2  --bootstrap-server localhost:9092 \
3  --topic shopify.inventory.updates \
4  --partitions 4 \
5  --replication-factor 1
6
7docker exec -it kafka kafka-topics --create \
8  --bootstrap-server localhost:9092 \
9  --topic agent.restock.decisions \
10  --partitions 4 \
11  --replication-factor 1

1# webhook_receiver.py
2from flask import Flask, request, jsonify
3from confluent_kafka import Producer
4import json
5import hmac
6import hashlib
7import os
8
9app = Flask(__name__)
10producer = Producer({"bootstrap.servers": "localhost:9092"})
11
12def verify_shopify_webhook(data, hmac_header):
13    secret = os.environ["SHOPIFY_WEBHOOK_SECRET"].encode("utf-8")
14    computed = hmac.new(secret, data, hashlib.sha256).hexdigest()
15    return hmac.compare_digest(computed, hmac_header)
16
17@app.route("/webhooks/inventory", methods=["POST"])
18def handle_inventory_webhook():
19    hmac_header = request.headers.get("X-Shopify-Hmac-Sha256", "")
20    if not verify_shopify_webhook(request.data, hmac_header):
21        return jsonify({"error": "unauthorized"}), 401
22
23    payload = request.json
24    event = {
25        "inventory_item_id": payload["inventory_item_id"],
26        "location_id": payload["location_id"],
27        "available": payload["available"],
28        "updated_at": payload["updated_at"]
29    }
30
31    producer.produce(
32        "shopify.inventory.updates",
33        key=str(payload["inventory_item_id"]),
34        value=json.dumps(event)
35    )
36    producer.flush()
37    return jsonify({"status": "received"}), 200
38
39if __name__ == "__main__":
40    app.run(port=5000)

1curl -X POST "https://{store}.myshopify.com/admin/api/2024-10/webhooks.json" \
2  -H "X-Shopify-Access-Token: ${SHOPIFY_ACCESS_TOKEN}" \
3  -H "Content-Type: application/json" \
4  -d '{
5    "webhook": {
6      "topic": "inventory_levels/update",
7      "address": "https://your-server.com/webhooks/inventory",
8      "format": "json"
9    }
10  }'

1# agent/tools.py
2from langchain.tools import tool
3import shopify
4import yaml
5import json
6
7with open("config/apac_locations.yaml") as f:
8    LOCATIONS = yaml.safe_load(f)["locations"]
9
10@tool
11def get_inventory_levels(variant_id: int) -> str:
12    """Get current inventory levels for a product variant across all APAC locations."""
13    inventory_item = shopify.InventoryItem.find(
14        from_="/admin/api/2024-10/variants/{}.json".format(variant_id)
15    )
16    levels = shopify.InventoryLevel.find(
17        inventory_item_ids=str(inventory_item.id)
18    )
19    result = []
20    for level in levels:
21        loc_meta = next(
22            (l for l in LOCATIONS if l["shopify_id"] == level.location_id), None
23        )
24        result.append({
25            "location": loc_meta["code"] if loc_meta else "UNKNOWN",
26            "available": level.available,
27            "safety_stock": loc_meta["min_safety_stock"] if loc_meta else 0
28        })
29    return json.dumps(result, indent=2)
30
31@tool
32def adjust_inventory(location_id: int, inventory_item_id: int, adjustment: int) -> str:
33    """Adjust inventory at a specific location. Positive = add stock, negative = remove."""
34    result = shopify.InventoryLevel.adjust(
35        location_id=location_id,
36        inventory_item_id=inventory_item_id,
37        available_adjustment=adjustment
38    )
39    return f"Adjusted by {adjustment} at location {location_id}. New level: {result.available}"
40
41@tool
42def get_demand_forecast(variant_id: int, location_code: str, days_ahead: int = 14) -> str:
43    """Query the demand forecast model for a variant at a specific APAC location."""
44    # In production, this calls your ML forecast service
45    # Here we show the interface contract
46    import requests
47    resp = requests.post("http://forecast-service:8080/predict", json={
48        "variant_id": variant_id,
49        "location": location_code,
50        "horizon_days": days_ahead
51    })
52    return resp.text

1# agent/inventory_agent.py
2from langchain_openai import ChatOpenAI
3from langchain.agents import AgentExecutor, create_tool_calling_agent
4from langchain_core.prompts import ChatPromptTemplate
5from agent.tools import get_inventory_levels, adjust_inventory, get_demand_forecast
6
7llm = ChatOpenAI(model="gpt-4o", temperature=0)
8
9prompt = ChatPromptTemplate.from_messages([
10    ("system", """You are an inventory management agent for a Shopify Plus store 
11    operating across Asia-Pacific. You manage stock across Hong Kong, Singapore, 
12    Taiwan, and Australia warehouses.
13    
14    Rules:
15    - Never let any location drop below its safety stock level
16    - Factor in lead times: HK=1d, SG=2d, TW=3d, AU=5d
17    - Prioritise transfers from nearest warehouse before reordering from supplier
18    - Log every decision with reasoning for audit compliance
19    - Consider upcoming regional events (11.11, Chinese New Year, Boxing Day AU)
20    """),
21    ("human", "{input}"),
22    ("placeholder", "{agent_scratchpad}")
23])
24
25tools = [get_inventory_levels, adjust_inventory, get_demand_forecast]
26agent = create_tool_calling_agent(llm, tools, prompt)
27executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

1result = executor.invoke({
2    "input": "Check variant 44012345678 across all locations. "
3             "If any location is within 20% of safety stock, "
4             "recommend a transfer or reorder."
5})
6print(result["output"])

1> Checking inventory levels for variant 44012345678...
2> SG_JURONG: 35 available (safety: 30) — within 17% of safety stock
3> Recommended: Transfer 50 units from HK_CENTRAL (current: 280, safety: 50)
4> Lead time: 2 days. Estimated arrival: 2025-07-20
5> Action: Initiated adjustment — HK_CENTRAL -50, SG_JURONG +50 (pending transit)

1# agent/consumer.py
2from confluent_kafka import Consumer
3import json
4import yaml
5from agent.inventory_agent import executor
6
7with open("config/apac_locations.yaml") as f:
8    LOCATIONS = {l["shopify_id"]: l for l in yaml.safe_load(f)["locations"]}
9
10consumer = Consumer({
11    "bootstrap.servers": "localhost:9092",
12    "group.id": "inventory-agent-group",
13    "auto.offset.reset": "latest"
14})
15consumer.subscribe(["shopify.inventory.updates"])
16
17THRESHOLD_MULTIPLIER = 1.3  # Trigger when stock < 1.3x safety stock
18
19def should_trigger_agent(event: dict) -> bool:
20    loc = LOCATIONS.get(event["location_id"])
21    if not loc:
22        return False
23    return event["available"] < loc["min_safety_stock"] * THRESHOLD_MULTIPLIER
24
25print("Inventory agent consumer started. Listening...")
26try:
27    while True:
28        msg = consumer.poll(1.0)
29        if msg is None:
30            continue
31        if msg.error():
32            print(f"Consumer error: {msg.error()}")
33            continue
34
35        event = json.loads(msg.value().decode("utf-8"))
36        if should_trigger_agent(event):
37            loc = LOCATIONS[event["location_id"]]
38            prompt = (
39                f"Inventory alert: {loc['code']} has {event['available']} units "
40                f"for inventory item {event['inventory_item_id']}. "
41                f"Safety stock is {loc['min_safety_stock']}. "
42                f"Analyse demand forecast and recommend action."
43            )
44            result = executor.invoke({"input": prompt})
45            print(f"Agent decision: {result['output']}")
46finally:
47    consumer.close()

1# agent/compliance.py
2APAC_RULES = {
3    "TW": {
4        "bonded_transfer_requires_clearance": True,
5        "clearance_lead_time_days": 2,
6        "max_domestic_transfer_value_usd": 5000  # per shipment without broker
7    },
8    "AU": {
9        "gst_threshold_aud": 1000,
10        "requires_abr_for_transfers": True,
11        "biosecurity_categories": ["food", "timber", "plants"]
12    },
13    "SG": {
14        "gst_rate": 0.09,  # 9% as of 2024
15        "free_trade_zone_locations": ["JURONG_FTZ"],
16        "permit_required_above_sgd": 400
17    },
18    "ID": {
19        "bpom_registration_required": True,  # for cosmetics/food
20        "max_import_value_without_api": 1500  # USD
21    }
22}
23
24def validate_transfer(source_location: dict, dest_location: dict, quantity: int, unit_value_usd: float) -> dict:
25    dest_country = dest_location["country"]
26    rules = APAC_RULES.get(dest_country, {})
27    warnings = []
28    blocked = False
29
30    total_value = quantity * unit_value_usd
31
32    if source_location.get("supports_bonded") and rules.get("bonded_transfer_requires_clearance"):
33        warnings.append(f"Bonded warehouse transfer to {dest_country} requires customs clearance (+{rules['clearance_lead_time_days']}d)")
34
35    if dest_country == "AU" and total_value > rules.get("gst_threshold_aud", float("inf")):
36        warnings.append(f"Transfer exceeds AU GST threshold. GST will apply.")
37
38    if dest_country == "ID" and total_value > rules.get("max_import_value_without_api", float("inf")):
39        warnings.append("Indonesia API-U import license required for this value.")
40        blocked = True
41
42    return {"allowed": not blocked, "warnings": warnings}

1# monitoring/health.py
2from flask import Flask, jsonify
3from datetime import datetime, timedelta
4import psycopg2
5import os
6
7app = Flask(__name__)
8
9@app.route("/health/agent", methods=["GET"])
10def agent_health():
11    conn = psycopg2.connect(os.environ["DATABASE_URL"])
12    cur = conn.cursor()
13
14    # Decisions in last 24 hours
15    cur.execute("""
16        SELECT 
17            COUNT(*) as total_decisions,
18            COUNT(*) FILTER (WHERE status = 'executed') as executed,
19            COUNT(*) FILTER (WHERE status = 'blocked') as blocked,
20            AVG(response_time_ms) as avg_response_ms
21        FROM agent_decisions
22        WHERE created_at > %s
23    """, [datetime.utcnow() - timedelta(hours=24)])
24
25    row = cur.fetchone()
26    conn.close()
27
28    return jsonify({
29        "status": "healthy" if row[0] > 0 else "idle",
30        "last_24h": {
31            "total_decisions": row[0],
32            "executed": row[1],
33            "blocked_by_compliance": row[2],
34            "avg_response_ms": round(row[3], 1) if row[3] else None
35        }
36    })