Recipe — Python HTTP inference server

For LYNX SDK 1.0. Complete, end-to-end. Install: install/python.md. API: api/python.md. Conventions: api/conventions.md.

Goal: run lynx-basic behind an HTTP endpoint so a client that can't run the SDK on-device — a React Native / JS app, a Go/Rust/Java service, plain curl — can POST an image and get JSON detections back. Model: lynx-basic (keyless, 80 COCO classes — see models/catalog.md).

There is no built-in serve() in the SDK — the API surface is lynx.open / model.predict (see api/python.md). You wrap those in a tiny web server yourself. Below is a complete one using only the standard library (plus Pillow to decode the upload), then a FastAPI variant.

Key idea

Open the model once, at process start — the first open downloads + verifies + caches the model, so you never want that on the request path. Reuse the one Model across every request.
model.predict wants an image path or a C-contiguous (H, W, 3) uint8 RGB numpy array. The upload arrives as bytes, so decode it (Pillow) to that array before calling predict.
A Model is a single native handle — serialize calls to it with a lock. Threaded servers otherwise call predict concurrently on the same handle.

METHODThe whole thing

The whole thing (standard library)

1"""A minimal LYNX inference server. Run:  python server.py
2POST an image:  curl -F image=@photo.jpg http://localhost:8080/detect
3"""
4import io
5import json
6import threading
7from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
8 
9import numpy as np
10from PIL import Image                      # pip install pillow
11 
12import lynx
13from lynx.errors import LynxError
14 
15# ── load the model ONCE, at startup (keyless: no set_license needed) ──────────
16# First open downloads + verifies + caches lynx-basic; every request reuses this.
17MODEL = lynx.open("lynx-basic")
18# One native handle → serialize predict() across worker threads.
19MODEL_LOCK = threading.Lock()
20 
21MIN_CONFIDENCE = 0.4
22 
23 
24def run_detection(image_bytes: bytes) -> list:
25    """Decode uploaded bytes → (H, W, 3) uint8 RGB → predict → list of dicts."""
26    rgb = np.asarray(Image.open(io.BytesIO(image_bytes)).convert("RGB"))
27    with MODEL_LOCK:                        # one handle, one predict at a time
28        frame = MODEL.predict(rgb, conf=MIN_CONFIDENCE)
29        results = []
30        for d in frame.detections:
31            x1, y1, x2, y2 = (float(v) for v in d.box)   # [x1,y1,x2,y2] px
32            results.append({
33                "box": [x1, y1, x2, y2],
34                "class": MODEL.classes(d.class_id).name,  # e.g. "PERSON"
35                "confidence": float(d.confidence),
36            })
37    return results
38 
39 
40class Handler(BaseHTTPRequestHandler):
41    def _send_json(self, code, payload):
42        body = json.dumps(payload).encode("utf-8")
43        self.send_response(code)
44        self.send_header("Content-Type", "application/json")
45        self.send_header("Content-Length", str(len(body)))
46        self.end_headers()
47        self.wfile.write(body)
48 
49    def do_GET(self):
50        if self.path == "/health":
51            self._send_json(200, {"status": "ok"})
52        elif self.path == "/info":
53            self._send_json(200, {
54                "model": "lynx-basic",
55                "version": MODEL.version,
56                "classes": [c.name for c in MODEL.classes],
57            })
58        else:
59            self._send_json(404, {"error": "not found"})
60 
61    def do_POST(self):
62        if self.path != "/detect":
63            return self._send_json(404, {"error": "not found"})
64        length = int(self.headers.get("Content-Length", 0))
65        if length <= 0:
66            return self._send_json(400, {"error": "empty body"})
67        raw = self.rfile.read(length)
68        image_bytes = _extract_image(raw, self.headers.get("Content-Type", ""))
69        try:
70            detections = run_detection(image_bytes)
71        except LynxError as e:
72            return self._send_json(500, {"error": f"lynx [{e.code}]: {e.message}"})
73        except Exception as e:                        # bad/undecodable upload
74            return self._send_json(400, {"error": str(e)})
75        self._send_json(200, detections)
76 
77    def log_message(self, *a):                        # quiet the default logging
78        pass
79 
80 
81def _extract_image(raw: bytes, content_type: str) -> bytes:
82    """Accept either a raw image body (Content-Type: image/*) or a single
83    multipart/form-data 'image' field. Pillow sniffs the format, so we only
84    need to peel off the multipart envelope when present."""
85    if "multipart/form-data" not in content_type or "boundary=" not in content_type:
86        return raw                                    # raw image bytes
87    boundary = ("--" + content_type.split("boundary=", 1)[1]).encode()
88    for part in raw.split(boundary):
89        head, _, body = part.partition(b"\r\n\r\n")
90        if b"Content-Disposition" in head and b"filename" in head:
91            return body.rsplit(b"\r\n", 1)[0]         # strip trailing CRLF
92    raise ValueError("no image part in multipart body")
93 
94 
95if __name__ == "__main__":
96    print("lynx-basic loaded:", MODEL.version, "| listening on :8080")
97    ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()

Call it:

1# multipart (what an HTML form / RN FormData sends)
2curl -F image=@photo.jpg http://localhost:8080/detect
3 
4# or raw bytes
5curl --data-binary @photo.jpg -H "Content-Type: image/jpeg" \
6     http://localhost:8080/detect

Response:

1[
2  {"box": [34.0, 51.0, 220.0, 470.0], "class": "PERSON",  "confidence": 0.94},
3  {"box": [255.0, 120.0, 610.0, 430.0], "class": "BICYCLE", "confidence": 0.81}
4]

FastAPI variant

If you'd rather have automatic multipart handling, OpenAPI docs, and an ASGI server, the same three moves (open once, lock, decode-to-array) drop into FastAPI. pip install fastapi uvicorn pillow, then:

1# app.py  —  run: uvicorn app:app --host 0.0.0.0 --port 8080 --workers 1
2import io
3import threading
4 
5import numpy as np
6from PIL import Image
7from fastapi import FastAPI, File, UploadFile
8 
9import lynx
10 
11MODEL = lynx.open("lynx-basic")            # once, at import (keyless)
12MODEL_LOCK = threading.Lock()
13app = FastAPI()
14 
15 
16@app.get("/health")
17def health():
18    return {"status": "ok"}
19 
20 
21@app.get("/info")
22def info():
23    return {"model": "lynx-basic", "version": MODEL.version,
24            "classes": [c.name for c in MODEL.classes]}
25 
26 
27@app.post("/detect")
28async def detect(image: UploadFile = File(...)):
29    rgb = np.asarray(Image.open(io.BytesIO(await image.read())).convert("RGB"))
30    with MODEL_LOCK:
31        frame = MODEL.predict(rgb, conf=0.4)
32        return [
33            {"box": [float(v) for v in d.box],
34             "class": MODEL.classes(d.class_id).name,
35             "confidence": float(d.confidence)}
36            for d in frame.detections
37        ]

Run it single-process so there's exactly one loaded Model:

1uvicorn app:app --host 0.0.0.0 --port 8080 --workers 1

Notes

One model instance. Keep --workers 1 (FastAPI) / one process (stdlib). Each extra worker process re-runs lynx.open — another full model in memory (and, cold, another download). To scale out, run N single-model processes behind a load balancer rather than N workers in one process.
Serialize predict. A Model is one native handle; the MODEL_LOCK makes threaded/async servers call it one request at a time. If you need real parallelism, put a pool of separate Model instances (each its own lynx.open) behind the lock-free path, or shard across processes. On GPU, set lynx.set_workers(gpus=1) once before open (see api/python.md).
Keyless. lynx-basic needs no key — the SDK mints a per-device trial on first load. No lynx.set_license(...) call.
box is [x1, y1, x2, y2] in the uploaded image's pixels (top-left origin). Class name comes from model.classes(d.class_id).name — there's no Detection.class_name. conf= accepts a bare float, a ConfMode, or None for the model's calibrated default (see recipes/python-detection.md).
Not a production service. This is the "make it callable over HTTP" escape hatch — no auth, no rate limiting, no batching. Put it behind a real gateway (TLS, auth, timeouts) before it faces anything but your own network.