Quickstart Code Example

import os
import time
import requests
BASE = "https://formfill.emtelligent.com"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['FORMFILL_API_KEY']}"

document = (
    "SURGICAL PATHOLOGY REPORT\n"
    "Specimen: shave biopsy, right forearm.\n"
    "Diagnosis: Nodular basal cell carcinoma.\n"
    "Margins: clear. Perineural invasion: not identified.\n"
)
response = session.post(
    f"{BASE}/v1/jobs",
    headers={"Idempotency-Key": "biopsy-0001"},
    json={
        "type": "questions",
        "document": {"text": document},
        "questions": [
            "What is the diagnosis?",
            {"id": "side", "text": "Which side was biopsied?",
             "type": "choice", "options": ["Left", "Right", "Not stated"]},
            {"id": "pni", "text": "Is perineural invasion present?",
             "type": "boolean"},
        ],
    },
)
response.raise_for_status()
job_id = response.json()["id"]

while True:
    job = session.get(f"{BASE}/v1/jobs/{job_id}").json()
    if job["status"] not in ("queued", "running"):
        break
    time.sleep(1)
if job["status"] != "succeeded":
    raise RuntimeError(f"job {job_id} {job['status']}: {job.get('error')}")

result = session.get(f"{BASE}/v1/jobs/{job_id}/result").json()
for a in result["answers"]:
    if a["status"] == "answered":
        quote = a["evidence"]["quote"] if a["evidence"] else ""
        print(f"{a['id']}: {a['answer']!r} ({a['confidence']:.2f}) ← {quote!r}")
    else:
        print(f"{a['id']}: {a['status']}")

"""Ask questions of a clinical document with the Form-fill API.
    pip install requests
    export FORMFILL_API_KEY='...'
    python formfill_quickstart.py
"""
import os
import sys
import time
import uuid
import requests
BASE = "https://formfill.emtelligent.com"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['FORMFILL_API_KEY']}"
DOCUMENT = (
    "SURGICAL PATHOLOGY REPORT\n"
    "Specimen: shave biopsy, right forearm.\n"
    "Diagnosis: Nodular basal cell carcinoma.\n"
    "Margins: clear. Perineural invasion: not identified.\n"
)
QUESTIONS = [
    "What is the diagnosis?",
    {"id": "side", "text": "Which side was biopsied?",
     "type": "choice", "options": ["Left", "Right", "Not stated"]},
    {"id": "pni", "text": "Is perineural invasion present?", "type": "boolean"},
]
def call(method, path, **kwargs):
    """One API request. Waits and retries when the API asks it to (429, 503);
    exits with the API's error code for anything else that fails."""
    for attempt in range(5):
        response = session.request(method, f"{BASE}{path}", timeout=30, **kwargs)
        if response.status_code in (429, 503):
            time.sleep(int(response.headers.get("Retry-After", 2 ** attempt)))
            continue
        if response.status_code >= 400:
            try:
                error = response.json()["error"]
            except (ValueError, KeyError):
                error = {"code": "http_error", "message": response.text[:200]}
            sys.exit(f"{method} {path} failed: {response.status_code} {error.get('code')}: "
                     f"{error.get('message')} (request_id {error.get('request_id')})")
        return response.json()
    sys.exit(f"{method} {path}: the API is still busy after 5 attempts")
def main():
    # Submit. The Idempotency-Key makes a retried submission return the same job
    # instead of creating (and charging for) a second one.
    job = call("POST", "/v1/jobs",
               headers={"Idempotency-Key": str(uuid.uuid4())},
               json={"type": "questions",
                     "document": {"text": DOCUMENT},
                     "questions": QUESTIONS})
    print(f"submitted {job['id']}")
    # Wait for it to finish. A job usually takes 5 to 10 seconds.
    deadline = time.monotonic() + 120
    while job["status"] in ("queued", "running"):
        if time.monotonic() > deadline:
            sys.exit(f"job {job['id']} is still {job['status']} after 2 minutes")
        time.sleep(1)
        job = call("GET", f"/v1/jobs/{job['id']}")
    if job["status"] != "succeeded":
        sys.exit(f"job {job['id']} {job['status']}: {job.get('error')}")
    # Download the result. This works once: store it before doing anything else.
    result = call("GET", f"/v1/jobs/{job['id']}/result")
    for answer in result["answers"]:
        if answer["status"] == "answered":
            quote = (answer.get("evidence") or {}).get("quote", "")
            print(f"{answer['id']}: {answer['answer']!r} "
                  f"(confidence {answer['confidence']:.2f}) <- {quote!r}")
        else:
            print(f"{answer['id']}: {answer['status']}")
if __name__ == "__main__":
    main()