Create a Case by Contract with the API

Response rounds

The server infers the round from the case. Send only the fields in that round's list. Each file field may also be sent as its *_tickets sibling from POST /api/v1/uploads.

"""DecisionLayer API: file an arbitration case by contract and track it.

The flow this script walks through:

1. Create the case with the contract, your opening argument, evidence, and
   the respondent's identity and address (one API call). The case lands in
   the ``awaiting_signature`` state. Files under 8 MiB go in the multipart
   body. A larger file is uploaded with ``POST /api/v1/uploads``, then sent
   as a ticket (``contract_file_ticket`` or ``evidence_tickets``).
2. Sign the arbitration terms with ``POST /api/v1/cases/{id}/sign``. Open
   the returned ``signing_url`` before ``expires_at``. Call that endpoint
   again for a fresh URL. Payment and identity verification are not API
   calls: follow ``action_url`` for those.
3. Poll the case. ``action_required`` tells you whether the case is still
   waiting on you, ``next_action`` says what to do, and ``action_url`` is
   where to do it. Your dashboard shows the same list. ``GET /api/v1/events``
   is the change feed for the same cases.
4. Respond in turns. When ``status`` is ``awaiting_response`` and
   ``current_turn`` is your role, read ``next_round`` and
   ``accepted_fields`` on the case, then ``POST /api/v1/cases/{id}/responses``
   with only those fields (``argument`` and ``affirmation=true`` are always
   required). A field outside that list is a ``422`` that names the inferred
   round. ``GET .../responses`` returns the whole thread, oldest first.
   Posting when it is not your turn returns ``409`` with details saying
   whose turn it is. Create and submit send a fresh ``Idempotency-Key``
   scoped to this API key. A retry is 409 while that key holds the lease.
5. When the case is decided, ``GET /api/v1/cases/{id}/decision`` returns the
   award text and, while it is still served, ``pdf_url``. ``view_url`` is
   the same answer on the web. A ``404`` means the award is not published yet.

Respondents use the same script with their own API key and CASE_ID set to the
case from the notification email. The script prints who the key is
(GET /api/v1/me) and follows case["role"]. If the case has not been claimed,
GET returns reason "not_claimed". Set VERIFICATION_CODE to claim it with
POST /cases/{id}/claim in this run; leave the code blank to stop and claim
later.

Test keys (dvarb_test_...): creating the case records signature, payment,
and identity verification for you and opens respondent round 1 at once. No
email is sent and nothing is left to do on the web. Set RESPONDENT_EMAIL to
the email of a second account and RESPONDENT_API_KEY to that account's test
key, and this script plays both sides through all three rounds. A test
filing stops at ``submitted``; no award is generated. With a production key
the case is ``in_review`` after each round until DecisionLayer passes the
turn; with test keys the turn passes immediately.

Claimants only need to change API_KEY below. Respondents also set CASE_ID
and, to claim in this run, VERIFICATION_CODE.
Create a key at https://www.decisionlayer.ai/settings/api-keys

Then run:

    pip install requests
    python create_real_case.py
"""

import mimetypes
import os
import time
import uuid

import requests

# 1. Paste your API key here (from /settings/api-keys).
API_KEY = "PASTE_YOUR_KEY_HERE"

# 2. Base URL of the DecisionLayer API. Use http://localhost:8000 locally.
BASE_URL = "https://www.decisionlayer.ai"

HEADERS = {"Authorization": f"Bearer {API_KEY}"}

# 3. Leave blank to file a new case. Respondents paste the notified case_ id.
CASE_ID = ""

# 4. Respondents: the verification code from the notice. Blank skips claim.
VERIFICATION_CODE = ""

# 5. The respondent's email. A production case notifies this address, so use
#    a real one. With test keys, use the respondent key owner's email.
RESPONDENT_EMAIL = "jordan.chen@example.com"

# 6. Test keys only: the respondent's dvarb_test_ key, to play both sides.
RESPONDENT_API_KEY = ""

# Multipart below this size. At or above it, use POST /uploads and a ticket.
LARGE_FILE_BYTES = 8 * 1024 * 1024

# Direct-upload ticket field for each multipart file field.
TICKET_FIELDS = {
    "contract_file": "contract_file_ticket",
    "evidence": "evidence_tickets",
    "evidence_response_files": "evidence_response_files_tickets",
    "counterclaim_files": "counterclaim_files_tickets",
}

# Case ids this run has already requested a signing URL for.
_signing_started = set()


def write_sample_file(name: str, text: str) -> str:
    """Create a small attachment so the demo is self-contained."""
    with open(name, "w") as handle:
        handle.write(text)
    return name


def idempotent_headers() -> dict:
    """Copy the auth header and add a one-time Idempotency-Key.

    The token is scoped to this API key. The same token on another key,
    including a test key, is a separate reservation. Reuse a key only to
    retry the same body. A different body is 409. A retry is 409 while
    this key holds the lease. An abandoned attempt can be reused after
    the lease expires, about a minute.
    """
    headers = dict(HEADERS)
    headers["Idempotency-Key"] = str(uuid.uuid4())
    return headers


def upload_large_file(path: str) -> str:
    """PUT a file of 8 MiB or larger and return its ticket.

    POST /uploads declares the name and exact size. PUT the bytes to
    ``upload_url`` with the returned Content-Type and Content-Length, then
    send the ticket in the matching form field.
    """
    with open(path, "rb") as handle:
        body = handle.read()
    prepared = requests.post(
        f"{BASE_URL}/api/v1/uploads",
        headers=HEADERS,
        json={"files": [{"filename": os.path.basename(path), "size": len(body)}]},
        timeout=60,
    )
    prepared.raise_for_status()
    upload = prepared.json()["uploads"][0]
    put = requests.put(
        upload["upload_url"],
        headers={
            "Content-Type": upload["content_type"],
            "Content-Length": str(len(body)),
        },
        data=body,
        timeout=300,
    )
    put.raise_for_status()
    return upload["ticket"]


def append_attachment(parts: list, handles: list, field: str, path: str) -> None:
    """Add ``path`` as a multipart file, or as a ticket when it is large."""
    if os.path.getsize(path) >= LARGE_FILE_BYTES:
        parts.append((TICKET_FIELDS[field], (None, upload_large_file(path))))
        return
    handle = open(path, "rb")
    handles.append(handle)
    mime = mimetypes.guess_type(path)[0] or "application/octet-stream"
    parts.append((field, (os.path.basename(path), handle, mime)))


def post_form(url: str, fields: dict, attachments: list) -> requests.Response:
    """POST multipart fields. ``attachments`` is a list of (field, path)."""
    parts = [(name, (None, value)) for name, value in fields.items()]
    handles = []
    try:
        for field, path in attachments:
            append_attachment(parts, handles, field, path)
        return requests.post(
            url,
            headers=idempotent_headers(),
            files=parts,
            timeout=60,
        )
    finally:
        for handle in handles:
            handle.close()


def create_case() -> dict:
    """File the case: contract + argument + evidence + respondent."""
    write_sample_file(
        "sample_contract.txt", "Sample contract for the DecisionLayer demo.\n"
    )
    write_sample_file("sample_evidence.txt", "Sample evidence: deposit receipt.\n")

    # Your own name, email, and address default to your account and the
    # claimant address below (filing_capacity=personal). Mailing addresses
    # are required so the respondent can be formally notified.
    data = {
        "question_for_arbitration": "Respondent kept a $3,500 deposit.",
        "argument": (
            "The contract required return of the deposit within 30 days. "
            "It has been 90 days and the deposit has not been returned."
        ),
        "financial_demand_usd": "3500.00",
        "respondent_first_name": "Jordan",
        "respondent_last_name": "Chen",
        "respondent_email": RESPONDENT_EMAIL,
        "respondent_street_address": "12 Main St",
        "respondent_city": "Austin",
        "respondent_state": "TX",
        "respondent_zipcode": "78701",
        "claimant_street_address": "800 Oak Ave",
        "claimant_city": "Denver",
        "claimant_state": "CO",
        "claimant_zipcode": "80202",
        # Affirm the information you are submitting is accurate and complete.
        "claimant_affirmation": "true",
    }
    response = post_form(
        f"{BASE_URL}/api/v1/cases",
        data,
        [
            ("contract_file", "sample_contract.txt"),
            ("evidence", "sample_evidence.txt"),
        ],
    )

    if response.status_code != 201:
        print(f"Create failed ({response.status_code}):")
        print(response.json())
        raise SystemExit(1)

    result = response.json()
    case = result["case"]
    print(f"Created case: {case['id']} (status: {case['status']})")
    if case["test"]:
        # A test key has nothing to sign or pay. The message says so.
        print(result["message"])
        return case
    print(f"Next step: {case['next_action']}")
    print(f"Finish it here: {result['action_url']}")
    print("Payment and identity verification follow there after you sign.")
    return case


def who_am_i() -> dict:
    """The account behind this API key. Role is per case, not per key."""
    response = requests.get(f"{BASE_URL}/api/v1/me", headers=HEADERS, timeout=60)
    response.raise_for_status()
    return response.json()


def claim_case(case_id: str, code: str) -> dict:
    """Claim the case as the respondent with the notice's verification code."""
    response = requests.post(
        f"{BASE_URL}/api/v1/cases/{case_id}/claim",
        headers=HEADERS,
        json={"verification_code": code},
        timeout=60,
    )
    if response.status_code != 200:
        print(f"Claim failed ({response.status_code}):")
        print(response.json())
        raise SystemExit(1)
    case = response.json()
    print(f"Claimed case: {case_id} (role: {case['role']})")
    return case


def get_case(case_id: str) -> dict:
    """Fetch the case: status, next action, current_turn, and view_url.

    A named respondent who has not claimed the case gets reason not_claimed.
    That is not a missing id. With VERIFICATION_CODE set, this claims it.
    """
    response = requests.get(
        f"{BASE_URL}/api/v1/cases/{case_id}", headers=HEADERS, timeout=60
    )
    if response.status_code == 404:
        error = response.json()["error"]
        if error.get("reason") == "not_claimed":
            code = VERIFICATION_CODE.strip()
            if code:
                print("This case has not been claimed yet. Claiming it now.")
                return claim_case(case_id, code)
            print("This case has not been claimed yet.")
            for detail in error["details"]:
                print(f"  - {detail}")
            raise SystemExit(1)
    response.raise_for_status()
    return response.json()


def list_cases_needing_action() -> list:
    """Every case that is waiting on you (unfinished filings, your turn, ...)."""
    response = requests.get(
        f"{BASE_URL}/api/v1/cases",
        headers=HEADERS,
        params={"action_required": "true"},
        timeout=60,
    )
    response.raise_for_status()
    return response.json()


def list_events() -> dict:
    """The change feed. Pass next_cursor as cursor on the following call."""
    response = requests.get(f"{BASE_URL}/api/v1/events", headers=HEADERS, timeout=60)
    response.raise_for_status()
    return response.json()


def list_responses(case_id: str) -> list:
    """The case's response thread, oldest first (both parties see the same)."""
    response = requests.get(
        f"{BASE_URL}/api/v1/cases/{case_id}/responses", headers=HEADERS, timeout=60
    )
    response.raise_for_status()
    return response.json()


def print_thread(case_id: str) -> None:
    """Show every submitted round on the case."""
    thread = list_responses(case_id)
    print(f"{len(thread)} response(s) on the case:")
    for entry in thread:
        names = ", ".join(f["name"] for f in entry["evidence_files"]) or "no files"
        argument = entry["argument"] or "no argument"
        print(
            f"  Round {entry['round']} by the {entry['submitted_by']} "
            f"at {entry['submitted_at']}: {argument[:60]} [{names}]"
        )


def submit_response(case_id: str, argument: str, evidence_paths=()) -> dict | None:
    """Submit your turn using only the fields this round accepts.

    ``GET /cases/{id}`` reports ``next_round`` and ``accepted_fields``.
    ``argument`` and ``affirmation`` are always sent. Optional evidence and
    a round-1 ``evidence_demands`` value are added only when that name is
    in ``accepted_fields``. A file of 8 MiB or larger is sent as a ticket.
    A 409 means the case is not waiting on you. A 422 means a field was
    rejected; the details name the inferred round.
    """
    case = get_case(case_id)
    accepted = case.get("accepted_fields") or []
    print(f"Next round: {case.get('next_round')}")
    print(f"Accepted fields: {', '.join(accepted) or 'none'}")

    data = {"argument": argument, "affirmation": "true"}
    if "evidence_demands" in accepted:
        data["evidence_demands"] = "The claimant's March bank statement."

    attachments = []
    if "evidence" in accepted:
        attachments = [("evidence", path) for path in evidence_paths]
    response = post_form(
        f"{BASE_URL}/api/v1/cases/{case_id}/responses",
        data,
        attachments,
    )

    if response.status_code == 409:
        print("Not your turn yet:")
        for detail in response.json()["error"]["details"]:
            print(f"  - {detail}")
        return None
    if response.status_code == 422:
        print("The response was rejected:")
        for detail in response.json()["error"]["details"]:
            print(f"  - {detail}")
        return None
    response.raise_for_status()

    result = response.json()
    submitted = result["response"]
    print(f"Response submitted. Round {submitted['round']} is on record.")
    case = result["case"]
    if case["action_required"]:
        print(f"Waiting on you: {case['next_action']} -> {case['action_url']}")
    return result


def start_signing(case_id: str) -> dict | None:
    """Ask for a signing URL. Open it before expires_at, then call again."""
    response = requests.post(
        f"{BASE_URL}/api/v1/cases/{case_id}/sign",
        headers=HEADERS,
        timeout=60,
    )
    if response.status_code == 409:
        print("Signing is not available yet:")
        for detail in response.json()["error"]["details"]:
            print(f"  - {detail}")
        return None
    response.raise_for_status()
    session = response.json()
    print(f"Sign the terms: {session['signing_url']}")
    print(
        f"Provider: {session['provider']}; " f"open it before {session['expires_at']}."
    )
    return session


def follow_next_action(case: dict) -> None:
    """Print the step waiting on this key.

    A signing URL is requested once in this run. Call
    ``POST /cases/{id}/sign`` again after it expires.
    """
    if not case.get("action_required"):
        return
    if case.get("next_action") == "sign_terms":
        case_id = case["id"]
        if case_id in _signing_started:
            print("Still waiting on you to sign. Open the signing URL above.")
            return
        if start_signing(case_id):
            _signing_started.add(case_id)
        return
    print(f"Waiting on you: {case['next_action']} -> {case['action_url']}")


def submit_as_test_respondent(case_id: str) -> dict | None:
    """Submit the respondent's round with RESPONDENT_API_KEY (test keys only).

    A test respondent named by email needs no claim code and no signature.
    """
    global HEADERS
    claimant_headers = HEADERS
    HEADERS = {"Authorization": f"Bearer {RESPONDENT_API_KEY.strip()}"}
    try:
        return submit_response(case_id, "The deposit was forfeited under section 4.2.")
    finally:
        HEADERS = claimant_headers


def show_decision(case: dict) -> None:
    """Read the published award. view_url is the same answer on the web."""
    response = requests.get(
        f"{BASE_URL}/api/v1/cases/{case['id']}/decision",
        headers=HEADERS,
        timeout=60,
    )
    if response.status_code == 404:
        print(
            "The award is not published yet. "
            f"View the case online: {case['view_url']}"
        )
        return
    response.raise_for_status()
    award = response.json()
    print(f"The decision is ready. View it online: {case['view_url']}")
    print(award["text"])
    if award.get("pdf_url"):
        print(f"PDF: {award['pdf_url']}")


def main() -> None:
    me = who_am_i()
    print(f"Using API key for {me['email']}")

    case_id = CASE_ID.strip()
    if case_id:
        case = get_case(case_id)
        print(
            f"Using existing case: {case_id} "
            f"(role: {case['role']}, status: {case['status']})"
        )
    else:
        case = create_case()
        case_id = case["id"]

    # Poll for the case state. In practice you'd check back after finishing
    # the web steps; polling here just demonstrates the read surface. A test
    # filing played from both sides needs a few more reads to finish.
    for _ in range(8 if case["test"] else 5):
        case = get_case(case_id)
        status, turn = case["status"], case["current_turn"]
        print(f"Status: {status}, turn: {turn}")

        if status == "decided":
            show_decision(case)
            return
        if case["test"] and status == "submitted":
            print("All three rounds are in. Test filings stop at submitted;")
            print("no award is generated, so GET .../decision stays 404.")
            return
        follow_next_action(case)
        waiting_on_other = status == "awaiting_response" and turn != case["role"]
        if case["test"] and waiting_on_other:
            if turn == "respondent" and RESPONDENT_API_KEY.strip():
                if submit_as_test_respondent(case_id):
                    print_thread(case_id)
                continue
            print(f"It is the {turn}'s turn, and this key is the {case['role']}.")
            print(
                "Set RESPONDENT_API_KEY (and RESPONDENT_EMAIL) to play both "
                "sides, or run this script with the other party's test key "
                f"and CASE_ID = {case_id!r}."
            )
            return
        if status == "awaiting_response" and turn == case["role"]:
            reply_evidence = write_sample_file(
                "sample_reply_evidence.txt", "Sample evidence: the signed addendum.\n"
            )
            submitted = submit_response(
                case_id,
                "Section 4.2 does not apply; the termination was mutual.",
                evidence_paths=[reply_evidence],
            )
            if submitted:
                print_thread(case_id)
        time.sleep(5)

    pending = list_cases_needing_action()
    print(f"You have {len(pending)} case(s) waiting on you.")
    feed = list_events()
    print(f"{len(feed['events'])} event(s) on the change feed.")
    print("Still in progress. Re-run later, or check your dashboard.")
    print(f"You can always view the case online: {case['view_url']}")


if __name__ == "__main__":
    main()

Understanding errors

Every error returns a consistent JSON shape with a list of actionable details. Validation problems are reported together so one retry fixes them all:

{
  "error": {
    "status": 422,
    "message": "Your request could not be processed.",
    "details": [
      "Field 'contract_file' is required. Attach the governing contract (.pdf, .docx, .rtf, or .txt) that contains the arbitration clause.",
      "Field 'respondent_city' is required. Provide the respondent's mailing address; e.g. respondent_city='Austin'."
    ]
  }
}

A 403 means your account is not yet approved to file cases; a 404 means the case does not exist or your key does not belong to a party on it; a 409 on a response means the case is not waiting on you — the details say whose turn it is or what to finish on the web first.