Provider pattern

A provider is a class that turns some external seller catalogue into a list of SellerRecord objects. enrichment-scraper has one — SmartScoutProvider — and the abstraction exists so a second can be added without touching the dispatcher, the registry, the queue service, or any existing provider.

The split is deliberate: everything that is the same for every HTTP seller API lives in the base class, and everything that is specific to one vendor lives in the subclass.

Owner Concern

BaseProvider

Retries, timeout backoff, 401 re-authentication, 429 throttling, 5xx handling, proactive rate-limit sleeping, and safe row mapping

The subclass

Request headers, pagination, request-body shape, and mapping one raw row to a SellerRecord

BaseProvider

lambdas/enrichment_scraper/providers/base.py. Generic over the raw row type, so a subclass declares the Pydantic model its API returns:

RawSellerT = TypeVar("RawSellerT", bound=BaseModel)

class BaseProvider(ABC, Generic[RawSellerT]):
    ...

class SmartScoutProvider(BaseProvider[SmartScoutSeller]):
    ...

Abstract methods

Method Contract

fetch_sellers(filters) → list[SellerRecord]

Paginate the source and return matching records.

_to_seller_record(raw) → SellerRecord

Map one raw API row to the domain model. May raise ValidationError.

_build_request_headers() → dict[str, str]

Full headers for every outbound request — auth, content type, fingerprint, tracing ids.

SmartScoutProvider widens the fetch_sellers signature to fetch_sellers(filters, start_from_row=0) → tuple[list[SellerRecord], int], returning the next cursor alongside the records. The base class’s narrower declaration is the older contract; the tuple form is what the handler actually calls, because discovery has to resume across scheduled runs. A second provider should follow the tuple form.

Overridable hook

_on_unauthorized() is called on HTTP 401 before the single retry. The default is a no-op; SmartScoutProvider overrides it to call self._auth.invalidate().

This is the seam that makes token refresh work without the base class knowing anything about tokens. _post_with_retry() rebuilds headers on every attempt, so invalidating the cache in the hook is enough — the retry picks up a fresh token through _build_request_headers().

Inherited methods

Method Behaviour

map_seller(raw)

Wraps _to_seller_record(), catching ValidationError, logging it, and returning None so the caller can skip the row.

_post_with_retry(url, body, retry_count=0)

POST with the full failure matrix below.

_handle_rate_limit_headers(response)

Sleeps until the reset time when X-Rate-Limit-Remaining is 1 or 0.

_parse_rate_limit_reset(header_value)

Parses X-Rate-Limit-Reset into seconds to sleep.

SellerFilters

The universal filter contract. Each provider translates it into whatever its API wants.

@dataclass
class SellerFilters:
    monthly_revenue_min: float | None = None
    monthly_revenue_max: float | None = None
    categories: list[str] = field(default_factory=list)
    marketplace: str = "amazon"

Revenue is monthly here, while the ICP_REVENUE_MIN / ICP_REVENUE_MAX settings are annual — the handler divides by twelve when building the filters. See Seller discovery.

Failure handling

_post_with_retry() treats each failure class differently, and the distinctions are the point of the method:

Condition Behaviour

httpx.TimeoutException

Exponential backoff — HTTP_RETRY_BASE_SECONDS * 2^retry — up to HTTP_MAX_RETRIES, then skip the page.

HTTP 401

Call _on_unauthorized(), retry once with rebuilt headers. A second 401 skips the page.

HTTP 429

Sleep on Retry-After if present, otherwise the parsed X-Rate-Limit-Reset, then retry without incrementing the attempt count.

HTTP 5xx

Log a truncated body and skip the page. Continue with the next page.

ValidationError on one row

Skip that row only, via map_seller().

ValidationError on the response envelope

Skip the whole page.

Two details carry the design. A 429 does not consume a retry, because rate limiting is not a failure — it is the server telling you to wait, and counting it would exhaust the budget on a request that was always going to succeed. And a 5xx skips the page rather than aborting the run, so one bad page does not cost the whole invocation’s progress.

SmartScoutProvider

providers/smartscout/provider.py.

Headers

Every request, including login, carries the browser fingerprint headers in utils/constants.py as SMARTSCOUT_BROWSER_HEADERS. Search requests add:

Authorization: Bearer <jwt>
Content-Type:  application/json-patch+json
Request-Id:    <fresh uuid4 per request>

The unusual application/json-patch+json content type and the per-request Request-Id are what the API expects; neither is a mistake.

Pagination

SmartScout uses the AG Grid infinite row model, so pages are row ranges rather than page numbers:

Page 1:  { "endRow": 1000 }                    (1)
Page 2:  { "startRow": 1000, "endRow": 2000 }
Page 3:  { "startRow": 2000, "endRow": 3000 }
1 startRow is omitted, not sent as 0.

The model declares startRow: int | None = None and the body is serialised with exclude_none=True, which is what drops the field on the first page — matching the API’s own convention.

The first response carries totalRowCount. Pagination stops when startRow >= totalRowCount, when max_pages_per_run is reached, or when a page returns an empty payload.

fetch_sellers() returns (records, next_start_row), where 0 means the result set was fully consumed and the next run should restart from the beginning. The handler persists that cursor as enrichment.amazon.next_start_row — see Runtime state in configs.

Request body

{
  "loadDefaultData": false,
  "requiresFilter": false,
  "filter": {
    "estimateSalesMin": 8333.33,
    "estimateSalesMax": 83333.33
  },
  "pageFilter": {
    "endRow": 1000,
    "includeTotalRowCount": true,
    "sortModel": [],
    "fields": [
      "name", "amazonSellerId", "primaryCategoryId", "primarySubCategory",
      "estimateSales", "percentFba", "numberWinningBrands", "numberAsins",
      "numberTopAsins", "state", "country", "businessName",
      "moMGrowth", "moMGrowthCount", "startedSellingDate", "note"
    ]
  }
}

estimateSales is monthly USD. Both bounds are optional — omitting them fetches unfiltered. When ICP_CATEGORIES is set, primaryCategoryId is added to filter, and only the first category is sent per request.

The explicit fields list is a cost control: the API returns far more columns than the mapping uses, and naming them keeps responses small across thousands of rows.

Response

{
  "payload": [ { "amazonSellerId": "...", ... } ],
  "totalRowCount": 42800
}

SmartScoutSeller

models/smartscout.py. model_config = ConfigDict(extra="allow").

Field Type Notes

amazonSellerId

str

The only required field — the primary identifier

name

str | None

Display name

businessName

str | None

Legal or registered name

primaryCategoryId

int | None

An integer id, not a string

primarySubCategory

str | None

Human-readable subcategory

estimateSales

float | None

Monthly revenue estimate, USD

avgPrice

float | None

Average item price

percentFba

float | None

FBA proportion

numberWinningBrands

int | None

Brands where the seller wins the buy box

numberAsins

int | None

Total ASINs

numberTopAsins

int | None

High-performing ASINs

state, country

str | None

Location

moMGrowth

float | None

Month-over-month growth

moMGrowthCount

int | None

Months of consecutive growth

threeMonthGrowth, sixMonthGrowth, yearGrowth

float | None

Longer growth windows

sixMonthGrowthCount

int | None

—

isSuspended

bool

Defaults to False

startedSellingDate

str | None

ISO datetime string, parsed best-effort

note

str | None

Free text

extra="allow" is load-bearing. The SmartScout API is undocumented and changes without notice; strict parsing would turn a new column into a failed page. Unknown fields survive into SellerRecord.raw, so a field discovered later is already captured in historical rows.

Mapping to SellerRecord

SellerRecord field Source

seller_id

amazonSellerId

marketplace

Always Marketplace.AMAZON

business_name

businessName, then name, then ""

storefront_url

Built as https://www.amazon.com/sp?seller=<amazonSellerId>;

estimated_monthly_revenue

estimateSales, default 0.0

asin_count

numberAsins, default 0

primary_category

primarySubCategory, then str(primaryCategoryId), then ""

seller_since

Parsed from startedSellingDate with dateutil; None on failure

percent_fba

percentFba

mom_growth

moMGrowth

state, country

Direct

source

Always DataSource.SMARTSCOUT

raw

Full model_dump() of the row

SellerRecord is frozen (ConfigDict(str_strip_whitespace=True, frozen=True)) with validators enforcing non-negative revenue and ASIN counts. Immutability means a record cannot be mutated between mapping and publication, so what reaches the queue is what the provider produced.

SmartScout authentication

providers/smartscout/auth.py. SmartScoutAuth owns the whole JWT lifecycle: obtaining, caching in memory, persisting to Secrets Manager, and deciding when to refresh.

The login endpoint returns an envelope, not a bare token:

{
  "token": "eyJhbGci...",
  "refreshToken": null,
  "requireTwoFactor": false,
  "requireMobilePhoneConfirmation": false
}

token is extracted and decoded with PyJWT — without signature verification, since the only claim needed is exp.

Resolution order

get_token()
  │
  ├─ in-memory cache valid (>5 min remaining)? ──────────► return
  │
  ├─ SMARTSCOUT_TOKEN + SMARTSCOUT_TOKEN_EXPIRES_AT
  │  in the shared secret still valid? ──────────────────► cache, return
  │
  └─ authenticate
       POST /api/authentication/login with SMARTSCOUT_USERNAME / _PASSWORD
       decode exp, convert to UTC
       read-modify-write the shared secret
       cache, return

Three layers, each avoiding work the next would cost. The in-memory cache serves a warm container; the secret serves a cold container, which is the case that matters — Lambda recycles containers constantly, and without the shared cache every cold start would re-authenticate.

A five-minute buffer treats a token as expired before its actual exp, so a token cannot lapse mid-page.

_store_session() is a read-modify-write: it reads the full secret JSON, merges only SMARTSCOUT_TOKEN and SMARTSCOUT_TOKEN_EXPIRES_AT, and writes the whole document back.

Writing only the two session keys would destroy every other credential in the shared secret. This is why enrichment-scraper is the one function with secretsmanager:PutSecretValue, and it is the single most dangerous write in the platform — see Secrets Manager.

invalidate() clears the in-memory cache only; the secret is untouched, so the next get_token() re-evaluates the stored session before authenticating.

Forcing a fresh login

Delete the two session keys and the next invocation authenticates:

aws secretsmanager get-secret-value --secret-id agentcore-configs \
  --query SecretString --output text \
  | jq 'del(.SMARTSCOUT_TOKEN, .SMARTSCOUT_TOKEN_EXPIRES_AT)' \
  > /tmp/secret.json

aws secretsmanager put-secret-value --secret-id agentcore-configs \
  --secret-string file:///tmp/secret.json && rm /tmp/secret.json

Omitting those keys from a manual put-secret-value is harmless for the same reason — it costs one extra authentication.

Adding a provider

Three steps, and no change to registry.py, handler.py, queue.py, or any existing model.

  1. Response model — models/<vendor>.py, with ConfigDict(extra="allow"):

    class JungleScoutSeller(BaseModel):
    model_config = ConfigDict(extra="allow")
    ...

  2. Provider — providers/<vendor>/provider.py:

    class JungleScoutProvider(BaseProvider[JungleScoutSeller]):
    def _build_request_headers(self) -> dict[str, str]: ...
    def _to_seller_record(self, raw: JungleScoutSeller) -> SellerRecord: ...
    def fetch_sellers(self, filters: SellerFilters,
                      start_from_row: int = 0) -> tuple[list[SellerRecord], int]: ...

    Use _post_with_retry() and map_seller() from the base class. Override _on_unauthorized() if the vendor uses tokens.

  3. Call it from a handler — results compose, because every provider returns the same type:

    smartscout_records, next_row = SmartScoutProvider().fetch_sellers(filters, start_row)
jungle_records, _ = JungleScoutProvider().fetch_sellers(filters)
queue.publish(smartscout_records + jungle_records)

Keep vendor authentication inside the vendor’s own package. SmartScoutAuth lives in providers/smartscout/ rather than in a shared services/ module because its JWT flow is proprietary and means nothing to another provider. A shared auth module would accumulate every vendor’s quirks.

Add the source’s value to DataSource in models/seller.py — it already carries jungle_scout and walmart_scraper members. See Enumerations.

Design decisions

Why retry logic lives in the base class

Every seller API fails the same handful of ways. Centralising them means a new provider gets robust HTTP handling for free, and a fix to backoff behaviour applies everywhere at once.

Why a registry rather than if/elif

The dispatcher never learns about a specific handler. Adding an event type is three lines in registry.py and no change to the entrypoint. See Handler and registry pattern.

Why SHA256 for MessageDeduplicationId

SQS FIFO caps the id at 128 characters. A SHA256 hex digest is exactly 64 and deterministic, so the same seller and marketplace always produce the same id — which is what makes a rerun idempotent rather than duplicative. See Publishing events.

Why the JWT is stored in Secrets Manager

A container is reused for warm invocations but discarded soon after. An in-memory-only cache would re-authenticate on every cold start; the shared secret lets any container inherit a valid token.

Tests

cd lambdas/enrichment_scraper && uv run pytest tests -v

tests/providers/smartscout/test_provider.py covers mapping, pagination, and the whole failure matrix; test_auth.py covers the three-layer token resolution and invalidation. tests/integration/test_smartscout.py hits the live API and needs credentials — run it with uv run pytest -m integration -v. See Testing Lambdas.