Seller enrichment

Enrichment turns a bare marketplace row — a seller id, a business name, a revenue estimate — into something a salesperson can act on: a website, a LinkedIn page, named people with email addresses and phone numbers, and a HubSpot company match if one already exists.

Owned by enrichment-scorer. Implementation: lambdas/enrichment_scorer/events/base_enrichment.py, agents/seller_enrichment_agent.py, agents/prompts.py.

The four-step pipeline

For every SellerRecord.Received message, SellerReceivedHandler runs:

flowchart TD
  A["1. Upsert seller row\nkeyed on sellerId\nCOMMITS"] --> B{"enrichedAt set?"}
  B -->|yes| D{"payoutPainScore set?"}
  B -->|no| C1["2. _enrich()\nAI agent + Tavily + HubSpot MCP"]
  C1 --> C2["3. _apply_enrichment()\nseller updates + contact upserts\nlifeCycleStage = enriched\nCOMMITS"]
  C2 --> C3["publish Seller.CrmSync"]
  C3 --> D
  D -->|yes| E["done"]
  D -->|no| F1["4. _score()\nsecond AI pass, no tools"]
  F1 --> F2["_apply_score()\npayoutPainScore\nCOMMITS"]
  F2 --> E

Step 1 commits before any AI runs. Steps 2, 3, and 4 are each wrapped in try/except: a failure is logged at ERROR and the handler returns normally, so the SQS message is acked. The seller row is never lost because an LLM call failed, and a failed enrichment is picked up later by the Seller.EnrichPending backfill.

Step 1 — persist

_upsert_seller() writes via SellerRepository.upsert_by_seller_id() (ON CONFLICT (sellerId) DO UPDATE), overwriting all mutable fields.

Two clamps matter:

  • estimateSales, numberAsins, and numBrands are clamped to INT32_MAX (2,147,483,647). PostgreSQL INTEGER is 32-bit signed and outlier sellers such as Amazon Retail exceed it; clamping beats failing the upsert.

  • PLATFORM_MAP and DATASOURCE_MAP translate the scraper’s Marketplace and DataSource enums into the database’s SellerPlatform and SellerDatasource. jungle_scout maps to smartscout; walmart_scraper maps to manual.

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 after the session closes.

Step 2 — enrich

run_seller_enrichment() builds a Claude agent, cached per container with @lru_cache(maxsize=1):

Setting Value

Model

AgentModelIds.CLAUDE_HAIKU_4_5 — cheap and fast; no deep reasoning needed

max_tokens

2048

max_turns

6

Structured output

SellerEnrichmentResult

Tools (factory)

create_tavily_mcp(…​), create_hubspot_mcp(…​)

Tools are supplied as a lambda factory so each run() opens fresh MCP transports. MCP clients hold an open HTTP session and cannot be reused across agent invocations.

The prompt receives: business name (falling back to the seller id), marketplace, seller id, storefront URL, primary category, and a location string assembled from state and country ("Unknown" when both are absent). Prompts live in agents/prompts.py.

Structured result (models/enrichment.py):

class SellerContact(BaseModel):
    name: str | None                  # person's name, or None if only a role was found
    role: str | None                  # "Founder", "CEO", "Sales", "Support", …
    email: str | None
    phone_number: str | None          # validator strips all non-digits
    hubspot_contact_id: str | None

class SellerEnrichmentResult(BaseModel):
    website_url: str | None
    linkedin_url: str | None
    hubspot_company_id: str | None
    hubspot_found: bool = False
    contacts: list[SellerContact] = []
    confidence: str = "low"           # "high" = 3+ verified facts, "medium" = 1–2, "low" = minimal
    notes: str | None                 # source summary + anything that did not fit a field

Token usage (including cache read/write) is logged per seller.

Step 3 — write back

_apply_enrichment() opens one session and, inside it:

  1. Loads the seller by sellerId. A missing row logs a warning and returns.

  2. Updates the seller: lifeCycleStage = enriched, enrichedAt = now, plus website, linkedinUrl, hubspotCompanyId, and note — each written only when the agent produced a value, so an empty result never blanks an existing field.

  3. Upserts each contact through ContactRepository.upsert_for_seller().

  4. Snapshots the fields needed for the CRM event while the session is open, since ORM objects detach after commit.

Contact rules:

  • A contact with no email, no phone number, and no HubSpot contact id is skipped — there is nothing to reach them by, and nothing to deduplicate on.

  • An individual contact upsert failure is logged at WARNING and counted as skipped; the remaining contacts still persist.

  • split_name() splits the agent’s name on the first whitespace: first token is firstName, everything after is lastName.

  • source is "hubspot" when the agent returned a hubspot_contact_id, otherwise "manual".

The summary line reports upserted= and skipped= counts.

Step 4 — score

See Payout-pain score. _apply_score() opens its own short-lived session so a scoring failure can never roll back the enrichment commit.

CRM publication

When CRM_SYNC_QUEUE_URL is configured, _build_crm_sync_event() assembles a fat SellerSyncEvent from the in-session ORM objects and _publish_crm_sync() sends it to CrmSyncQueue after the transaction commits:

SQS field Value

MessageGroupId

seller_id

MessageDeduplicationId

sha256("crm-sync-" + seller_id)

Publishing is fire-and-forget: a failure is logged at WARNING and the enrichment commit stands. The event carries company fields plus a ContactSyncPayload per upserted contact — including internal_id so morpheus-hubspot can write hubspotContactId back — and life_cycle_stage = enriched. See CRM sync.

Skip conditions

Condition Effect

seller.enrichedAt is not null

Enrichment is skipped with an INFO log. Scoring is still evaluated.

seller.payoutPainScore is not null

Scoring is skipped with an INFO log.

Enrichment raised

Logged at ERROR with exc_info. Scoring still runs, using seller.note for the enrichment context instead of the fresh result.

Contact has no email, phone, or HubSpot id

Contact skipped, counted in the summary line.

CRM_SYNC_QUEUE_URL empty

No CRM event is built or published. Database state is unaffected.

Backfill

Sellers that missed enrichment — because the AI step failed, or because they were ingested before the agent existed — are recovered by the Seller.EnrichPending scheduled handler (lambdas/enrichment_scorer/events/enrich_pending.py).

  • Selects up to 10 sellers per run via SellerRepository.get_pending_enrichment(): lifeCycleStage = new and enrichedAt IS NULL.

  • Runs the same _enrich() → _apply_enrichment() → _score() → _apply_score() sequence, each in its own try/except.

  • One seller’s failure logs an ERROR and the loop moves on.

Its EventBridge rule (EnrichPendingBackfill) is currently commented out in lambdas/template.yaml; invoke it manually or re-enable the rule. See EventBridge schedules.

Cost and latency shape

The enrichment pass is the most expensive step in the platform: a reasoning model plus up to six tool turns across two remote MCP servers. Controls in place:

  • Haiku 4.5 rather than a frontier model.

  • max_turns = 6 and max_tokens = 2048.

  • ReservedConcurrentExecutions: 5 on the function.

  • Timeout: 120 seconds, below the queue’s 180-second visibility timeout.

  • Prompt and tool caching enabled for Anthropic models by ClaudeAgentService — see prompt caching.

  • BedrockTokenAlarm fires above 5M input tokens per hour.