enrichment-scorer

Physical name: morpheus-sam-enrichment-scorer. Source: lambdas/enrichment_scorer/.

Consumes raw seller records, persists them, runs two AI passes over them, and publishes the result to the CRM queue. The only function whose behaviour is genuinely probabilistic — which is why every AI step is best-effort and every database write commits before one runs.

Aspect Value

Handler

handler.lambda_handler

Timeout

120 s (queue VisibilityTimeout is 180 s)

Memory

512 MB

Reserved concurrency

5

Layers

common, data-layer, llm, pyb-hubspot

Registered events

One registry serves both transports.

Routing key Handler Model Trigger

SellerRecord.Received

SellerReceivedHandler

SellerRecord

EnrichmentPipelineQueue (trigger commented out)

Seller.EnrichPending

EnrichPendingSellersHandler

ScheduledEvent

EventBridge (rule commented out)

Seller.ScorePending

ScorePendingSellersHandler

ScheduledEvent

EventBridge (rule commented out)

All three triggers are currently commented out in lambdas/template.yaml. The function deploys and its handlers work when invoked directly, but nothing drives it automatically. See EventBridge schedules.

The SQS path defaults its routing key to SellerRecord.Received when the event_type message attribute is absent, for compatibility with messages published before the scraper set it.

BaseEnrichmentHandler

events/base_enrichment.py — the one base class in the repository that carries real behaviour rather than just an abstract method. Subclasses differ only in how they source sellers and in what order they call the inherited steps.

BaseEventHandler                          (events/base.py)
  └── BaseEnrichmentHandler               (events/base_enrichment.py)
        │  _seller_repo, _contact_repo    (class attributes)
        │  _enrich(...)                   AI pass 1 — Tavily + HubSpot MCP
        │  _apply_enrichment(...)         write seller + contacts, publish CRM event
        │  _score(...)                    AI pass 2 — no tools
        │  _apply_score(...)              write payoutPainScore
        │  _build_crm_sync_event(...)     snapshot ORM state into a fat event
        │  _publish_crm_sync(...)         fire-and-forget SQS publish
        ├── SellerReceivedHandler         SQS — adds _upsert_seller()
        ├── EnrichPendingSellersHandler   schedule — queries pending sellers
        └── ScorePendingSellersHandler    schedule — scoring only

Repositories are class attributes, instantiated once at import. Repositories are stateless and never open sessions, so sharing them is safe.

SellerRecord.Received

Full rules: Seller enrichment.

seller = self._upsert_seller(event)            # step 1 — COMMITS

if seller.enrichedAt is None:                  # steps 2 + 3, best-effort
    try:
        enrichment = self._enrich(...)
        self._apply_enrichment(event.seller_id, enrichment)
    except Exception:
        logger.error(..., exc_info=True)

if seller.payoutPainScore is None:             # step 4, best-effort
    try:
        score = self._score(...)
        self._apply_score(event.seller_id, score)
    except Exception:
        logger.error(..., exc_info=True)

Steps 2–4 must never cause an SQS retry. The database write in step 1 has already committed; re-delivering the message would repeat the AI cost for no benefit. Catch everything, log at ERROR, return.

_upsert_seller() also handles two data-shape concerns: estimateSales, numberAsins, and numBrands are clamped to INT32_MAX (PostgreSQL INTEGER is 32-bit signed and outliers such as Amazon Retail exceed it), and PLATFORM_MAP / DATASOURCE_MAP translate the scraper’s enums into the database’s.

After the upsert the row is refreshed inside the session and enrichedAt, payoutPainScore, and note are touched, so they remain readable on the detached object once the session closes.

Seller.EnrichPending

Backfill for sellers that missed the SQS path. Selects up to 10 per run (get_pending_enrichment(): lifeCycleStage = new and enrichedAt IS NULL) and runs the same four steps per seller, each in its own try/except. One seller’s failure logs and the loop continues.

Note the fallback: when enrichment fails, scoring still runs using seller.note as its enrichment context instead of the fresh result.

Seller.ScorePending

Scoring only. Selects up to 10 enriched sellers with no score (get_pending_scoring(): enrichedAt IS NOT NULL and payoutPainScore IS NULL) and runs _score() → _apply_score(). Intended for backfilling after the scoring feature was introduced, or after a deliberate rescore.

The two agents

Enrichment agent Scoring agent

Module

agents/seller_enrichment_agent.py

agents/seller_scoring_agent.py

Model

CLAUDE_HAIKU_4_5

CLAUDE_HAIKU_3

max_tokens

2048

4096

max_turns

6

3

Tools

Tavily MCP + HubSpot MCP (factory)

none

Output model

SellerEnrichmentResult

PayoutPainScoreResult

Both are built inside @lru_cache(maxsize=1) functions with deferred imports of morpheus_llm, so the module can be imported in a test environment without the Lambda layer installed:

@lru_cache(maxsize=1)
def _get_agent():
    from morpheus_llm.providers import ClaudeAgentService
    from morpheus_llm.tools import create_hubspot_mcp, create_tavily_mcp
    from morpheus_llm.types import AgentConfig
    from morpheus_llm.utils import AgentModelIds
    return ClaudeAgentService(config=AgentConfig(...), tools=lambda: [...])

Preserve that pattern when adding an agent — it is what keeps the tests runnable.

Prompts are in agents/prompts.py. The scoring prompt contains the entire 8-signal formula; see Payout-pain score.

Token usage — including cache read and write — is logged per seller for both agents.

CRM publication

When CRM_SYNC_QUEUE_URL is set, _build_crm_sync_event() assembles a fat SellerSyncEvent from the in-session ORM objects (it must run before the session closes; detached objects cannot lazy-load) and _publish_crm_sync() sends it after the commit:

SQS field Value

MessageGroupId

seller_id

MessageDeduplicationId

sha256("crm-sync-" + seller_id)

Fire-and-forget: a publish failure logs at WARNING and the committed enrichment stands. An empty CRM_SYNC_QUEUE_URL skips event construction entirely.

Configuration

Setting Default Purpose

AGENTCORE_SECRET_ARN

(template)

Shared secret

SUBSCRIPTION_ID

(template)

Must match the scraper’s, so sellers are owned by the same tenant

CRM_SYNC_QUEUE_URL

""

Optional — the function still writes to the database without it

LOG_LEVEL

INFO

ScorerSecrets reads DATABASE_URL, TAVILY_API_KEY, and the HubSpot OAuth triple.

IAM

  • SQSPollerPolicy on EnrichmentPipelineQueue

  • SQSSendMessagePolicy on CrmSyncQueue

  • secretsmanager:GetSecretValue on AgentcoreSecret

  • bedrock:InvokeModel and InvokeModelWithResponseStream on *

Cost control

ReservedConcurrentExecutions: 5, BatchSize: 1, max_turns caps on both agents, and Anthropic prompt/tool caching enabled by ClaudeAgentService. The BedrockTokenAlarm fires above 5M input tokens per hour.

The commented-out SQS trigger also carries ScalingConfig.MaximumConcurrency: 50 with a note explaining the constraint: each invocation holds a database connection for the batch, and the data layer’s pool default is 5 per process. Raise it once RDS Proxy sits in front of the database.

Operating notes

  • Both AI steps skip cleanly when enrichedAt / payoutPainScore is already set, so re-running is cheap.

  • To rescore a cohort, null payoutPainScore for those sellers and drive Seller.ScorePending — 10 per invocation, or raise _BATCH_SIZE in the handler.

  • A seller stuck at lifeCycleStage = new with a null enrichedAt means enrichment has failed repeatedly. Check the function’s logs for the Enrichment pipeline failed line.