How to Automate Shopify Order Fulfillment using the API

Understanding the Shopify Fulfillment API

Setting Up Shopify API Authentication for Order Fulfillment

Fulfillment automation requires the `read_orders`, `write_fulfillments`, and `read_fulfillments` scopes. For custom apps (single-store use), generate a token via the Dev Dashboard. For distributed apps, implement OAuth 2.0 with the authorization code grant. As of January 1, 2026, all new apps must be created via dev.shopify.com or the Shopify CLI — not from the Shopify Admin.

1. 1Go to dev.shopify.com and log in with your Shopify Partner account.
2. 2Click 'Create app' and select 'Custom app' for single-store automation.
3. 3Under 'Configuration > Admin API access scopes', enable: read_orders, write_fulfillments, read_fulfillments.
4. 4Click 'Save', then navigate to 'API credentials' and click 'Install app'.
5. 5Copy the generated Admin API access token (starts with shpat_) — visible only once.
6. 6Store the token in an environment variable: export SHOPIFY_ACCESS_TOKEN='shpat_...'
7. 7Set your shop domain: export SHOPIFY_SHOP_DOMAIN='yourstore.myshopify.com'

1import os
2import requests
3
4SHOP_DOMAIN = os.environ['SHOPIFY_SHOP_DOMAIN']
5ACCESS_TOKEN = os.environ['SHOPIFY_ACCESS_TOKEN']
6
7GRAPHQL_URL = f'https://{SHOP_DOMAIN}/admin/api/2026-04/graphql.json'
8HEADERS = {
9    'X-Shopify-Access-Token': ACCESS_TOKEN,
10    'Content-Type': 'application/json',
11}
12
13# Verify auth by querying shop name
14test = requests.post(GRAPHQL_URL, json={'query': '{shop{name}}'}, headers=HEADERS)
15print(test.json())  # Should show {data: {shop: {name: 'Your Store Name'}}}

Key endpoints

1{"query": "mutation fulfillmentCreate($fulfillment: FulfillmentInput!) { fulfillmentCreate(fulfillment: $fulfillment) { fulfillment { id status trackingInfo { number url } } userErrors { field message } } }", "variables": {"fulfillment": {"lineItemsByFulfillmentOrder": [{"fulfillmentOrderId": "gid://shopify/FulfillmentOrder/12345"}], "trackingInfo": {"company": "USPS", "number": "9400111899223148739475", "url": "https://tools.usps.com/go/TrackConfirmAction?tLabels=9400111899223148739475"}, "notifyCustomer": true}}}

1{"data": {"fulfillmentCreate": {"fulfillment": {"id": "gid://shopify/Fulfillment/98765", "status": "success", "trackingInfo": [{"number": "9400111899223148739475", "url": "https://tools.usps.com/..."}]}, "userErrors": []}}}

1{"query": "{ order(id: \"gid://shopify/Order/123456\") { fulfillmentOrders(first: 5) { edges { node { id status lineItems(first: 10) { edges { node { id sku quantity } } } assignedLocation { name } } } } } }"}

1{"data": {"order": {"fulfillmentOrders": {"edges": [{"node": {"id": "gid://shopify/FulfillmentOrder/12345", "status": "OPEN", "lineItems": {"edges": [{"node": {"id": "gid://shopify/FulfillmentOrderLineItem/777", "sku": "WIDGET-RED-M", "quantity": 2}}]}, "assignedLocation": {"name": "Main Warehouse"}}}]}}}}

1{"query": "mutation { webhookSubscriptionCreate(topic: ORDERS_PAID, webhookSubscription: {callbackUrl: \"https://yourapp.com/webhooks/orders-paid\", format: JSON}) { userErrors { field message } webhookSubscription { id } } }"}

1{"data": {"webhookSubscriptionCreate": {"userErrors": [], "webhookSubscription": {"id": "gid://shopify/WebhookSubscription/555"}}}}

1{"query": "mutation fulfillmentOrderMove($id: ID!, $newLocationId: ID!) { fulfillmentOrderMove(id: $id, newLocationId: $newLocationId) { movedFulfillmentOrder { id } userErrors { field message } } }", "variables": {"id": "gid://shopify/FulfillmentOrder/12345", "newLocationId": "gid://shopify/Location/87654321"}}

1{"data": {"fulfillmentOrderMove": {"movedFulfillmentOrder": {"id": "gid://shopify/FulfillmentOrder/12346"}, "userErrors": []}}}

Step-by-step automation

1curl -X POST 'https://mystore.myshopify.com/admin/api/2026-04/graphql.json' \
2  -H 'X-Shopify-Access-Token: shpat_your_token' \
3  -H 'Content-Type: application/json' \
4  -d '{
5    "query": "mutation { webhookSubscriptionCreate(topic: ORDERS_PAID, webhookSubscription: {callbackUrl: \\"https://yourapp.com/webhooks/orders-paid\\", format: JSON}) { userErrors { field message } webhookSubscription { id } } }"
6  }'

1curl -X POST 'https://mystore.myshopify.com/admin/api/2026-04/graphql.json' \
2  -H 'X-Shopify-Access-Token: shpat_your_token' \
3  -H 'Content-Type: application/json' \
4  -d '{
5    "query": "{ order(id: \\"gid://shopify/Order/123456\\") { fulfillmentOrders(first: 5) { edges { node { id status assignedLocation { name } lineItems(first: 50) { edges { node { id sku quantity } } } } } } } }"
6  }'

1curl -X POST 'https://mystore.myshopify.com/admin/api/2026-04/graphql.json' \
2  -H 'X-Shopify-Access-Token: shpat_your_token' \
3  -H 'Content-Type: application/json' \
4  -d '{
5    "query": "mutation fulfillmentCreate($fulfillment: FulfillmentInput!) { fulfillmentCreate(fulfillment: $fulfillment) { fulfillment { id status } userErrors { field message } } }",
6    "variables": {
7      "fulfillment": {
8        "lineItemsByFulfillmentOrder": [{"fulfillmentOrderId": "gid://shopify/FulfillmentOrder/12345"}],
9        "trackingInfo": {"company": "USPS", "number": "9400111899223148739475"},
10        "notifyCustomer": true
11      }
12    }
13  }'

1curl -X POST 'https://mystore.myshopify.com/admin/api/2026-04/graphql.json' \
2  -H 'X-Shopify-Access-Token: shpat_your_token' \
3  -H 'Content-Type: application/json' \
4  -d '{"query": "{ webhookSubscriptions(first: 20) { edges { node { id topic callbackUrl createdAt } } } }"}'

Complete working code

This script sets up the full order-to-fulfillment pipeline: it registers the webhook, handles incoming order events, fetches fulfillment orders, and creates fulfillments with tracking. In production, use a queue (Redis/SQS) between webhook receipt and fulfillment processing to stay within the 5-second response window.

1import os
2import hmac
3import hashlib
4import base64
5import logging
6import requests
7from flask import Flask, request, abort
8
9logging.basicConfig(level=logging.INFO)
10app = Flask(__name__)
11
12SHOP_DOMAIN = os.environ['SHOPIFY_SHOP_DOMAIN']
13ACCESS_TOKEN = os.environ['SHOPIFY_ACCESS_TOKEN']
14APP_SECRET = os.environ['SHOPIFY_APP_SECRET']
15
16URL = f'https://{SHOP_DOMAIN}/admin/api/2026-04/graphql.json'
17HEADERS = {'X-Shopify-Access-Token': ACCESS_TOKEN, 'Content-Type': 'application/json'}
18
19def gql(query, variables=None):
20    resp = requests.post(URL, json={'query': query, 'variables': variables or {}}, headers=HEADERS)
21    resp.raise_for_status()
22    return resp.json()
23
24def get_open_fulfillment_orders(order_id):
25    q = f'{{ order(id: "gid://shopify/Order/{order_id}") {{ fulfillmentOrders(first: 5) {{ edges {{ node {{ id status }} }} }} }} }}'
26    data = gql(q)['data']['order']
27    return [e['node']['id'] for e in data['fulfillmentOrders']['edges'] if e['node']['status'] == 'OPEN']
28
29def create_fulfillment(fo_id, carrier, tracking):
30    mutation = 'mutation fulfillmentCreate($f: FulfillmentInput!) { fulfillmentCreate(fulfillment: $f) { fulfillment { id status } userErrors { field message } } }'
31    variables = {'f': {'lineItemsByFulfillmentOrder': [{'fulfillmentOrderId': fo_id}], 'trackingInfo': {'company': carrier, 'number': tracking}, 'notifyCustomer': True}}
32    result = gql(mutation, variables)
33    errors = result['data']['fulfillmentCreate']['userErrors']
34    if errors:
35        logging.error(f'Fulfillment errors for {fo_id}: {errors}')
36        return None
37    return result['data']['fulfillmentCreate']['fulfillment']
38
39def get_tracking_from_3pl(order_id):
40    # Replace with your actual 3PL API call
41    return {'carrier': 'USPS', 'tracking': '9400111899223148739475'}
42
43@app.route('/webhooks/orders-paid', methods=['POST'])
44def handle_order_paid():
45    # Validate HMAC
46    hmac_header = request.headers.get('X-Shopify-Hmac-Sha256', '')
47    computed = base64.b64encode(hmac.new(APP_SECRET.encode(), request.get_data(), hashlib.sha256).digest()).decode()
48    if not hmac.compare_digest(computed, hmac_header):
49        abort(401)
50
51    order = request.get_json()
52    order_id = order['id']
53    logging.info(f'Processing order {order_id}')
54
55    # Respond immediately — process async in production
56    fo_ids = get_open_fulfillment_orders(order_id)
57    tracking_info = get_tracking_from_3pl(order_id)
58    for fo_id in fo_ids:
59        fulfillment = create_fulfillment(fo_id, tracking_info['carrier'], tracking_info['tracking'])
60        if fulfillment:
61            logging.info(f'Created fulfillment {fulfillment["id"]} for FO {fo_id}')
62
63    return '', 200
64
65if __name__ == '__main__':
66    app.run(port=3000)

Error handling

Rate Limits for Shopify Fulfillment API

1import time
2
3def gql_with_retry(url, headers, payload, max_retries=5):
4    delay = 1
5    for attempt in range(max_retries):
6        resp = requests.post(url, json=payload, headers=headers)
7        if resp.status_code == 429:
8            print(f'Rate limited (attempt {attempt+1}), waiting {delay}s')
9            time.sleep(delay)
10            delay = min(delay * 2, 64)
11            continue
12        resp.raise_for_status()
13        result = resp.json()
14        available = result.get('extensions', {}).get('cost', {}).get('throttleStatus', {}).get('currentlyAvailable', 1000)
15        if available < 100:
16            time.sleep(2)
17        return result
18    raise RuntimeError('Max retries exceeded')

• Use a message queue (Redis, SQS, Celery) between webhook receipt and fulfillment processing — always respond to Shopify within 5 seconds.
• Implement idempotency keys using the Shopify order ID to prevent duplicate fulfillments from retry webhook deliveries.
• Monitor `extensions.cost.throttleStatus.currentlyAvailable` and slow down processing when the bucket drops below 150 points.
• Set up a daily health check to verify the webhook subscription still exists — Shopify's silent deletion is the hardest failure mode to detect.
• During high-traffic events, consider upgrading to Shopify Plus for the 10,000-point bucket and 500 pts/sec restore rate.

Security checklist

Automation use cases

No-code alternatives

Don't want to write code? These platforms can automate the same workflows visually.

Best practices

• Always respond to Shopify webhook deliveries within 5 seconds — use background queues (Celery, BullMQ, SQS) for the actual fulfillment processing.
• Implement idempotency: store each order ID as 'fulfilled' in a database before calling fulfillmentCreate to prevent duplicate fulfillments from re-delivered webhooks.
• Monitor webhook subscription health daily — Shopify's silent deletion after 19 failures is the #1 cause of unexplained gaps in order processing.
• Filter fulfillment orders by `status === 'OPEN'` before attempting to create fulfillments — already-fulfilled or cancelled orders return user errors.
• Use the modern `fulfillmentCreate` with `lineItemsByFulfillmentOrder` — the legacy REST `POST /fulfillments.json` is deprecated and may stop working.
• For partial fulfillments (shipping line items from the same order separately), specify `fulfillmentOrderLineItems` in each fulfillment request.
• Log all fulfillment mutations with order ID, fulfillment order ID, tracking number, and timestamp for customer service lookup.
• Set `notifyCustomer: false` if you're sending custom shipping notifications via your own ESP — otherwise customers get duplicate emails.

Ask AI to help

Copy one of these prompts to get a personalized, working implementation.

Related automations