enrichment-scraper

Physical name: morpheus-sam-enrichment-scraper. Source: lambdas/enrichment_scraper/.

The only EventBridge-only function — it has no SQS trigger. It fetches data from external sources and either publishes it to a queue or writes it to the database.

Aspect Value

Handler

handler.lambda_handler

Timeout

300 s

Memory

256 MB

Layers

common, data-layer, bigquery

Reserved concurrency

none

Registered events

detail-type Handler Model Status

SellerDiscovery.Amazon

AmazonDiscoveryHandler

SellerDiscoveryEvent

live (rule commented out)

SellerDiscovery.Walmart

WalmartDiscoveryHandler

SellerDiscoveryEvent

stub — raises NotImplementedError

SellerDiscovery.RunCategory

RunCategoryHandler

RunCategoryEvent

stub — raises NotImplementedError

Scraper.BigQueryContactSync

BigQueryContactSyncHandler

ScheduledEvent

live, rate(2 hours)

Only BigQueryContactSync has an enabled schedule. See EventBridge schedules.

SellerDiscovery.Amazon

Full business rules: Seller discovery.

flowchart LR
  A["check configs\nenrichment.amazon.disabled"] -->|disabled| A1["log + return"]
  A --> B["read next_start_row"]
  B --> C["SmartScoutProvider.fetch_sellers()"]
  C --> D["save next_start_row"]
  D --> E["QueueService.publish()\nbatches of 10"]

The cursor is saved before publishing, so a publish failure does not cause the same page to be re-fetched forever.

Event payload — every field defaults, so the EventBridge Input can be minimal:

{
  "detail-type": "SellerDiscovery.Amazon",
  "source": "enrichment.scraper",
  "detail": { "marketplace": "amazon", "provider": "smartscout", "max_pages": 5 }
}

SellerDiscoveryEvent is populate_by_name=True with detail_type aliased to detail-type, and id / version / account / region optional — an Input override does not include the standard EventBridge envelope fields.

Scraper.BigQueryContactSync

Incrementally mirrors HubSpot contacts from bigqueryexport-183608.hubspot4.contacts into three local tables. This is the main source of contacts, target_emails, and target_phone_numbers rows, and therefore of the phone numbers phone-agent later classifies.

Watermark: configs.key = 'bq_contact_sync_watermark', default 2026-01-01T00:00:00Z. Batch size: BIGQUERY_CONTACT_SYNC_BATCH_SIZE, default 1000.

Per row, best-effort — an exception is logged and the row counted as failed:

  1. Map the BigQuery row to contact fields (map_row_to_contact_data).

  2. Look up the existing contact by hubspotContactId.

  3. If the lifecycle stage is changing on an existing contact, insert an Activity NOTE recording the transition before updating.

  4. Upsert the contact — find-or-create, updating only non-null incoming fields.

  5. Ensure a target_emails row exists for property_email, propagating property_hs_email_optout to all three opt-out flags.

  6. Ensure a target_phone_numbers row exists for phone and mobilePhone.

Unique-constraint guard

contacts.email and contacts.phone are unique per subscription. When an update would move a value onto a contact that already holds it, the field is silently dropped from the update rather than raising:

for field, lookup in (("phone", _contact_repo.get_by_phone),
                      ("email", _contact_repo.get_by_email)):
    incoming = updates.get(field)
    if incoming and getattr(existing, field) != incoming:
        conflict = lookup(session, incoming)
        if conflict is not None and conflict.id != existing.id:
            updates.pop(field)

Watermark advance and the stall guard

The new watermark is the maximum updatedat across the processed rows, written only when it advanced. If a full batch is returned but the maximum did not advance — a cluster of identical timestamps larger than the batch size — the watermark is bumped by one second and a warning is logged, so the sync cannot loop forever on the same rows.

Each row is processed in its own get_session() block, so one failure does not roll back the others.

SmartScout provider

providers/smartscout/. Extends the generic BaseProvider, which owns retry and rate-limit handling; the subclass supplies three hooks:

The full provider reference — BaseProvider’s interface, the SmartScout request body and response model, the field-by-field mapping to `SellerRecord, the token lifecycle, and how to add a second provider — is in Provider pattern.
Hook Implementation

_build_request_headers()

Browser fingerprint headers, Authorization: Bearer <token>, Content-Type: application/json-patch+json, a fresh Request-Id UUID

_on_unauthorized()

self._auth.invalidate() — clears the cached token so the retry authenticates

_to_seller_record(raw)

Maps SmartScoutSeller → SellerRecord

_post_with_retry() handles: timeouts (exponential backoff, then skip the page), 401 (one retry after invalidating the token), 429 (sleep on Retry-After or the parsed X-Rate-Limit-Reset, then retry without incrementing the attempt count), 5xx (log a truncated body and skip), and proactive throttling when X-Rate-Limit-Remaining reaches 1 or 0.

fetch_sellers(filters, start_from_row) returns (records, next_start_row), where a returned 0 means the result set was fully consumed. map_seller() wraps _to_seller_record() and turns a ValidationError into a logged skip, so one bad row never aborts the page.

Token caching

SmartScoutAuth caches its session token in the shared Secrets Manager secret (SMARTSCOUT_TOKEN, SMARTSCOUT_TOKEN_EXPIRES_AT). This is why this function alone holds secretsmanager:PutSecretValue, and why a manual put-secret-value that omits those keys simply forces one extra authentication.

Configuration

Setting Default Purpose

SQS_QUEUE_URL

(template)

EnrichmentPipelineQueue

AGENTCORE_SECRET_ARN

(template)

Shared secret

SUBSCRIPTION_ID

(template)

Owner of the configs rows this function writes

SMARTSCOUT_BASE_URL

https://smartscoutapi-east.azurewebsites.net

SMARTSCOUT_MARKETPLACE

US

Marketplace header

ICP_REVENUE_MIN

100000

Annual, divided by 12 for the API

ICP_REVENUE_MAX

1000000

Annual, divided by 12

ICP_CATEGORIES

[]

Comma-separated; a field_validator splits the string

MAX_PAGES_PER_RUN

1

Overridable per event

PAGE_SIZE

10

BIGQUERY_CONTACT_SYNC_BATCH_SIZE

1000

LOG_LEVEL

INFO

A model_validator rejects ICP_REVENUE_MIN >= ICP_REVENUE_MAX at import, so a bad value fails the cold start.

Secrets read (ScraperSecrets): DATABASE_URL, SMARTSCOUT_USERNAME, SMARTSCOUT_PASSWORD (all validated non-empty), plus optional SMARTSCOUT_TOKEN, SMARTSCOUT_TOKEN_EXPIRES_AT, and BIGQUERY_CREDENTIALS.

IAM

  • SQSSendMessagePolicy on EnrichmentPipelineQueue

  • secretsmanager:GetSecretValue and PutSecretValue on AgentcoreSecret

No Bedrock access — this function runs no agents.

Tests

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

tests/providers/smartscout/test_provider.py (456 lines) and test_auth.py (336 lines) cover mapping, pagination, retry, and token caching. tests/integration/test_smartscout.py hits the real API and needs credentials. lambdas/tests/unit/test_bigquery_contact_sync.py covers the contact sync.

Invoking locally

cd lambdas
sam local invoke EnrichmentScraperFunction \
  --event events/event.json \
  --env-vars local-env.json

local-env.json is not checked in — create it, keyed by the function’s logical id:

{
  "EnrichmentScraperFunction": {
    "SQS_QUEUE_URL": "https://sqs.us-east-1.amazonaws.com/625663093496/morpheus-sam-enrichment-pipeline.fifo",
    "AGENTCORE_SECRET_ARN": "arn:aws:secretsmanager:us-east-1:625663093496:secret:agentcore-configs-XXXXX",
    "LOG_LEVEL": "DEBUG"
  }
}

The invocation uses your local AWS credentials, so it publishes to the real queue and reads the real secret. Point SQS_QUEUE_URL at a scratch queue unless you intend to feed the pipeline.

Two event payloads worth keeping around, both of which exercise routing rather than discovery:

{ "detail-type": "SellerDiscovery.RunCategory", "source": "enrichment.scraper",
  "detail": { "category": "health-beauty", "marketplace": "amazon" } }

routes through the dispatcher and raises NotImplementedError — the expected stub behaviour. An unknown detail-type returns successfully with a warning and no error, which is how the function tolerates a bus that routes it events it does not own.

Monitoring

Log output is structured JSON. handler.py injects lambda_request_id and event_type at startup and the _JsonFormatter in utils/logging.py carries them into every subsequent record, so a single invocation can be reconstructed from any log line:

{
  "timestamp": "2026-03-31T07:00:01.234Z",
  "level": "INFO",
  "lambda_request_id": "abc-123-def",
  "event_type": "SellerDiscovery.Amazon",
  "logger": "enrichment_scraper.providers.smartscout.provider",
  "message": "Fetched 847 valid seller records total"
}

CloudWatch Logs Insights:

fields timestamp, level, message, logger
| filter level = "ERROR"
| sort timestamp desc

fields timestamp, message, lambda_request_id
| filter message like "Published"
| sort timestamp desc

fields timestamp, message
| filter message like "Rate limit"
| sort timestamp desc
Signal What it means

Duration P99 approaching 300 s

Pagination is too aggressive for the timeout — lower MAX_PAGES_PER_RUN.

Frequent Rate limited (429)

The SmartScout window is too narrow for the current page count. The provider is handling it, but throughput is being spent on waiting.

Skipping seller errors

The API response shape changed. Check SmartScoutSeller — see Provider pattern.

No handler registered warnings

Something is routing unexpected events at this function.

Operating notes

  • Discovery is off by default today — both SellerDiscovery.* rules are commented out. Invoke manually; see current schedule state.

  • The downstream SQS trigger on enrichment-scorer is also commented out, so published messages accumulate on the queue.

  • enrichment.amazon.disabled in configs is the kill switch and is checked before any network call.

  • Empty BIGQUERY_CREDENTIALS makes the contact sync log a warning and return.

  • To force a fresh SmartScout login, delete the two session keys from the shared secret — see SmartScout authentication.