Carrier Dispute API Integration

Once a variance has been proven and priced, the audit is only half finished. The dispute still has to reach the carrier, be acknowledged, and be tracked to a settled or denied state — and every carrier exposes that lifecycle through a different HTTP surface. One carrier offers an OAuth2 claims API that returns a claim_id synchronously; another wants an API key in a header and reports outcomes only through webhooks; a TMS gateway accepts a batch of disputes but rate-limits to a handful of requests per second. This layer is the tier that turns a classified variance into a submitted claim, keeps that claim’s state in sync with the carrier of record, and hands the resolved outcome to the settlement ledger. Where Charge Variance Classification decides what is wrong and by how much, this tier is responsible for filing it and following it.

The failure mode of doing this badly is expensive and quiet. A brittle integration double-files claims after a retry, drops disputes when a token expires mid-batch, or silently accepts a 422 schema rejection as if it were a success. Weeks later a reconciliation shows recovered dollars that the carrier never actually credited, or a stack of disputes that were never filed at all. The design goal here is a deterministic, idempotent submission-and-tracking layer where every dispute has exactly one authoritative state, every carrier lives behind a uniform adapter, and every terminal outcome flows into the Reconciliation Ledger Schema as an auditable event.

Scope and where this sits

This integration layer owns the boundary between the internal audit engine and the carrier’s dispute or claims API. It does not classify variance, it does not compute the disputed amount, and it does not adjudicate the outcome — carriers do that. Its responsibilities are narrow and mechanical: authenticate to each carrier, map a classified variance onto that carrier’s claim schema, submit the claim idempotently, poll or receive its status, and reconcile the returned state back into the ledger. Everything upstream produces a ClassifiedDispute; everything downstream reads a DisputeRecord whose status this tier keeps current.

Prerequisites

Before wiring any carrier adapter, the following upstream components and configuration must exist. A missing prerequisite here surfaces as a runtime KeyError mid-batch, so validate them at startup rather than per-request.

Python dependencies are pinned so CI and production agree exactly:

# requirements.txt
httpx==0.28.1
tenacity==9.0.0
pydantic==2.10.6
structlog==24.4.0

The two config keys every adapter reads — the auth mode and the per-carrier rate ceiling — belong in the settings layer, not in adapter code:

Adapter and field-mapping architecture

The central design decision is that no calling code ever talks to a carrier directly. A classified dispute is handed to an adapter registry, which resolves the carrier_scac to a concrete client that knows that carrier’s auth scheme, URL shape, and claim schema. Every client implements the same abstract interface, so submission, polling, and reconciliation are written once against the interface and reused across every carrier. Adding a carrier means writing one adapter, not touching the pipeline.

The field-mapping table each adapter owns is the contract between the internal variance model and the carrier’s wire format. The abstract fields are stable; the per-carrier target column is what changes.

Step-by-step implementation

The implementation is built bottom-up: an abstract client that fixes the lifecycle contract, a concrete adapter that fills in one carrier’s specifics, and a registry that selects between them. Retries and backoff live in the transport layer so every adapter inherits them uniformly.

Step 1 — Define the abstract client and the domain models

The abstract CarrierDisputeClient fixes the four operations every carrier must support: authenticate, map, submit, and fetch status. Concrete adapters override the carrier-specific pieces and inherit the shared submit/poll machinery.

# dispute/base.py
from __future__ import annotations
import abc
from decimal import Decimal
from enum import Enum
from typing import Optional
import httpx
from pydantic import BaseModel


class DisputeStatus(str, Enum):
    SUBMITTED = "SUBMITTED"
    PENDING = "PENDING"
    APPROVED = "APPROVED"
    PARTIAL = "PARTIAL"
    DENIED = "DENIED"
    ERROR = "ERROR"


class ClassifiedDispute(BaseModel):
    dispute_id: str            # our stable id, unique per variance
    invoice_number: str
    carrier_scac: str
    variance_type: str         # e.g. OVERCHARGE, DUPLICATE_BILL, FSC_DRIFT
    disputed_amount: Decimal
    evidence_refs: list[str] = []


class DisputeResult(BaseModel):
    dispute_id: str
    carrier_claim_id: Optional[str] = None
    status: DisputeStatus
    detail: Optional[str] = None


class CarrierDisputeClient(abc.ABC):
    """One instance per carrier. Adapters implement the carrier-specific parts."""

    scac: str

    def __init__(self, transport: httpx.Client):
        # The transport is injected so tests can swap in a mock (see testing).
        self._http = transport

    @abc.abstractmethod
    def build_claim(self, dispute: ClassifiedDispute) -> dict:
        """Map the internal variance onto this carrier's claim schema."""

    @abc.abstractmethod
    def submit(self, dispute: ClassifiedDispute) -> DisputeResult:
        """POST the claim and return the carrier claim id + status."""

    @abc.abstractmethod
    def fetch_status(self, carrier_claim_id: str) -> DisputeStatus:
        """Poll the carrier for the current lifecycle state of a claim."""

Common mistake: letting each adapter invent its own status vocabulary. Normalize every carrier’s raw state string into the shared DisputeStatus enum inside the adapter, so the ledger and every consumer downstream see one closed set of terminal states rather than a dozen carrier-specific strings.

Step 2 — Wrap the transport with retry and backoff

Transient failures — a 429 throttle, a 503, a dropped connection — must be retried with exponential backoff and jitter, but non-transient failures (422, 409) must not be retried, because retrying a schema rejection just burns rate budget and retrying a duplicate re-triggers the same conflict. The retry predicate encodes exactly which HTTP outcomes are worth another attempt.

# dispute/transport.py
import httpx
from tenacity import (
    retry, stop_after_attempt, wait_exponential_jitter,
    retry_if_exception_type,
)


class RetryableError(Exception):
    """Transient failure worth another attempt (429, 5xx, network)."""


class PermanentError(Exception):
    """Non-transient failure — do not retry (422 schema, 409 duplicate)."""
    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail
        super().__init__(f"{status_code}: {detail}")


def classify_response(resp: httpx.Response) -> httpx.Response:
    """Turn HTTP status into the right exception class for the retry policy."""
    if resp.status_code == 429 or resp.status_code >= 500:
        raise RetryableError(f"transient {resp.status_code}")
    if resp.status_code in (409, 422, 400):
        # 401 is handled separately by the auth layer (token refresh).
        raise PermanentError(resp.status_code, resp.text[:200])
    resp.raise_for_status()
    return resp


@retry(
    retry=retry_if_exception_type(RetryableError),
    wait=wait_exponential_jitter(initial=0.5, max=30),
    stop=stop_after_attempt(5),
    reraise=True,
)
def post_with_retry(http: httpx.Client, url: str, *, json: dict, headers: dict) -> httpx.Response:
    resp = http.post(url, json=json, headers=headers, timeout=15.0)
    return classify_response(resp)

Common mistake: retrying on every non-2xx. A retried 409 duplicate will conflict on every attempt and a retried 422 will fail identically five times — both waste rate-limit budget that a genuinely transient 429 needs. Split the outcomes into RetryableError and PermanentError and let the policy act only on the former.

Step 3 — Implement a concrete carrier adapter

The concrete adapter binds the abstract contract to one carrier’s reality: an OAuth2 token it refreshes on demand, an Idempotency-Key header derived from the stable dispute_id, and a translation from the internal variance_type to the carrier’s reason codes.

# dispute/adapters/abcd.py
import time
import httpx
import structlog
from decimal import Decimal
from dispute.base import (
    CarrierDisputeClient, ClassifiedDispute, DisputeResult, DisputeStatus,
)
from dispute.transport import post_with_retry, PermanentError

log = structlog.get_logger()

# Internal variance types → this carrier's claim reason codes.
_REASON_MAP = {
    "OVERCHARGE": "RATE_DISCREPANCY",
    "DUPLICATE_BILL": "DUPLICATE_INVOICE",
    "FSC_DRIFT": "FUEL_SURCHARGE",
    "ACCESSORIAL": "ACCESSORIAL_DISPUTE",
}

_STATUS_MAP = {
    "open": DisputeStatus.PENDING,
    "under_review": DisputeStatus.PENDING,
    "granted": DisputeStatus.APPROVED,
    "partially_granted": DisputeStatus.PARTIAL,
    "rejected": DisputeStatus.DENIED,
}


class AbcdDisputeAdapter(CarrierDisputeClient):
    scac = "ABCD"

    def __init__(self, transport: httpx.Client, base_url: str, oauth):
        super().__init__(transport)
        self._base = base_url.rstrip("/")
        self._oauth = oauth            # supplies + refreshes bearer tokens

    def build_claim(self, dispute: ClassifiedDispute) -> dict:
        reason = _REASON_MAP.get(dispute.variance_type)
        if reason is None:
            # Fail loud: an unmapped variance would silently misfile the claim.
            raise PermanentError(422, f"unmapped variance_type {dispute.variance_type}")
        return {
            "externalReference": dispute.dispute_id,
            "invoiceNo": dispute.invoice_number,
            "claimReasonCode": reason,
            # Decimal → fixed-precision string; never a float on the wire.
            "amountClaimed": f"{dispute.disputed_amount:.2f}",
            "attachments": dispute.evidence_refs,
        }

    def submit(self, dispute: ClassifiedDispute) -> DisputeResult:
        body = self.build_claim(dispute)
        headers = {
            "Authorization": f"Bearer {self._oauth.token()}",
            # Stable key → carrier dedups a resubmission to the same claim.
            "Idempotency-Key": dispute.dispute_id,
        }
        try:
            resp = post_with_retry(
                self._http, f"{self._base}/v2/claims", json=body, headers=headers,
            )
        except PermanentError as exc:
            if exc.status_code == 409:
                # Carrier already has this claim — treat as an idempotent hit,
                # not a failure. Recover the existing id and keep going.
                return self._recover_existing(dispute)
            log.error("dispute_submit_permanent", scac=self.scac,
                      dispute_id=dispute.dispute_id, code=exc.status_code)
            return DisputeResult(dispute_id=dispute.dispute_id,
                                 status=DisputeStatus.ERROR, detail=exc.detail)
        payload = resp.json()
        return DisputeResult(
            dispute_id=dispute.dispute_id,
            carrier_claim_id=payload["claimId"],
            status=DisputeStatus.SUBMITTED,
        )

    def _recover_existing(self, dispute: ClassifiedDispute) -> DisputeResult:
        headers = {"Authorization": f"Bearer {self._oauth.token()}"}
        resp = self._http.get(
            f"{self._base}/v2/claims",
            params={"externalReference": dispute.dispute_id},
            headers=headers, timeout=15.0,
        )
        claim = resp.json()["items"][0]
        return DisputeResult(
            dispute_id=dispute.dispute_id,
            carrier_claim_id=claim["claimId"],
            status=self.fetch_status(claim["claimId"]),
        )

    def fetch_status(self, carrier_claim_id: str) -> DisputeStatus:
        headers = {"Authorization": f"Bearer {self._oauth.token()}"}
        resp = self._http.get(
            f"{self._base}/v2/claims/{carrier_claim_id}",
            headers=headers, timeout=15.0,
        )
        raw = resp.json().get("state", "").lower()
        return _STATUS_MAP.get(raw, DisputeStatus.PENDING)

Common mistake: treating a 409 Conflict on submit as an error. When a claim already exists — because a prior run submitted it before crashing — the correct behaviour is to recover the existing claim_id and reconcile its current status, exactly as _recover_existing does. Re-raising the conflict marks a filed dispute as failed and re-files it under a new reference.

Step 4 — Register adapters and route by SCAC

The registry is the only place that knows the full set of carriers. Callers ask it for a client by SCAC and get back something that satisfies the abstract contract; they never import a concrete adapter.

# dispute/registry.py
from dispute.base import CarrierDisputeClient, ClassifiedDispute, DisputeResult


class AdapterRegistry:
    def __init__(self):
        self._by_scac: dict[str, CarrierDisputeClient] = {}

    def register(self, client: CarrierDisputeClient) -> None:
        self._by_scac[client.scac.upper()] = client

    def resolve(self, scac: str) -> CarrierDisputeClient:
        client = self._by_scac.get(scac.upper())
        if client is None:
            raise KeyError(f"no dispute adapter registered for SCAC {scac}")
        return client

    def submit(self, dispute: ClassifiedDispute) -> DisputeResult:
        return self.resolve(dispute.carrier_scac).submit(dispute)

Validation and testing

Because every adapter takes its httpx transport by injection, the whole submission path is testable without a live carrier. A MockTransport maps request URLs to canned responses, letting you assert that the adapter maps fields correctly, sends the idempotency header, and normalizes each carrier status string into the shared enum.

# tests/test_abcd_adapter.py
import httpx
from decimal import Decimal
from dispute.base import ClassifiedDispute, DisputeStatus
from dispute.adapters.abcd import AbcdDisputeAdapter


class _FakeOAuth:
    def token(self) -> str:
        return "test-token"


def _dispute() -> ClassifiedDispute:
    return ClassifiedDispute(
        dispute_id="D-1001", invoice_number="INV-77", carrier_scac="ABCD",
        variance_type="OVERCHARGE", disputed_amount=Decimal("214.50"),
        evidence_refs=["s3://audit/D-1001.pdf"],
    )


def test_submit_maps_and_returns_claim_id():
    seen = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["idem"] = request.headers.get("Idempotency-Key")
        seen["body"] = request.read()
        return httpx.Response(201, json={"claimId": "ABCD-555"})

    transport = httpx.MockTransport(handler)
    client = httpx.Client(transport=transport)
    adapter = AbcdDisputeAdapter(client, "https://api.abcd.test", _FakeOAuth())

    result = adapter.submit(_dispute())

    assert result.carrier_claim_id == "ABCD-555"
    assert result.status is DisputeStatus.SUBMITTED
    assert seen["idem"] == "D-1001"           # stable idempotency key sent
    assert b"RATE_DISCREPANCY" in seen["body"]  # variance_type translated


def test_409_recovers_existing_claim():
    def handler(request: httpx.Request) -> httpx.Response:
        if request.method == "POST":
            return httpx.Response(409, text="duplicate claim")
        if request.url.path.endswith("/claims"):
            return httpx.Response(200, json={"items": [{"claimId": "ABCD-9"}]})
        return httpx.Response(200, json={"state": "under_review"})

    transport = httpx.MockTransport(handler)
    adapter = AbcdDisputeAdapter(httpx.Client(transport=transport),
                                 "https://api.abcd.test", _FakeOAuth())

    result = adapter.submit(_dispute())
    assert result.carrier_claim_id == "ABCD-9"
    assert result.status is DisputeStatus.PENDING

The fixture matrix worth keeping covers one case per failure mode: a clean 201, a 409 that recovers, a 422 that returns ERROR without retrying, a 429 that succeeds on the second attempt, and an expired-token 401 that triggers a refresh. Each asserts on the resulting DisputeStatus, so a regression in the branching logic fails a test rather than misfiling a real claim.

Performance and tuning

Throughput here is bounded by the carrier, not by your code — every carrier publishes a request ceiling, and exceeding it earns 429s that cost more in backoff than the requests you saved. The tuning model is a per-carrier client-side rate limiter that keeps you just under the published ceiling, plus a bounded concurrency pool so one slow carrier does not stall the others.

As a rough benchmark, a single worker submitting against a carrier that permits 5 requests per second clears a few hundred disputes per minute; the bottleneck is almost always the carrier’s throttle, which is why webhooks are strongly preferred over polling for status — they remove the entire poll loop from the rate budget.

Failure modes

Four HTTP-level failures account for nearly every production incident in this tier. Each has a stable signature and a specific handling rule; the wrong reflex on any of them corrupts the dispute’s authoritative state.

The 401 case is the one most often mishandled: teams either fail the whole batch or, worse, retry the request with the same expired token and burn all attempts. The correct handling is a token-refresh hook that runs on the first 401, swaps in a fresh bearer, and replays only the failed request. The end-to-end mechanics of idempotent submission under exactly these conditions are covered in Submitting Freight Claims to Carrier Dispute APIs.

Integration points

The output of this tier is a DisputeResult whose status is the authoritative state of the dispute, and every state transition is written to the Reconciliation Ledger Schema as an event. The ledger is where a submitted dispute, its carrier claim_id, and its terminal outcome — approved, partial, or denied — become an auditable record that AP and settlement read from. A partial or approved outcome carries the credited amount, which reconciles against the carrier’s remittance; a denied outcome closes the variance without recovery. The field contract handed downstream is deliberately small:

Status is kept current either by the poll loop or by inbound webhooks; the correctness rules for handling those re-delivered, out-of-order callbacks are the subject of Handling Dispute Webhook Callbacks Idempotently.

In this section

• Submitting Freight Claims to Carrier Dispute APIs — how to make batch submission idempotent so partial failures, duplicate 409s, schema 422s, and mid-run token expiry never double-file or silently drop a claim.
• Handling Dispute Webhook Callbacks Idempotently — verifying, de-duplicating, and state-guarding re-delivered status webhooks so an out-of-order callback never double-applies a credit or regresses a closed dispute.

Related

• Discrepancy Resolution & Dispute Routing — the parent tier that classifies, files, and settles freight billing variance.
• Charge Variance Classification — produces the ClassifiedDispute this integration files against the carrier.
• Reconciliation Ledger Schema — the double-entry ledger that records every submission and terminal outcome as an auditable event.
• Wiring Webhook Endpoints for Real-Time Invoice Intake — the inbound webhook plumbing this tier reuses to receive dispute-status callbacks.
• Threshold Tuning & Alerting — alerts on dead-letter growth and dispute-submission failure rates.

---

Up: Discrepancy Resolution & Dispute Routing