API and webhooks

curl "https://conductcolony.com/api/v1/me" \
  -H "Authorization: Bearer $CONDUCTCOLONY_API_KEY"

Authorization: Bearer cck_your_key

{
  "error": {
    "code": "VALIDATION",
    "message": "status must be one of: alive, dead, sacrificed, transferred, missing."
  }
}

import os

import requests

BASE = "https://conductcolony.com/api/v1"
HEADERS = {"Authorization": "Bearer " + os.environ["CONDUCTCOLONY_API_KEY"]}

animals, cursor = [], None
while True:
    params = {"limit": 200}
    if cursor:
        params["cursor"] = cursor
    response = requests.get(BASE + "/animals", headers=HEADERS, params=params, timeout=30)
    response.raise_for_status()
    page = response.json()
    animals += page["data"]
    cursor = page["nextCursor"]
    if not cursor:
        break

print(len(animals), "animals")

{
  "id": "evt_cmg1x2y3z0000",
  "type": "animal.created",
  "createdAt": "2026-10-11T15:04:05.123Z",
  "apiVersion": "2026-10-11",
  "data": {
    "action": "animal_added",
    "details": {
      "animalId": "M-2026-0150",
      "sex": "male",
      "strainId": "cmg1strain0000",
      "species": "mouse"
    },
    "animal": {
      "id": "cmg1animal0001",
      "animalId": "M-2026-0150"
    },
    "cage": {
      "id": "cmg1cage0000",
      "cageNumber": "C-014"
    }
  }
}

const crypto = require("node:crypto");

// secret: the webhook's signing secret (whsec_...).
// rawBody: the request body exactly as received, as a Buffer or string.
function verifyConductColonySignature(secret, rawBody, header, toleranceSeconds = 300) {
  const fields = {};
  for (const part of String(header || "").split(",")) {
    const at = part.indexOf("=");
    if (at > 0) (fields[part.slice(0, at).trim()] ||= []).push(part.slice(at + 1).trim());
  }
  const timestamp = Number(fields.t && fields.t[0]);
  if (!Number.isInteger(timestamp)) return false;
  if (Math.abs(Date.now() / 1000 - timestamp) > toleranceSeconds) return false;
  const expected = crypto
    .createHmac("sha256", secret)
    .update(timestamp + ".")
    .update(rawBody)
    .digest();
  return (fields.v1 || []).some((given) => {
    const candidate = Buffer.from(given, "hex");
    return candidate.length === expected.length && crypto.timingSafeEqual(candidate, expected);
  });
}

const http = require("node:http");

http
  .createServer((req, res) => {
    const chunks = [];
    req.on("data", (chunk) => chunks.push(chunk));
    req.on("end", () => {
      const rawBody = Buffer.concat(chunks);
      const ok = verifyConductColonySignature(
        process.env.CONDUCTCOLONY_WEBHOOK_SECRET,
        rawBody,
        req.headers["conductcolony-signature"],
      );
      if (!ok) {
        res.writeHead(400).end("Bad signature");
        return;
      }
      const event = JSON.parse(rawBody.toString("utf8"));
      console.log(event.type, event.id);
      res.writeHead(200).end("OK");
    });
  })
  .listen(3000);

import hashlib
import hmac
import time


def verify_conductcolony_signature(secret, raw_body, header, tolerance_seconds=300):
    """secret: the webhook's signing secret (whsec_...).
    raw_body: the request body as bytes, exactly as received."""
    fields = {}
    for part in (header or "").split(","):
        name, _, value = part.strip().partition("=")
        fields.setdefault(name, []).append(value)
    try:
        timestamp = int(fields["t"][0])
    except (KeyError, ValueError):
        return False
    if abs(time.time() - timestamp) > tolerance_seconds:
        return False
    signed = str(timestamp).encode() + b"." + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, given) for given in fields.get("v1", []))

import os

from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["CONDUCTCOLONY_WEBHOOK_SECRET"]


@app.post("/conductcolony")
def conductcolony_webhook():
    ok = verify_conductcolony_signature(
        SECRET, request.get_data(), request.headers.get("ConductColony-Signature")
    )
    if not ok:
        abort(400)
    event = request.get_json()
    print(event["type"], event["id"])
    return "", 200

import os

import requests

BASE = "https://conductcolony.com/api/v1"
HEADERS = {"Authorization": "Bearer " + os.environ["CONDUCTCOLONY_API_KEY"]}

results = [
    {"animalId": "M-2026-0142", "locus": "Cre", "alleles": ["Cre", "+"], "method": "PCR"},
    {"animalId": "M-2026-0143", "locus": "Pten", "alleles": "fl/+", "testedDate": "2026-10-09"},
]
response = requests.post(BASE + "/genotypes", headers=HEADERS, json={"results": results}, timeout=60)
response.raise_for_status()
summary = response.json()["data"]
print(summary["saved"], "saved,", summary["failed"], "not saved")
for row in summary["results"]:
    if row["status"] == "error":
        print("Row", row["index"], row["error"])

curl "https://conductcolony.com/api/v1/animals?updatedSince=2026-10-10T05:59:00Z&limit=200" \
  -H "Authorization: Bearer $CONDUCTCOLONY_API_KEY"

import json
import os
import pathlib

import requests

BASE = "https://conductcolony.com/api/v1"
HEADERS = {"Authorization": "Bearer " + os.environ["CONDUCTCOLONY_API_KEY"]}
STATE = pathlib.Path("last_event.json")

after = json.loads(STATE.read_text())["after"] if STATE.exists() else None
while True:
    params = {"limit": 200}
    if after:
        params["after"] = after
    response = requests.get(BASE + "/events", headers=HEADERS, params=params, timeout=30)
    response.raise_for_status()
    page = response.json()
    for event in page["data"]:
        print(event["createdAt"], event["type"], event["data"]["details"])
        after = event["id"]
    if after:
        STATE.write_text(json.dumps({"after": after}))
    if not page["nextCursor"]:
        break