Ashby API Guide: Authentication, Candidates, Resumes and Webhooks

# ashby.py: call the Ashby API and sync candidates page by page (Python 3.10+).
import os
import time
import requests

session = requests.Session()
session.auth = (os.environ["ASHBY_API_KEY"], "")  # key as username, empty password
session.headers.update({"Accept": "application/json", "Content-Type": "application/json"})
RESTART = {"sync_token_expired", "sync_token_invalid", "incremental_sync_too_large",
           "next_cursor_expired", "cursor_invalid", "invalid_next_cursor"}  # drop token, start over

def ashby(method: str, payload: dict | None = None) -> dict:
    for attempt in range(5):
        r = session.post(f"https://api.ashbyhq.com/{method}", json=payload or {}, timeout=30)
        if r.status_code == 429 or r.status_code >= 500:
            time.sleep(2**attempt)  # back off, then retry
            continue
        r.raise_for_status()  # 401 missing key, 403 wrong key or missing permission
        body = r.json()
        if not body.get("success"):  # client errors arrive as HTTP 200
            raise RuntimeError((body.get("errorInfo") or {}).get("code", "ashby_error"))
        return body
    raise RuntimeError(f"{method} still failing after 5 attempts")

def sync_candidates(sync_token: str | None, handle) -> str | None:
    cursor = None
    while True:
        payload = {"limit": 100, "syncToken": sync_token, "cursor": cursor}
        try:
            page = ashby("candidate.list", {k: v for k, v in payload.items() if v})
        except RuntimeError as error:
            if str(error) not in RESTART or not (sync_token or cursor):
                raise
            sync_token = cursor = None  # clean full sync
            continue
        for candidate in page["results"]:
            handle(candidate)
        if not page.get("moreDataAvailable"):
            return page.get("syncToken")  # only on the last page: store it for next run
        cursor = page["nextCursor"]

# resume_file.py: turn a candidate's resume file handle into bytes.
import requests

from ashby import ashby


def download_resume(candidate: dict) -> tuple[str, bytes] | None:
    handle = candidate.get("resumeFileHandle")  # {"id", "name", "handle"} or None
    if not handle:
        return None
    url = ashby("file.info", {"fileHandle": handle["handle"]})["results"]["url"]
    r = requests.get(url, timeout=60)  # plain GET: don't send Ashby credentials here
    r.raise_for_status()
    return handle["name"], r.content

# parse.py: send the resume to HireLayer CV Extract and get JSON back.
import os

import requests


def parse_resume(file_name: str, content: bytes, candidate_id: str) -> dict | None:
    r = requests.post(
        "https://hirelayer.co/api/v3/parser",
        headers={"X-API-Key": os.environ["HIRELAYER_API_KEY"]},
        files={"file": (file_name, content)},
        data={"application_id": f"ashby:{candidate_id}"},  # echoed in info_resume
        timeout=150,  # usually about 35 s; the gateway returns 504 after 145 s
    )
    if r.status_code in (400, 413, 415):
        return None  # bad, too large or unsupported file: don't resend it unchanged
    r.raise_for_status()  # 401 bad key, 403 no credits left, 5xx: retry later
    return r.json()

{
  "info_resume": { "application_id": "ashby:3ae2b801-...", "language": "EN" },
  "info_candidate": {
    "full_name": "Alex Morgan",
    "job_title": "Senior Software Engineer",
    "experience_level": "5 to 10 years"
  },
  "work_experiences": [{ "company_name": "Northstar Labs", "start_date": "2022-03-01", "currently_active": true }],
  "skills": [{ "skill_title": "TypeScript", "skill_type": "Software skill", "status": "normalized" }]
}

# webhook.py: parse resumes as applications arrive (pip install flask requests).
import hashlib
import hmac
import os
from concurrent.futures import ThreadPoolExecutor

from flask import Flask, abort, request

from ashby import ashby
from parse import parse_resume
from resume_file import download_resume

app = Flask(__name__)
SECRET = os.environ["ASHBY_WEBHOOK_SECRET"].encode()
workers = ThreadPoolExecutor(max_workers=4)


def process(candidate_id: str) -> None:
    candidate = ashby("candidate.info", {"id": candidate_id})["results"]
    resume = download_resume(candidate)
    if resume:
        parsed = parse_resume(*resume, candidate_id)
        # store parsed, or post a summary with candidate.createNote


@app.post("/webhooks/ashby")
def ashby_webhook():
    raw = request.get_data()  # verify the raw body, before any JSON parsing
    expected = "sha256=" + hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, request.headers.get("Ashby-Signature", "")):
        abort(401)
    event = request.get_json()
    if event.get("action") == "applicationSubmit":
        workers.submit(process, event["data"]["application"]["candidate"]["id"])
    return "", 200  # answer now, parse in the background