#!/usr/bin/env python3 """Liney starter, Python: the four handlers (write, undo, and the two pushes to Liney), the pull-mode worker and the signature check, over a tiny in-memory CRM. Standard library only; Python 3.9 or later. The contract: https://www.liney.ai/developers/llms.txt LINEY_URL=https://www.liney.ai LINEY_KEY=lk_test_… LINEY_SECRET=ls_… MODE=pull python3 server.py MODE=push PORT=8787 … python3 server.py (then give Liney https:/// as the endpoint) """ import hashlib import hmac import json import os import threading import time import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer LINEY_URL = os.environ.get("LINEY_URL", "https://www.liney.ai").rstrip("/") LINEY_KEY = os.environ.get("LINEY_KEY", "") SECRETS = [s.strip() for s in os.environ.get("LINEY_SECRET", "").split(",") if s.strip()] # current, and the previous one during a rotation MODE = os.environ.get("MODE", "pull") PORT = int(os.environ.get("PORT", "8787")) API = f"{LINEY_URL}/api/crm/custom/v1" WINDOW_S = 300 # ---- the CRM --------------------------------------------------------------------------------------- OWNERS = [{"id": "u-sam", "name": "Sam Rivera", "email": "sam@northwind.example"}, {"id": "u-lee", "name": "Lee Park", "email": "lee@northwind.example"}] STAGES = [{"id": "new", "label": "New", "order": 0}, {"id": "quote", "label": "Quote sent", "order": 1}, {"id": "signed", "label": "Signed", "order": 2, "closed": "won"}, {"id": "dead", "label": "Dead", "order": 3, "closed": "lost"}] RULE = {"field": "po_number", "stage": "signed"} db = {"deals": {}, "contacts": {}, "companies": {}, "activities": {}} replies = {} # Idempotency-Key -> reply nonces = {} # nonce -> seen at lock = threading.Lock() RUN = format(int(time.time() * 1000), "x") # ids minted during a run carry a run stamp: a tombstoned id from an earlier run is never reused counter = {"n": 100} def next_id(prefix): counter["n"] += 1 return f"{prefix}-{RUN}-{counter['n']}" def seed(): db["companies"]["CO-1"] = {"owner": "u-sam", "fields": {"name": "Acme Roofing", "website": "acme-roofing.example"}} db["contacts"]["C-1"] = {"owner": "u-sam", "fields": {"full_name": "Ada Lovelace", "email": "ada@acme-roofing.example", "title": "Ops"}, "companies": ["CO-1"]} db["deals"]["D-1"] = {"owner": "u-sam", "fields": {"title": "Kettle Works Boilers", "stage": "quote", "value": "1200", "close_date": None, "next_step": "Send quote", "owner": "u-sam", "po_number": None}, "contacts": ["C-1"], "companies": ["CO-1"]} db["deals"]["D-2"] = {"owner": "u-lee", "fields": {"title": "Maple Dental X-Ray Suite", "stage": "new", "value": "31000", "close_date": None, "next_step": None, "owner": "u-lee", "po_number": None}, "contacts": [], "companies": []} def schema_push(): return { "name": "Northwind CRM", "timezone": "America/Toronto", "currency": "CAD", "fields": { "deals": [ {"key": "title", "label": "Title", "type": "string", "role": "name"}, {"key": "stage", "label": "Stage", "type": "enum", "role": "stage", "options": [{"value": s["id"], "label": s["label"]} for s in STAGES]}, {"key": "value", "label": "Value", "type": "number", "role": "amount"}, {"key": "close_date", "label": "Close date", "type": "date", "role": "closeDate"}, {"key": "next_step", "label": "Next step", "type": "string", "role": "nextStep"}, {"key": "owner", "label": "Owner", "type": "string", "role": "owner"}, {"key": "po_number", "label": "PO number", "type": "string"}, ], "contacts": [{"key": "full_name", "label": "Name", "type": "string", "role": "name"}, {"key": "email", "label": "Email", "type": "string", "role": "email"}, {"key": "title", "label": "Job title", "type": "string", "role": "title"}], "companies": [{"key": "name", "label": "Name", "type": "string", "role": "name"}, {"key": "website", "label": "Website", "type": "string", "role": "domain"}], }, "stages": STAGES, "owners": OWNERS, "rules": [{"field": RULE["field"], "requiredAt": [RULE["stage"]]}], "capabilities": {"activities": ["note", "call", "task", "meeting"], "links": "create"}, } def records_push(touched=None, gone=None, mode="upsert"): def pick(table, ids): out = [] for i in (ids if ids is not None else list(db[table].keys())): r = db[table].get(i) if r is None: continue row = {"id": i, "owner": r["owner"], "fields": r["fields"]} if "contacts" in r: row["contacts"] = r["contacts"] if "companies" in r: row["companies"] = r["companies"] out.append(row) return out def tomb(ids): return [{"id": i, "owner": OWNERS[0]["id"], "deleted": True} for i in (ids or [])] t = touched or {} g = gone or {} return {"mode": mode, "deals": pick("deals", t.get("deals")) + tomb(g.get("deals")), "contacts": pick("contacts", t.get("contacts")) + tomb(g.get("contacts")), "companies": pick("companies", t.get("companies")) + tomb(g.get("companies"))} # ---- calls to Liney (with the key) --------------------------------------------------------------- def liney(method, path, body=None, headers=None): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(f"{API}{path}", data=data, method=method) req.add_header("authorization", f"Bearer {LINEY_KEY}") req.add_header("content-type", "application/json") req.add_header("accept", "application/json") for k, v in (headers or {}).items(): req.add_header(k, v) try: with urllib.request.urlopen(req, timeout=40) as res: text = res.read().decode() return res.status, (json.loads(text) if text else None) except urllib.error.HTTPError as e: text = e.read().decode() try: err = json.loads(text).get("error", {}) except Exception: err = {} raise RuntimeError(f"{method} {path} -> {e.code} {err.get('code', '')}: {err.get('message', text)}" + (f" (fix: {err['hint']})" if err.get("hint") else "")) def push_schema(): return liney("POST", "/schema", schema_push()) def push_records(touched=None, gone=None, mode="upsert"): p = records_push(touched, gone, mode) if not (p["deals"] or p["contacts"] or p["companies"]): return None return liney("POST", "/records", p) # ---- the handlers ------------------------------------------------------------------------------------ def refuse(code, message, field=None): e = {"code": code, "message": message} if field: e["field"] = field return {"ok": False, "error": e} def s(v): return "" if v is None else str(v) def write(body, dry_run): previous = {"deals": {}, "contacts": {}, "companies": {}} created = [] touched = {"deals": set(), "contacts": set(), "companies": set()} # Validate everything first, then apply: a refused call must leave the record as it was. for obj in ("deals", "contacts", "companies"): for rid, vals in (body.get(obj) or {}).items(): row = db[obj].get(rid) if row is None: return refuse("not_found", f"{obj[:-1]} {rid} is not in Northwind CRM") if obj == "deals" and vals.get("stage") == RULE["stage"] and not s(vals.get(RULE["field"], row["fields"].get(RULE["field"]))).strip(): return refuse("required_field", "PO number is needed to mark a deal Signed", RULE["field"]) previous[obj][rid] = {k: row["fields"].get(k) for k in vals} if dry_run: return {"ok": True, "previous": previous, "created": [], "dryRun": True} for obj in ("deals", "contacts", "companies"): for rid, vals in (body.get(obj) or {}).items(): row = db[obj][rid] for k, v in vals.items(): row["fields"][k] = None if v == "" else v if obj == "deals" and isinstance(vals.get("owner"), str) and vals["owner"]: row["owner"] = vals["owner"] touched[obj].add(rid) for c in body.get("create") or []: t = c.get("type") if t == "deal": i = next_id("D") f = dict(c.get("fields") or {}) db["deals"][i] = {"owner": str(f.get("owner") or body.get("rep")), "fields": f, "contacts": [], "companies": [str(c["companyId"])] if c.get("companyId") else []} created.append({"type": "deal", "id": i}) touched["deals"].add(i) elif t == "contact": i = next_id("C") db["contacts"][i] = {"owner": str(body.get("rep")), "fields": {"full_name": str(c.get("name")), "email": c.get("email"), "title": c.get("role")}, "companies": []} d = db["deals"].get(c.get("dealId") or "") if d is not None: d["contacts"].append(i) touched["deals"].add(c["dealId"]) created.append({"type": "contact", "id": i}) touched["contacts"].add(i) elif t == "company": i = next_id("CO") db["companies"][i] = {"owner": str(body.get("rep")), "fields": {"name": str(c.get("name")), "website": c.get("domain")}} d = db["deals"].get(c.get("dealId") or "") if d is not None: d["companies"].append(i) touched["deals"].add(c["dealId"]) created.append({"type": "company", "id": i}) touched["companies"].add(i) else: # note, call, task, meeting: an activity on the target record i = next_id("A") fields = {k: v for k, v in c.items() if k not in ("type", "target", "also", "ownerId")} db["activities"][i] = {"type": t, "target": c.get("target"), "also": c.get("also") or [], "fields": fields, "status": "open", "owner": str(c.get("ownerId") or body.get("rep"))} created.append({"type": t, "id": i}) for l in body.get("links") or []: d = db["deals"].get(l["from"]["id"]) if l["from"]["objectType"] == "deals" else None if d is not None: key = "contacts" if l["to"]["objectType"] == "contacts" else "companies" if l["to"]["objectType"] == "companies" else None if key and l["to"]["id"] not in d[key]: d[key].append(l["to"]["id"]) touched["deals"].add(l["from"]["id"]) for i in (body.get("tasks") or {}).get("complete") or []: a = db["activities"].get(i) if a is not None: a["status"] = "done" try: push_records({k: list(v) for k, v in touched.items()}) except Exception as e: # the write stands; Liney's copy catches up on the next push print("push after write:", e) return {"ok": True, "previous": previous, "created": created} def undo(body, dry_run): expect = body.get("expect") or {} restore = body.get("restore") or {} # Compare first, everything, before writing anything. for obj in ("deals", "contacts", "companies"): for rid in (restore.get(obj) or {}): row = db[obj].get(rid) if row is None: return refuse("not_found", f"{obj[:-1]} {rid} is not in Northwind CRM") for k, want in ((expect.get(obj) or {}).get(rid) or {}).items(): if s(row["fields"].get(k)) != s(want): return refuse("conflict", f"{k} changed since Liney wrote it", k) if dry_run: return {"ok": True, "dryRun": True} touched = {"deals": set(), "contacts": set(), "companies": set()} gone = {"deals": set(), "contacts": set(), "companies": set()} for obj in ("deals", "contacts", "companies"): for rid, vals in (restore.get(obj) or {}).items(): row = db[obj][rid] for k, v in vals.items(): row["fields"][k] = None if v == "" else v touched[obj].add(rid) for r in body.get("remove") or []: t, i = r.get("type"), r.get("id") if t == "deal": db["deals"].pop(i, None) gone["deals"].add(i) elif t == "contact": db["contacts"].pop(i, None) gone["contacts"].add(i) for did, d in db["deals"].items(): if i in d["contacts"]: d["contacts"] = [c for c in d["contacts"] if c != i] touched["deals"].add(did) elif t == "company": db["companies"].pop(i, None) gone["companies"].add(i) elif t == "task_done": a = db["activities"].get(i) if a is not None: a["status"] = "open" else: db["activities"].pop(i, None) # a second remove of something already gone is not an error try: push_records({k: list(v) for k, v in touched.items()}, {k: list(v) for k, v in gone.items()}) except Exception as e: print("push after undo:", e) return {"ok": True} def handle(kind, body, key, dry_run): """One call, whichever way it arrived: the idempotency key replays the first reply.""" with lock: if not dry_run and key and key in replies: return replies[key] reply = write(body, dry_run) if kind == "write" else undo(body, dry_run) if not dry_run and key: replies[key] = reply return reply # ---- push mode: the signature ----------------------------------------------------------------------- def verify(headers, method, path, raw): ts, nonce, sig = headers.get("x-liney-timestamp"), headers.get("x-liney-nonce"), headers.get("x-liney-signature") if not ts or not nonce or not sig: return "missing" try: if abs(time.time() - float(ts)) > WINDOW_S: return "stale" except ValueError: return "stale" canonical = f"{ts}\n{nonce}\n{method.upper()}\n{path}\n{hashlib.sha256(raw).hexdigest()}".encode() given = [p.strip()[3:] for p in sig.split(",") if p.strip().startswith("v1=")] good = any(hmac.compare_digest(hmac.new(secret.encode(), canonical, hashlib.sha256).hexdigest(), g) for secret in SECRETS for g in given) if not good: return "bad_signature" now = time.time() with lock: for k in [k for k, at in nonces.items() if now - at > WINDOW_S * 2]: del nonces[k] if nonce in nonces: return "replay" nonces[nonce] = now return None class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass def send(self, status, body): data = json.dumps(body).encode() self.send_response(status) self.send_header("content-type", "application/json") self.send_header("content-length", str(len(data))) self.end_headers() self.wfile.write(data) def do_POST(self): path = self.path.split("?")[0] kind = "write" if path.endswith("/liney/write") else "undo" if path.endswith("/liney/undo") else None raw = self.rfile.read(int(self.headers.get("content-length") or 0)) if not kind: return self.send(404, {"ok": False, "error": {"code": "not_found", "message": "no such handler"}}) why = verify({k.lower(): v for k, v in self.headers.items()}, "POST", path, raw) if why: return self.send(401, {"ok": False, "error": {"code": "replay" if why == "replay" else "bad_signature", "message": why}}) try: body = json.loads(raw or b"{}") self.send(200, handle(kind, body, self.headers.get("idempotency-key"), self.headers.get("x-liney-dry-run") == "1")) except Exception as e: self.send(200, refuse("invalid_value", f"the call could not be applied: {e}")) # ---- pull mode: the worker -------------------------------------------------------------------------- def poll(): while True: try: status, item = liney("GET", "/outbox") if status == 204 or not item: continue reply = handle(item["kind"], item["body"], item.get("idempotencyKey"), bool(item.get("dryRun"))) liney("POST", f"/outbox/{item['id']}", reply, {"x-liney-lease": item["lease"]}) except Exception as e: print("poll:", e) time.sleep(2) if __name__ == "__main__": seed() if not LINEY_KEY: print("LINEY_KEY is not set: nothing will be pushed to Liney and pull mode cannot poll.") if MODE == "push" and not SECRETS: print("LINEY_SECRET is not set: every call from Liney will be refused.") if LINEY_KEY: try: push_schema() push_records(mode="replace") # on start: Liney's copy becomes exactly this CRM's records print("schema and records pushed to Liney") except Exception as e: print("push:", e) if MODE == "pull": threading.Thread(target=poll, daemon=True).start() print(f"Northwind CRM on http://localhost:{PORT} ({MODE} mode)") ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()