"""Acme Orders: a tiny in-memory service for the governed-writes example. Python 3.12+, standard library only. It stands in for an internal API you already own. Every request is logged so you can see exactly which calls the agent made and that a refund lands only after a human approves it. """ import argparse from datetime import datetime, timezone import hmac from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import os from pathlib import Path import re import secrets import sys import threading SPEC_PATH = Path(__file__).with_name("orders-api.openapi.json") MAX_BODY_BYTES = 16 * 1024 LOCK = threading.Lock() def now(): return datetime.now(timezone.utc).isoformat(timespec="seconds") def seed_orders(): def order(order_id, status, customer, email, tier, items): total = sum(price for _, _, price in items) return { "id": order_id, "status": status, "placed_at": "2026-10-14T16:02:11+00:00", "total_cents": total, "refundable_cents": total, "customer": {"name": customer, "email": email, "tier": tier}, "items": [{"sku": sku, "name": name, "price_cents": price} for sku, name, price in items], "notes": [], "refunds": [], } orders = [ order("A-1042", "delivered", "Dana Whitfield", "dana@example.com", "gold", [("KTL-220", "Stoneware kettle", 8900), ("MUG-004", "Mug set (4)", 4000)]), order("A-1043", "shipped", "Ravi Menon", "ravi@example.com", "standard", [("LMP-310", "Desk lamp", 5400)]), order("A-1044", "processing", "Lee Okafor", "lee@example.com", "standard", [("CHR-118", "Task chair", 31900)]), ] return {item["id"]: item for item in orders} ORDERS = seed_orders() def color(text, code): return f"\033[{code}m{text}\033[0m" if sys.stdout.isatty() else text class OrdersHandler(BaseHTTPRequestHandler): server_version = "AcmeOrders/1.0" token = "" def log_message(self, format, *args): # Replaced by explicit request logging. pass def send_json(self, status, payload): body = json.dumps(payload, indent=2).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) self.log_call(status) def log_call(self, status, extra=""): stamp = datetime.now().strftime("%H:%M:%S") line = f"{stamp} {self.command:<5} {self.path:<28} -> {status}{extra}" print(color(line, "2" if self.command == "GET" else "1"), flush=True) def authorized(self): supplied = self.headers.get("Authorization", "") return hmac.compare_digest(supplied.encode(), f"Bearer {self.token}".encode()) def read_json(self): length = int(self.headers.get("Content-Length") or 0) if length > MAX_BODY_BYTES: raise ValueError("request body too large") payload = json.loads(self.rfile.read(length) or b"{}") if not isinstance(payload, dict): raise ValueError("request body must be a JSON object") return payload def do_GET(self): if self.path == "/openapi.json": body = SPEC_PATH.read_bytes() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return self.log_call(200) if not self.authorized(): return self.send_json(401, {"error": "unauthorized"}) path, _, query = self.path.partition("?") if path == "/orders": status = dict(part.split("=", 1) for part in query.split("&") if "=" in part).get("status") with LOCK: orders = [ {"id": o["id"], "status": o["status"], "total_cents": o["total_cents"], "customer_name": o["customer"]["name"]} for o in ORDERS.values() if status in (None, o["status"]) ] return self.send_json(200, {"orders": orders}) match = re.fullmatch(r"/orders/(A-\d{4})", path) if match: with LOCK: order = ORDERS.get(match[1]) snapshot = json.loads(json.dumps(order)) if order else None if snapshot is None: return self.send_json(404, {"error": "order_not_found"}) return self.send_json(200, snapshot) return self.send_json(404, {"error": "not_found"}) def do_POST(self): if not self.authorized(): return self.send_json(401, {"error": "unauthorized"}) match = re.fullmatch(r"/orders/(A-\d{4})/(notes|refunds)", self.path) if not match: return self.send_json(404, {"error": "not_found"}) try: payload = self.read_json() except ValueError as exc: return self.send_json(400, {"error": "invalid_request", "detail": str(exc)}) order_id, resource = match[1], match[2] with LOCK: order = ORDERS.get(order_id) if order is None: return self.send_json(404, {"error": "order_not_found"}) if resource == "notes": body = payload.get("body") if not isinstance(body, str) or not body.strip() or set(payload) - {"body"}: return self.send_json(400, {"error": "invalid_request"}) note = {"id": f"note_{secrets.token_hex(4)}", "body": body.strip()[:2000], "created_at": now()} order["notes"].append(note) print(color(f" note on {order_id}: {note['body']}", "36"), flush=True) return self.send_json(201, note) return self.issue_refund(order, payload) def issue_refund(self, order, payload): amount = payload.get("amount_cents") reason = payload.get("reason") notify = payload.get("notify_customer", True) if ( set(payload) - {"amount_cents", "reason", "notify_customer"} or not isinstance(amount, int) or isinstance(amount, bool) or amount < 1 or not isinstance(reason, str) or not reason.strip() or len(reason) > 500 or not isinstance(notify, bool) ): return self.send_json(400, {"error": "invalid_request"}) if amount > order["refundable_cents"]: return self.send_json(422, { "error": "refund_exceeds_refundable", "detail": f"At most {order['refundable_cents']} cents can be refunded on {order['id']}.", }) refund = { "id": f"re_{secrets.token_hex(6)}", "order_id": order["id"], "amount_cents": amount, "reason": reason.strip(), "notify_customer": notify, "created_at": now(), } order["refunds"].append(refund) order["refundable_cents"] -= amount if order["refundable_cents"] == 0: order["status"] = "refunded" banner = f" REFUND ISSUED {order['id']} ${amount / 100:,.2f} notify={notify} " print(color(banner, "30;42"), flush=True) return self.send_json(201, refund) def main(): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8090) args = parser.parse_args() OrdersHandler.token = os.environ.get("ORDERS_API_TOKEN") or secrets.token_urlsafe(24) if "ORDERS_API_TOKEN" not in os.environ: print("ORDERS_API_TOKEN was not set; generated a demo token for this run:") print(f" {OrdersHandler.token}") print("Enter it as the integration's bearer token. Set ORDERS_API_TOKEN to keep it stable.\n") print(f"Acme Orders listening on http://{args.host}:{args.port} (spec at /openapi.json)") print(f"Seeded orders: {', '.join(ORDERS)}\n") ThreadingHTTPServer((args.host, args.port), OrdersHandler).serve_forever() if __name__ == "__main__": main()