# Validating parsed RFP JSON against agency schemas

Once a solicitation has been read off the page and turned into JSON, that payload is still untrusted: a parser can emit a syntactically perfect object that violates the data contract the National Institutes of Health (NIH), the National Science Foundation (NSF), or the Department of Defense (DoD) will actually enforce at submission time. A missing compliance flag, a string where an aware timestamp belongs, or an activity code that fails a regular-expression check will pass shallow key-presence tests and then surface as a hard rejection inside eRA Commons or Grants.gov — after the funding cycle’s deadline has closed. This page is a step-by-step reference for building the deterministic validation gate that sits at the end of schema validation with Pydantic: it takes the JSON produced upstream and either promotes it to a strongly typed, agency-conformant record or fails loudly with an auditable reason. The rule is uniform — no payload reaches the document assembler until it has survived this boundary.

The failure mode this page prevents is silent structural drift: the parser succeeds, the JSON looks plausible, and the defect is only discovered by the agency portal. Federal contracts are far stricter than key-value presence. NIH distinguishes modular from detailed budget architectures; NSF requires validated program-director identifiers and specific activity-code enumerations; DoD announcements embed security-classification tags and export-control restrictions that must be explicitly typed and non-nullable. Encoding those rules as executable Pydantic models turns an ambiguous blob into an object that fails fast at parse time.

# Phase 1 — Decompose the payload and pin the agency contract

Before writing a single validator, separate the universal federal fields (present on every solicitation regardless of agency) from the agency-specific fields that only one funder enforces. This decomposition is what lets a base model carry the common contract while thin subclasses add the divergent rules, and it mirrors the section-scoping the upstream NLP section boundary detection stage already performed on the raw text.

Implementation steps:

1. Identify the agency early. The parsed JSON must carry a discriminator — an agency field set to "NIH", "NSF", or "DoD" — populated during extraction. Without it, the validator cannot choose the right subclass, so treat a missing discriminator as an immediate routed-review case, never a default.
2. Catalogue the universal fields. Every federal opportunity carries an opportunity_id, a Catalog of Federal Domestic Assistance (cfda_number), and a submission_deadline_utc. These belong on the base model.
3. Catalogue the divergent fields. NIH adds budget_modular and the conditional PHS 398 fields; NSF adds program_director_id and Proposal & Award Policies & Procedures Guide (PAPPG) activity codes; DoD adds export_control_restricted and security_classification_level.
4. Pin the environment. Lock the interpreter and library versions so a payload validated today reproduces byte-for-byte when it is re-checked after an agency amendment.

python -m venv .venv
source .venv/bin/activate
pip install "pydantic>=2.6"

# Phase 2 — Model the agency contract and validate the JSON atomically

The core transformation is to express each contract as a Pydantic v2 BaseModel, then validate the raw string with model_validate_json() rather than unpacking a dictionary. Parsing and validating in one call means the entire structure is evaluated atomically: any deviation raises a single ValidationError carrying every fault, instead of a cascade of KeyErrors discovered one at a time.

Use StrictStr, StrictInt, and AwareDatetime to forbid the implicit type coercion that lets "true" masquerade as a boolean or a naive timestamp slip past a deadline check. Set ConfigDict(strict=True, extra="forbid") so an undocumented field — often the first sign the parser drifted against a revised notice — is rejected rather than silently accepted.

from datetime import datetime
from pydantic import (
    AwareDatetime,
    BaseModel,
    ConfigDict,
    StrictBool,
    StrictStr,
    field_validator,
    model_validator,
)


class FederalRFPBase(BaseModel):
    """Universal contract every federal solicitation payload must satisfy."""

    model_config = ConfigDict(strict=True, extra="forbid")

    agency: StrictStr
    opportunity_id: StrictStr
    cfda_number: StrictStr
    issue_date: AwareDatetime
    submission_deadline_utc: AwareDatetime

    @field_validator("cfda_number")
    @classmethod
    def _cfda_shape(cls, v: str) -> str:
        # CFDA numbers are two digits, a dot, three digits (e.g. 93.279).
        import re

        if not re.fullmatch(r"\d{2}\.\d{3}", v):
            raise ValueError(f"malformed CFDA number: {v!r}")
        return v

    @model_validator(mode="after")
    def _deadline_after_issue(self) -> "FederalRFPBase":
        if self.submission_deadline_utc <= self.issue_date:
            raise ValueError("submission_deadline_utc must be strictly after issue_date")
        return self


class NIHRFP(FederalRFPBase):
    """NIH adds modular-vs-detailed budget branching and activity-code rules."""

    activity_code: StrictStr
    budget_modular: StrictBool
    phs_398_fields: dict[str, str] | None = None

    @field_validator("activity_code")
    @classmethod
    def _known_activity_code(cls, v: str) -> str:
        import re

        if not re.fullmatch(r"R01|R21|K99|U01|P01", v):
            raise ValueError(f"unrecognized NIH activity code: {v!r}")
        return v

    @model_validator(mode="after")
    def _detailed_budget_requires_phs398(self) -> "NIHRFP":
        # Detailed budgets (modular = False) MUST carry the PHS 398 line items.
        if not self.budget_modular and not self.phs_398_fields:
            raise ValueError("detailed NIH budget requires phs_398_fields")
        return self

Instantiation is now a single call that either returns a typed object or raises:

def validate_payload(raw_json: str, agency: str) -> FederalRFPBase:
    """Route raw parsed JSON to the correct agency model and validate atomically."""
    model = {"NIH": NIHRFP, "NSF": NSFRFP, "DoD": DoDRFP}.get(agency)
    if model is None:
        raise ValueError(f"no schema registered for agency {agency!r}")
    return model.model_validate_json(raw_json)  # raises ValidationError on any drift

The diagram below illustrates the conditional branching that the agency-specific validators enforce at parse time.

# Phase 3 — Edge cases and agency-specific overrides

The base model handles the common case; production breaks on the conditional rules that only one agency enforces, and on the fact that those rules are versioned. Two override classes recur.

Mutually exclusive and conditionally required fields. DoD announcements gate security_classification_level on export_control_restricted: when a solicitation is flagged under the International Traffic in Arms Regulations (ITAR) or the Export Administration Regulations (EAR), the classification level cannot be null. NSF gates activity-code enumerations on the PAPPG version in force. Encode these as @model_validator(mode="after") assertions so a contradictory payload never escapes.

class DoDRFP(FederalRFPBase):
    """DoD adds export-control gating and non-nullable classification."""

    export_control_restricted: StrictBool
    security_classification_level: StrictStr | None = None

    @model_validator(mode="after")
    def _restricted_requires_classification(self) -> "DoDRFP":
        if self.export_control_restricted and self.security_classification_level is None:
            raise ValueError(
                "export_control_restricted=True requires a security_classification_level"
            )
        return self


class NSFRFP(FederalRFPBase):
    program_director_id: StrictStr
    pappg_version: StrictStr  # e.g. "24-1"

Versioned policy drift. A payload is only conformant against the edition of the notice that was in force when it was parsed. Agency guidelines change annually, so tag every schema with the policy cycle it encodes — v2024.1-nsf-pappg, v2024-nih-activity-codes — and read the authoritative enumerations from the NIH FOA schema mapping and DoD BAA requirement extraction records rather than hard-coding them. The per-mechanism thresholds themselves are owned by the compliance threshold tuning layer, so a policy change is a data edit, not a code change. The differences worth encoding:

# Phase 4 — Validation, verification, and an audit-safe record

A rejected payload is worth nothing to a research administrator unless the rejection is traceable: which field, which expected type, which received value, which clause. Wrap instantiation in a try/except ValidationError, extract e.errors(), and transform the raw error list into a standardized report that maps each loc tuple to a human-readable field name before it reaches a ticket queue.

import hashlib
import json
from datetime import datetime, timezone

from pydantic import ValidationError

FIELD_LABELS: dict[tuple[str, ...], str] = {
    ("phs_398_fields",): "PHS 398 Budget Detail",
    ("security_classification_level",): "Security Classification Level",
    ("submission_deadline_utc",): "Submission Deadline (UTC)",
}


def compliance_drift_report(raw_json: str, agency: str) -> dict[str, object]:
    """Validate a payload and emit a reproducible, coordinate-free audit record."""
    digest = hashlib.sha256(raw_json.encode("utf-8")).hexdigest()
    base = {
        "payload_sha256": digest,
        "agency": agency,
        "checked_at": datetime.now(timezone.utc).isoformat(),
        "schema_version": f"{agency.lower()}_v2024.1",
    }
    try:
        validate_payload(raw_json, agency)
    except ValidationError as exc:
        return base | {
            "validation_status": "REJECTED",
            "errors": [
                {
                    "field": FIELD_LABELS.get(tuple(e["loc"]), ".".join(map(str, e["loc"]))),
                    "type": e["type"],
                    "message": e["msg"],
                    "input": e.get("input"),
                }
                for e in exc.errors()
            ],
        }
    return base | {"validation_status": "ACCEPTED", "errors": []}

A payload flagged REJECTED is serialized to the immutable compliance log with its hash and timestamp — a legally defensible trail for institutional review boards and sponsored-programs offices — and routed to a human-review queue. An ACCEPTED payload is released to the document assembler and, downstream, to the compliance validation rule engines that enforce the page-limit and font rules on the assembled output.

Confirm the gate against committed fixtures rather than live agency JSON, which changes without notice:

import pytest


def test_detailed_nih_budget_without_phs398_is_rejected() -> None:
    payload = json.dumps({
        "agency": "NIH", "opportunity_id": "PA-24-001", "cfda_number": "93.279",
        "issue_date": "2024-01-01T00:00:00Z", "submission_deadline_utc": "2024-06-01T17:00:00Z",
        "activity_code": "R01", "budget_modular": False, "phs_398_fields": None,
    })
    report = compliance_drift_report(payload, "NIH")
    assert report["validation_status"] == "REJECTED"
    assert any(e["field"] == "PHS 398 Budget Detail" for e in report["errors"])


def test_restricted_dod_without_classification_is_rejected() -> None:
    payload = json.dumps({
        "agency": "DoD", "opportunity_id": "HR001124S0001", "cfda_number": "12.910",
        "issue_date": "2024-01-01T00:00:00Z", "submission_deadline_utc": "2024-03-01T17:00:00Z",
        "export_control_restricted": True, "security_classification_level": None,
    })
    report = compliance_drift_report(payload, "DoD")
    assert report["validation_status"] == "REJECTED"


def test_conformant_nsf_payload_is_accepted() -> None:
    payload = json.dumps({
        "agency": "NSF", "opportunity_id": "NSF-24-500", "cfda_number": "47.070",
        "issue_date": "2024-01-01T00:00:00Z", "submission_deadline_utc": "2024-08-01T17:00:00Z",
        "program_director_id": "pd-40912", "pappg_version": "24-1",
    })
    assert compliance_drift_report(payload, "NSF")["validation_status"] == "ACCEPTED"

A manual acceptance checklist catches what fixtures miss: confirm every rejection carries a human-readable field label, that extra="forbid" actually trips on an undocumented key, that the schema version tag matches the policy cycle the payload was parsed under, and that a payload with a missing agency discriminator routes to review instead of defaulting. At high volume — a full funding cycle of hundreds of solicitations — run the gate under the async batch processor for large RFPs so validation never becomes the pipeline bottleneck. Only once these pass should the accepted record advance to assembly.

# Frequently asked questions

# Related

• Schema validation with Pydantic — the parent workflow stage this gate completes.
• PDF text extraction with pdfplumber — the coordinate-aware source of the JSON this stage validates.
• NLP section boundary detection — segments the extracted text before it is modeled.
• Async batch processing for large RFPs — running the validation gate across a full funding cycle.
• Compliance validation rule engines — where accepted payloads are enforced against page-limit, font, and section rules.

Up one level: Schema validation with Pydantic