SerializationAutomation DSCSA Track & Trace

Negotiating EPCIS Capability with Trading Partners

A wholesaler tells you they are “EPCIS ready,” you configure your exchange endpoint for EPCIS 2.0 JSON-LD, and the first production batch bounces because their system only emits EPCIS 1.2 XML with none of the TransactionEvent fields your contract expects. This guide is part of Trading Partner Verification & Onboarding, the area of the DSCSA Compliance Architecture & Standards Mapping framework that governs how a new partner earns the right to exchange serialized data with you, and it solves the exact problem above: query a partner’s capabilities before you ever send a saleable-unit event, and fail fast on a documented mismatch rather than on a rejected production message.

EPCIS 2.0 formalizes this as a discoverable interface. Every conformant repository exposes a capabilities resource that answers, in a single request, which EPCIS versions it speaks, which payload formats it accepts, which event types it generates, and which business vocabulary it uses. Treating that response as a typed contract — not a document you eyeball once during onboarding — lets your ingestion pipeline reject an incompatible partner automatically instead of discovering the mismatch when a shipment is already in transit.

EPCIS capability negotiation sequence Your onboarding service requests the trading partner's EPCIS capabilities document, receives supported versions, media types, and event types, computes the negotiated profile locally, and persists the agreed profile before any production event exchange begins. Onboarding service Negotiation logic Partner /capabilities request agreed profile GET /epcis/capabilities versions, formats, event types intersect versions, formats & events Store negotiated profile per GLN

Prerequisites

  • Python 3.10+ — the snippets use X | Y unions, list[str] generics, and structural pattern matching.
  • httpx with the h2 extra installed, for an async client that speaks HTTP/2 to partner endpoints that require it.
  • Pydantic v2 — field_validator and model_validator do the negotiation math as declarative rules instead of scattered conditionals.
  • The partner’s GLN already registered in your trading-partner master data, per registering GLNs for DSCSA trading partners — capability negotiation is meaningless if you cannot yet key the agreed profile to a verified partner identity.
  • Your own capabilities, expressed in the same shape as the partner’s document, so negotiation is symmetric rather than an assumption baked into one side’s code.
  • A staging or sandbox endpoint for the partner, so negotiation and the integration tests described in VRS integration testing with Python run before any production credential is issued.

Step-by-Step Solution

Step 1 — Model the capabilities document

The EPCIS 2.0 REST bindings define a /capabilities resource that returns a JSON document describing which specification version, media types, event types, and vocabulary an endpoint supports. Model it as a typed contract so a partner that omits a required field, or advertises an event type you have never heard of, fails validation immediately rather than surfacing as a downstream KeyError.

from pydantic import BaseModel, field_validator

KNOWN_EVENT_TYPES = {
    "ObjectEvent",
    "AggregationEvent",
    "TransactionEvent",
    "TransformationEvent",
}
KNOWN_MEDIA_TYPES = {
    "application/xml",
    "application/json",
    "application/ld+json",
}


class CapabilitiesDocument(BaseModel):
    """A trading partner's EPCIS capability advertisement."""

    supported_versions: list[str]        # e.g. ["1.2", "2.0"]
    supported_media_types: list[str]     # e.g. ["application/json", "application/xml"]
    supported_event_types: list[str]     # e.g. ["ObjectEvent", "TransactionEvent"]
    cbv_version: str                     # Core Business Vocabulary version, e.g. "2.0"
    query_interface_url: str | None = None

    @field_validator("supported_versions")
    @classmethod
    def _known_versions(cls, v: list[str]) -> list[str]:
        allowed = {"1.2", "2.0"}
        if not v or not set(v).issubset(allowed):
            raise ValueError(f"unsupported EPCIS version in {v!r}")
        return v

    @field_validator("supported_media_types")
    @classmethod
    def _known_media_types(cls, v: list[str]) -> list[str]:
        # Normalize away charset/profile parameters before comparing.
        normalized = [mt.split(";")[0].strip().lower() for mt in v]
        if not set(normalized) & KNOWN_MEDIA_TYPES:
            raise ValueError(f"no recognizable media type in {v!r}")
        return normalized

    @field_validator("supported_event_types")
    @classmethod
    def _requires_object_event(cls, v: list[str]) -> list[str]:
        # ObjectEvent is the minimum viable contract — commissioning and
        # decommissioning both depend on it, so no negotiation can proceed
        # without it regardless of what else the partner supports.
        if "ObjectEvent" not in v:
            raise ValueError("partner must support ObjectEvent at minimum")
        return v

DSCSA/GS1 note: the (01) GTIN and (21) serial that identify a saleable unit only ever travel inside ObjectEvent, AggregationEvent, TransactionEvent, and TransformationEvent payloads — if a partner’s document lists none of these, there is no EPCIS exchange to negotiate at all.

Step 2 — Fetch the partner’s capabilities document asynchronously

Onboarding runs many partner negotiations concurrently, so the fetch itself should be a small, retryable async function rather than a blocking call embedded in a larger script. Keep the timeout short: a capabilities endpoint that cannot answer in a few seconds is a signal worth surfacing on its own.

import httpx


class CapabilitiesFetchError(Exception):
    """Raised when a partner's capabilities endpoint cannot be reached or parsed."""


async def fetch_partner_capabilities(
    base_url: str, *, timeout: float = 8.0
) -> dict:
    """GET a trading partner's EPCIS 2.0 capabilities document."""
    url = base_url.rstrip("/") + "/epcis/capabilities"
    try:
        async with httpx.AsyncClient(timeout=timeout, http2=True) as client:
            response = await client.get(url, headers={"Accept": "application/json"})
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as exc:
        raise CapabilitiesFetchError(f"capabilities fetch failed for {url}: {exc}") from exc

DSCSA/GS1 note: querying the live capabilities endpoint rather than relying on a partner’s onboarding questionnaire catches the common drift case — a partner upgrades their EPCIS repository mid-relationship and the questionnaire answer from a year ago is now wrong.

Step 3 — Define the negotiated profile and negotiation rule

Negotiation is a pure function: given your capabilities and the partner’s, compute the single agreed version, media type, and event-type set you will both honor. Prefer EPCIS 2.0 over 1.2 and JSON-LD over plain JSON over XML when both sides support the option, but never negotiate a version or format neither side actually advertised.

from pydantic import BaseModel, model_validator

VERSION_PREFERENCE = ["2.0", "1.2"]
MEDIA_TYPE_PREFERENCE = ["application/ld+json", "application/json", "application/xml"]


class NegotiationError(Exception):
    """Raised when no mutually supported EPCIS profile exists."""


class NegotiatedProfile(BaseModel):
    """The EPCIS exchange profile both sides have agreed to honor."""

    partner_gln: str
    epcis_version: str
    media_type: str
    event_types: list[str]
    cbv_version: str

    @model_validator(mode="after")
    def _non_empty_event_types(self) -> "NegotiatedProfile":
        if not self.event_types:
            raise ValueError("negotiated profile has no shared event types")
        return self


def negotiate_profile(
    partner_gln: str,
    local: CapabilitiesDocument,
    partner: CapabilitiesDocument,
) -> NegotiatedProfile:
    """Compute the agreed EPCIS profile from two capability documents."""
    shared_versions = set(local.supported_versions) & set(partner.supported_versions)
    version = next((v for v in VERSION_PREFERENCE if v in shared_versions), None)
    if version is None:
        raise NegotiationError(
            f"no shared EPCIS version between {local.supported_versions} "
            f"and {partner.supported_versions}"
        )

    shared_media = set(local.supported_media_types) & set(partner.supported_media_types)
    media_type = next((m for m in MEDIA_TYPE_PREFERENCE if m in shared_media), None)
    if media_type is None:
        raise NegotiationError("no shared payload media type")

    shared_events = sorted(set(local.supported_event_types) & set(partner.supported_event_types))
    if "ObjectEvent" not in shared_events:
        raise NegotiationError("shared event types do not include ObjectEvent")

    return NegotiatedProfile(
        partner_gln=partner_gln,
        epcis_version=version,
        media_type=media_type,
        event_types=shared_events,
        cbv_version=min(local.cbv_version, partner.cbv_version),
    )

DSCSA/GS1 note: falling back to EPCIS 1.2 XML is a legitimate, standards-conformant outcome, not a failure — it simply means the EPCIS 2.0 event formatting guide’s JSON-LD structures do not apply to this partner, and your ingestion pipeline must branch on negotiated.epcis_version accordingly.

Step 4 — Tie fetch and negotiation together with graceful failure

Wrap the two prior steps in one coroutine that either returns a validated NegotiatedProfile or raises a typed error your onboarding workflow can act on — surfacing a clear reason to a human reviewer rather than a stack trace.

async def negotiate_with_partner(
    partner_gln: str,
    partner_base_url: str,
    local: CapabilitiesDocument,
) -> NegotiatedProfile:
    """Fetch, validate, and negotiate an EPCIS profile with one trading partner."""
    raw = await fetch_partner_capabilities(partner_base_url)
    try:
        partner_caps = CapabilitiesDocument.model_validate(raw)
    except Exception as exc:  # Pydantic ValidationError or malformed JSON
        raise NegotiationError(
            f"partner {partner_gln} returned an invalid capabilities document: {exc}"
        ) from exc
    return negotiate_profile(partner_gln, local, partner_caps)

Step 5 — Persist the agreed profile before any production exchange

Negotiation is only useful if the outcome is durable. Store the NegotiatedProfile keyed by GLN, and treat any change to it as an onboarding event that requires re-running the negotiation and, ideally, the integration test suite in VRS integration testing with Python before production traffic resumes under the new profile.

async def onboard_partner_profile(
    partner_gln: str,
    partner_base_url: str,
    local: CapabilitiesDocument,
    profile_store,
) -> NegotiatedProfile:
    """Negotiate and persist a partner's profile, refusing to overwrite silently."""
    profile = await negotiate_with_partner(partner_gln, partner_base_url, local)
    existing = await profile_store.get(partner_gln)
    if existing is not None and existing != profile.model_dump():
        # A previously agreed profile changed — this must go through
        # documented change control, not an unattended overwrite.
        raise NegotiationError(
            f"negotiated profile for {partner_gln} changed since last onboarding; "
            "route to compliance review before storing"
        )
    await profile_store.put(partner_gln, profile.model_dump())
    return profile

Verification

Confirm the negotiation logic before it ever touches a real partner. A table-driven pytest suite covering the version, media-type, and event-type intersections gives fast, deterministic coverage without any network call:

import pytest

LOCAL = CapabilitiesDocument(
    supported_versions=["1.2", "2.0"],
    supported_media_types=["application/ld+json", "application/xml"],
    supported_event_types=["ObjectEvent", "AggregationEvent", "TransactionEvent"],
    cbv_version="2.0",
)


def test_prefers_epcis_20_and_json_ld_when_both_support_it():
    partner = CapabilitiesDocument(
        supported_versions=["2.0"],
        supported_media_types=["application/ld+json", "application/json"],
        supported_event_types=["ObjectEvent", "TransactionEvent"],
        cbv_version="2.0",
    )
    profile = negotiate_profile("0614141000005", LOCAL, partner)
    assert profile.epcis_version == "2.0"
    assert profile.media_type == "application/ld+json"
    assert profile.event_types == ["ObjectEvent", "TransactionEvent"]


def test_falls_back_to_epcis_12_xml_when_that_is_all_partner_has():
    partner = CapabilitiesDocument(
        supported_versions=["1.2"],
        supported_media_types=["application/xml"],
        supported_event_types=["ObjectEvent", "AggregationEvent"],
        cbv_version="1.2",
    )
    profile = negotiate_profile("0614141000005", LOCAL, partner)
    assert profile.epcis_version == "1.2"
    assert profile.media_type == "application/xml"


def test_raises_when_no_shared_version_exists():
    partner = CapabilitiesDocument(
        supported_versions=["1.2"],
        supported_media_types=["application/xml"],
        supported_event_types=["ObjectEvent"],
        cbv_version="1.2",
    )
    local_20_only = LOCAL.model_copy(update={"supported_versions": ["2.0"]})
    with pytest.raises(NegotiationError):
        negotiate_profile("0614141000005", local_20_only, partner)

Beyond unit tests, run the negotiation coroutine against the partner’s sandbox /epcis/capabilities endpoint and diff the resulting NegotiatedProfile against the profile stored from the last onboarding cycle — any unexpected change should block production credential issuance until compliance signs off, exactly as onboard_partner_profile enforces above.

Gotchas & Edge Cases

  • A partner claims EPCIS 2.0 but only exposes the 1.2 query interface. The capabilities document is a self-reported advertisement, not a guarantee. Confirm the negotiated profile against a real sandbox call before trusting it for production — see the EPCIS 1.2 vs 2.0 migration checklist for the specific fields that differ between the two versions.
  • Media type strings with parameters. application/ld+json; charset=utf-8 and application/ld+json must negotiate as the same type; comparing raw strings without stripping parameters silently forces an XML fallback that neither side intended.
  • Stale cached capabilities. A partner that upgrades their repository mid-relationship invalidates any cached negotiation result. Give cached capabilities documents a short TTL and re-negotiate on a schedule, not only at initial onboarding.
  • Case and whitespace drift in event-type names. Some partners emit objectevent or Object_Event in non-conformant integrations. Normalize case before intersecting sets, or a real shared capability will be missed and negotiation will fail for the wrong reason.
  • Treating negotiation as a one-time gate. A profile agreed six months ago is a compliance record, not a cache entry to overwrite quietly. Any drift between a freshly negotiated profile and the stored one belongs in change control, exactly as it would for any other onboarding artifact tied to the partner’s registered GLN.

Continue reading

Related Registering Partner GLNs Related VRS Integration Testing