CrmSyncQueue events
Queue: morpheus-sam-crm-sync.fifo. Consumer:
morpheus-hubspot. Models:
libs/morpheus-common/morpheus_common/events/models.py and custom_events.py.
|
The routing key is the |
Sample payloads live in lambdas/morpheus_hubspot/events/samples/.
Seller.CrmSync
Producer: enrichment-scorer. The canonical fat event — everything needed to create a Company and its Contacts without a database read.
class ContactSyncPayload(BaseModel):
internal_id: str | None # contacts.id — used to write hubspotContactId back
first_name: str | None
last_name: str | None
job_title: str | None
email: str | None
phone: str | None
hubspot_contact_id: str | None # skip the CRM search when already known
life_cycle_stage: str | None # synced to the mor_life_cycle_stage Contact property
class SellerSyncEvent(BaseModel):
event_type: str = "Seller.CrmSync"
seller_id: str # sellers.sellerId — external marketplace id
seller_internal_id: str # sellers.id — for the hubspotCompanyId write-back
subscription_id: str
name: str | None
platform: str # required
website: str | None
linkedin_url: str | None
industry: str | None
annual_revenue: int | None
number_of_employees: int | None
description: str | None
hubspot_company_id: str | None # skip the CRM search when already known
contacts: list[ContactSyncPayload] = []{
"event_type": "Seller.CrmSync",
"seller_id": "ATVPDKIKX0DER",
"seller_internal_id": "00000000-0000-4000-8000-000000000001",
"subscription_id": "b191a905-5d19-4fa0-941b-d8eb81c0c477",
"name": "Example Seller",
"platform": "amazon",
"contacts": []
}FIFO keys: MessageGroupId = seller_id,
MessageDeduplicationId = sha256("crm-sync-" + seller_id).
Contact.Update
Producers: phone-agent, the web application. The generic Contact PATCH — no CRM search, no local lookup.
class ContactUpdateEvent(BaseModel):
event_type: str = "Contact.Update"
subscription_id: str
hubspot_contact_id: str # required, min_length=1
properties: dict[str, str] # required, must be non-empty
target_phone_number_id: str | None # write-back target after a successful PATCH
@model_validator(mode="after")
def _non_empty_properties(self): ...{
"event_type": "Contact.Update",
"subscription_id": "b191a905-5d19-4fa0-941b-d8eb81c0c477",
"hubspot_contact_id": "18719954",
"properties": { "phone_carrier_name": "Example Carrier", "phone_line_type": "mobile" },
"target_phone_number_id": "00000000-0000-4000-8000-000000000001"
}This is the event the campaign re-engagement segment and the Twilio line-intelligence job both use. See campaign re-engagement.
Contact.Unsubscribe
Updates HubSpot Communication Preferences. At least one of hubspot_contact_id or
contact_email is required.
SubscriptionPreferenceStatus = Literal["SUBSCRIBED", "UNSUBSCRIBED", "NOT_SPECIFIED"]
class SubscriptionPreferenceUpdate(BaseModel):
hubspot_subscription_type_id: str # validated: must be all digits
status: SubscriptionPreferenceStatus
class ContactUnsubscribeEvent(BaseModel):
event_type: str = "Contact.Unsubscribe"
subscription_id: str
hubspot_contact_id: str | None
contact_email: str | None
subscription_updates: list[SubscriptionPreferenceUpdate] # min_length=1
legal_basis: str | None
legal_basis_explanation: str | None
@model_validator(mode="after")
def _require_identifier(self): ... # id or emailcontact_email is preferred when present; otherwise the handler reads the email from the
CRM by id.
Contact.ResolveHubSpotIds
Trigger-only. The handler loads candidates itself, so the payload carries no PII.
class ContactResolveHubSpotIdsEvent(BaseModel):
event_type: str = "Contact.ResolveHubSpotIds"
batch_limit: int | None # ge=1, le=3000; falls back to the Lambda defaultCrm.NewActivity
Creates a logged HubSpot Email object.
class CrmNewActivityEvent(BaseModel):
event_type: str = "Crm.NewActivity"
activity_kind: str = "email" # only "email" supported; anything else raises
hubspot_contact_id: str | None
hubspot_company_id: str | None
campaign: str | None # goes on the follow-up AgentEmail event,
# not on the Email object
hubspot_owner_id: str | None # CRM owner id — assigns the incoming-email
# follow-up Task
properties: dict[str, str] # validated with EmailActivityCreateProperties
idempotency_key: str | None # producers may use for MessageDeduplicationId
@property
def has_association(self) -> bool: ... # true when a contact or company is presentAt least one association target is required, or the handler logs a warning and returns.
When hs_email_direction is INCOMING_EMAIL and hubspot_owner_id is set, a follow-up
Task is created for that owner. An AgentEmail behavioural event is sent afterwards.
|
|
Crm.UpdateActivity
class CrmUpdateActivityEvent(BaseModel):
event_type: str = "Crm.UpdateActivity"
object_type: str = "emails" # only "emails" supported; anything else raises
hubspot_object_id: str
properties: dict[str, str] # validated with EmailActivityUpdatePropertiesCrm.CustomEvent
A whitelisted HubSpot behavioural event. The wire format is a discriminated union on
custom_event_kind, wrapped in a RootModel so model_validate(body) works in the
registry.
class CustomEventKind(StrEnum):
AGENT_EMAIL = "agent_email"
class CrmCustomEventAgentEmail(BaseModel):
event_type: Literal["Crm.CustomEvent"] = "Crm.CustomEvent"
custom_event_kind: Literal[CustomEventKind.AGENT_EMAIL]
hubspot_contact_object_id: str
hubspot_event_name_override: str | None
status: Literal["opened", "clicked", "bounced", "complained",
"unsubscribed", "delivered", "replied", "sent"]
campaign: str | None
link_url: str | None
campaign_id: str | None # local campaign UUID, links follow-up tasks
CrmCustomEventRoot = Annotated[Union[CrmCustomEventAgentEmail],
Field(discriminator="custom_event_kind")]
class CrmCustomEventEnvelope(RootModel[CrmCustomEventRoot]): ...status, campaign, and link_url map to HubSpot properties of the same names. The
event is sent to POST /events/v3/send as pe9298797_agentemail unless
overridden.
An unknown custom_event_kind fails validation before any API call — the whitelist is
the security boundary.
Crm.TaskCreate
class CrmTaskCreateEvent(BaseModel):
event_type: str = "Crm.TaskCreate"
contact_id: str | None # internal contacts.id UUID
hubspot_contact_id: str | None
hubspot_company_id: str | None # optional association
hubspot_owner_id: str # required — assignment + sales_reps lookup
task_type: str # "call" | "email" | "todo"
subject: str # min_length=1 → hs_task_subject
due_at: datetime | None # → hs_timestamp
priority: str = "MEDIUM" # LOW | MEDIUM | HIGH
body: str | None # → hs_task_body
campaign_target_id: str | None
@model_validator(mode="after")
def _require_contact(self): ... # contact_id or hubspot_contact_id{
"event_type": "Crm.TaskCreate",
"hubspot_contact_id": "12345678",
"hubspot_owner_id": "98765432",
"task_type": "call",
"subject": "Follow up with prospect",
"due_at": "2026-06-13T14:00:00Z",
"priority": "HIGH",
"body": "Discuss pricing options"
}Crm.DealUpdate
class CrmDealUpdateEvent(BaseModel):
event_type: str = "Crm.DealUpdate"
deal_stage: str # required, min_length=1
hubspot_deal_id: str | None # update this specific deal
hubspot_contact_id: str | None # or every open deal for this contact
note: str | None # logged on each deal before the stage change
@model_validator(mode="after")
def _require_one_identifier(self): ...Crm.DialerCallEnded
Deliberately thin: the Call and CampaignTarget rows are large, mutable, and
authoritative, so the consumer loads them rather than trusting an embedded copy.
class CrmDialerCallEndedEvent(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True, extra="ignore")
event_type: str = "Crm.DialerCallEnded"
source: str = "dialer"
subscription_id: str
call_id: str
target_id: str
campaign_id: str
provider_call_id: str | None
campaign_name: str | None
disposition: str | None # descriptive; the handler recomputes it from the row
duration_sec: int | None{
"event_type": "Crm.DialerCallEnded",
"source": "dialer",
"subscription_id": "b191a905-5d19-4fa0-941b-d8eb81c0c477",
"call_id": "00000000-0000-4000-8000-000000000001",
"target_id": "00000000-0000-4000-8000-000000000002",
"campaign_id": "00000000-0000-4000-8000-000000000003",
"provider_call_id": "retell-call-abc",
"disposition": "Qualified_Lead",
"duration_sec": 120
}Consumer behaviour: Call logging and dispositions.
Crm.HubSpotWebhook
A batch of inbound HubSpot notifications relayed by an external forwarder. Inner events
arrive in HubSpot’s camelCase and are accepted in either case via AliasChoices.
class HubSpotWebhookInnerEvent(BaseModel):
model_config = ConfigDict(extra="ignore", populate_by_name=True)
subscription_type: str # subscriptionType
object_id: int # objectId — numeric strings are coerced
occurred_at: int # occurredAt, epoch milliseconds
property_name: str | None # propertyName
property_value: str | None # propertyValue
hubspot_event_id: int | None
hubspot_webhook_subscription_id: int | None
attempt_number: int | None
change_source: str | None
app_id: int | None
portal_id: int | None
source_id: str | None
class CrmHubSpotWebhookEvent(BaseModel):
event_type: Literal["Crm.HubSpotWebhook"] = "Crm.HubSpotWebhook"
source: str
portal_id: int # portalId
events: list[HubSpotWebhookInnerEvent] # min_length=1{
"event_type": "Crm.HubSpotWebhook",
"source": "hubspot-forwarder",
"portalId": 12345678,
"events": [
{
"subscriptionType": "contact.propertyChange",
"objectId": 987654321,
"occurredAt": 1713628800000,
"propertyName": "hs_email_optout_44467216",
"propertyValue": "true"
}
]
}Processors: HubSpot webhooks.
FollowUpTask.Sync
class FollowUpTaskSyncEvent(BaseModel):
event_type: str = "FollowUpTask.Sync"
task_id: str # internal follow_up_tasks.id UUID
hubspot_task_id: str # HubSpot task engagement id
status: str # a FollowUpTaskStatus valueThe handler reads HubSpot first: an already-completed task is accepted rather than overwritten. See Follow-up tasks and escalation.
Common guards
Every handler on this queue applies these before doing work; each logs at ERROR and
returns without raising, so the message is deleted rather than retried:
| Guard | Applies to |
|---|---|
|
|
Non-empty |
|
Required association present |
|
Supported |
|