diff --git a/applications/abri/__init__.py b/applications/abri/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/applications/abri/abri_trigger_request.py b/applications/abri/abri_trigger_request.py
new file mode 100644
index 000000000..d159105c7
--- /dev/null
+++ b/applications/abri/abri_trigger_request.py
@@ -0,0 +1,28 @@
+"""The message contract between the HubSpot scraper and the abri lambda.
+
+One message per scrape, naming every Abri flow that fired plus the deal
+fields those flows need. A message missing a field its flow needs fails
+validation — the resulting failed task / DLQ entry is the agreed
+visibility surface.
+"""
+
+from datetime import date
+from typing import Any, List, Literal, Optional
+
+from pydantic import BaseModel, ConfigDict
+
+AbriFlow = Literal["amend_job", "log_job", "sync_tenant_data"]
+
+
+class AbriTriggerRequest(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    hubspot_deal_id: str
+    flows: List[AbriFlow]
+    place_ref: Optional[str] = None
+    deal_name: Optional[str] = None
+    confirmed_survey_date: Optional[date] = None
+    confirmed_survey_time: Optional[str] = None
+
+    def model_post_init(self, __context: Any) -> None:
+        raise NotImplementedError
diff --git a/tests/applications/abri/__init__.py b/tests/applications/abri/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/applications/abri/test_abri_trigger_request.py b/tests/applications/abri/test_abri_trigger_request.py
new file mode 100644
index 000000000..9e9672a2b
--- /dev/null
+++ b/tests/applications/abri/test_abri_trigger_request.py
@@ -0,0 +1,103 @@
+from datetime import date
+from typing import Any, Dict
+
+import pytest
+from pydantic import ValidationError
+
+from applications.abri.abri_trigger_request import AbriTriggerRequest
+
+
+def _full_message() -> Dict[str, Any]:
+    return {
+        "hubspot_deal_id": "9876543210",
+        "flows": ["amend_job", "log_job", "sync_tenant_data"],
+        "place_ref": "1007165",
+        "deal_name": "49 Admers Crescent",
+        "confirmed_survey_date": "2026-06-24",
+        "confirmed_survey_time": "14:30",
+    }
+
+
+def test_a_full_message_parses_into_a_typed_request() -> None:
+    # Act
+    request = AbriTriggerRequest.model_validate(_full_message())
+
+    # Assert
+    assert request.hubspot_deal_id == "9876543210"
+    assert request.flows == ["amend_job", "log_job", "sync_tenant_data"]
+    assert request.place_ref == "1007165"
+    assert request.deal_name == "49 Admers Crescent"
+    assert request.confirmed_survey_date == date(2026, 6, 24)
+    assert request.confirmed_survey_time == "14:30"
+
+
+@pytest.mark.parametrize(
+    ("flow", "missing_field"),
+    [
+        ("log_job", "place_ref"),
+        ("log_job", "deal_name"),
+        ("log_job", "confirmed_survey_date"),
+        ("amend_job", "confirmed_survey_date"),
+        ("sync_tenant_data", "place_ref"),
+    ],
+)
+def test_a_message_missing_a_field_its_flow_needs_fails_validation(
+    flow: str, missing_field: str
+) -> None:
+    # Arrange
+    message = _full_message()
+    message["flows"] = [flow]
+    del message[missing_field]
+
+    # Act / Assert
+    with pytest.raises(ValidationError, match=missing_field):
+        AbriTriggerRequest.model_validate(message)
+
+
+def test_a_message_missing_only_fields_of_unfired_flows_parses() -> None:
+    # Arrange: tenant sync needs only the place_ref (and the deal id).
+    message = {
+        "hubspot_deal_id": "9876543210",
+        "flows": ["sync_tenant_data"],
+        "place_ref": "1007165",
+    }
+
+    # Act
+    request = AbriTriggerRequest.model_validate(message)
+
+    # Assert
+    assert request.flows == ["sync_tenant_data"]
+    assert request.confirmed_survey_date is None
+
+
+def test_a_message_without_a_confirmed_time_parses_for_logging() -> None:
+    # Arrange: the time slot is optional — no time means an all-day slot.
+    message = _full_message()
+    message["flows"] = ["log_job"]
+    del message["confirmed_survey_time"]
+
+    # Act
+    request = AbriTriggerRequest.model_validate(message)
+
+    # Assert
+    assert request.confirmed_survey_time is None
+
+
+def test_a_message_naming_an_unknown_flow_fails_validation() -> None:
+    # Arrange
+    message = _full_message()
+    message["flows"] = ["cancel_job"]
+
+    # Act / Assert
+    with pytest.raises(ValidationError):
+        AbriTriggerRequest.model_validate(message)
+
+
+def test_a_message_naming_no_flows_fails_validation() -> None:
+    # Arrange
+    message = _full_message()
+    message["flows"] = []
+
+    # Act / Assert
+    with pytest.raises(ValidationError):
+        AbriTriggerRequest.model_validate(message)