How to Automate Amazon Order Tracking using the API

1import boto3
2import json
3import requests
4import os
5import time
6import logging
7
8logging.basicConfig(level=logging.INFO)
9log = logging.getLogger(__name__)
10
11BASE_URL = 'https://sellingpartnerapi-na.amazon.com'
12SLACK_WEBHOOK = os.environ.get('SLACK_WEBHOOK_URL')
13
14class SPAPIAuth:
15    def __init__(self):
16        self._token = None
17        self._expires_at = 0
18
19    def get(self):
20        if time.time() >= self._expires_at - 300:
21            resp = requests.post('https://api.amazon.com/auth/o2/token', data={
22                'grant_type': 'refresh_token',
23                'refresh_token': os.environ['AMAZON_LWA_REFRESH_TOKEN'],
24                'client_id': os.environ['AMAZON_LWA_CLIENT_ID'],
25                'client_secret': os.environ['AMAZON_LWA_CLIENT_SECRET']
26            })
27            resp.raise_for_status()
28            d = resp.json()
29            self._token = d['access_token']
30            self._expires_at = time.time() + d['expires_in']
31        return self._token
32
33auth = SPAPIAuth()
34
35def get(path, use_token=None):
36    token = use_token or auth.get()
37    resp = requests.get(
38        f'{BASE_URL}{path}',
39        headers={'x-amz-access-token': token}
40    )
41    resp.raise_for_status()
42    return resp.json().get('payload', resp.json())
43
44def send_slack(msg):
45    if SLACK_WEBHOOK:
46        requests.post(SLACK_WEBHOOK, json={'text': msg}, timeout=5)
47
48def process_message(body):
49    try:
50        notification = json.loads(body)
51        payload = json.loads(notification.get('Message', '{}'))
52        data = payload.get('payload', payload)
53        order_id = data.get('OrderId') or data.get('AmazonOrderId')
54        change_type = data.get('OrderChangeType', 'unknown')
55    except Exception as e:
56        log.error(f'Failed to parse notification: {e}')
57        return False
58
59    if not order_id:
60        log.warning('No order ID in notification')
61        return True  # Acknowledge and discard
62
63    log.info(f'Processing ORDER_CHANGE: {order_id} ({change_type})')
64
65    order = get(f'/orders/v0/orders/{order_id}')
66    items = get(f'/orders/v0/orders/{order_id}/orderItems')['OrderItems']
67    status = order['OrderStatus']
68    channel = order.get('FulfillmentChannel', 'MFN')
69    total = order.get('OrderTotal', {})
70
71    msg = f'Order {order_id}: {status} ({channel})\n'
72    msg += f'Total: {total.get("CurrencyCode","USD")} {total.get("Amount","N/A")}\n'
73    msg += f'Items: {len(items)} SKUs'
74
75    if channel == 'AFN':  # FBA
76        msg += '\nFulfilled by Amazon — monitoring shipment status via FBA_OUTBOUND_SHIPMENT_STATUS'
77    else:  # FBM
78        if status == 'Unshipped':
79            msg += '\nAction required: submit tracking number via Listings API'
80
81    log.info(msg)
82    send_slack(msg)
83    return True
84
85def main():
86    sqs = boto3.client('sqs', region_name='us-east-1')
87    queue_url = os.environ['AMAZON_SQS_QUEUE_URL']
88
89    log.info('Starting ORDER_CHANGE notification processor')
90    while True:
91        result = sqs.receive_message(
92            QueueUrl=queue_url,
93            MaxNumberOfMessages=10,
94            WaitTimeSeconds=20
95        )
96        for msg in result.get('Messages', []):
97            try:
98                success = process_message(msg['Body'])
99                if success:
100                    sqs.delete_message(
101                        QueueUrl=queue_url,
102                        ReceiptHandle=msg['ReceiptHandle']
103                    )
104                    log.info(f'Message processed and deleted')
105            except Exception as e:
106                log.error(f'Error: {e} — message will retry')
107
108if __name__ == '__main__':
109    main()