pyb-hubspot

Import prefix pyb_hubspot. Layer PybHubspotLayer. Used by morpheus-hubspot and enrichment-scorer.

Source: libs/pyb-hubspot/pyb_hubspot/.

Every HubSpot API call lives here. Handlers never import hubspot.* directly, which keeps auth modes, property mapping, association type ids, and disposition GUIDs in one place.

Clients

Two, sharing the same bearer token.

HubSpotSyncClient

Wraps the official HubSpot Python SDK. Exposes companies, contacts, deals, associations, and lists from crm.*, plus access_token for callers that need the raw token.

HubSpotHttpClient

Direct api.hubapi.com requests for endpoints the SDK does not cover — currently Communication Preferences. Obtain one from a sync client with as_http_client(), or build it directly.

Authentication

# Private App token — preferred: static, no refresh
client = HubSpotSyncClient.from_private_app_token("pat-na1-…")

# OAuth credentials — fallback
client = HubSpotSyncClient.from_oauth(
    client_id=secrets.hubspot_client_id,
    client_secret=secrets.hubspot_client_secret,
    refresh_token=secrets.hubspot_refresh_token,
)

http = client.as_http_client()

Both factories return the same object, so library code taking a client does not need to know which mode produced it. Use the factory class methods rather than init.

auth.refresh_hubspot_access_token() performs the OAuth exchange. It lives here — not in morpheus-llm — because token refresh is a HubSpot credential concern needed by both the LLM layer’s MCP tool and the CRM sync Lambda; morpheus_llm.utils.hubspot re-exports it.

Module map

Module Contents

client.py

HubSpotSyncClient and its two factories

clients/http_client.py

HubSpotHttpClient — REST calls outside the SDK

auth.py

refresh_hubspot_access_token()

sync/company.py

upsert_company() — three-step company match

sync/contact.py

upsert_contact() — two-step contact match + company association

sync/result.py

UpsertResult(id, created)

models.py

seller_to_company_props() and friends — domain → HubSpot property mapping

contact_properties.py

ContactPatchProperties, update_contact_properties()

contact_search.py

get_crm_contact(), search_contact_ids_by_property_eq(), resolve_hubspot_contact_id_phone_first(), PhoneFirstResolveResult

activities.py

create_email_activity(), update_crm_email(), send_custom_event()

email_properties.py

EmailActivityCreateProperties, EmailActivityUpdateProperties

call_activities.py

CallEngagementInput, create_call_engagement()

disposition_outcomes.py

Disposition, NextAction, get_hubspot_outcome_id()

engagements.py

log_note()

tasks.py

create_crm_task(), get_crm_task(), update_crm_task(), create_incoming_email_followup_task()

deals.py

get_open_deal_ids_for_contact(), update_deal_stage(), add_note_to_deal()

lists.py

create_contact_list(), delete_list(), build_call_list_filter_branch()

communications/preferences.py

apply_subscription_preference_updates()

utils/enums/endpoints.py

HubSpotApiEndpoint

utils/constants.py, utils/payloads.py

Shared constants and payload builders

Sync — company and contact upsert

Both return enough information for the caller to decide whether to log an update Note.

upsert_company(client, event) → UpsertResult

Search priority, stopping at the first hit:

  1. hubspot_company_id from the event — skips searching entirely.

  2. The custom amazon_seller_id property — catches companies created by an earlier sync whose id was never written back.

  3. name — catches companies created outside Morpheus.

    No match creates a new Company.

upsert_contact(client, payload, company_id) → str | None

Search priority:

  1. hubspot_contact_id from the payload.

  2. email — HubSpot’s own primary deduplication key for Contacts.

    No match creates a new Contact, then associates it with the Company through the v4 Associations API. A payload with no email and no existing id returns None — there is nothing to deduplicate on.

Property mapping lives in models.py so the sync functions stay free of field-name strings and the mapping can be unit-tested on its own.

resolve_hubspot_contact_id_phone_first(client, *, email, phone, mobile_phone) returns a PhoneFirstResolveResult carrying the resolved id, the match_step that found it, and had_multiple_matches. Callers log match_step, which makes a resolution auditable after the fact, and log a warning on multiple matches while using the first.

search_contact_ids_by_property_eq(client, prop, value, limit) is the primitive underneath — an equality filter on any standard property. The scheduled Hubspot.ContactResolve handler uses it directly to try phone and mobilephone against every related target_phone_numbers row.

Typed properties

Arbitrary HubSpot property names are allowed via extra="allow", while the documented custom fields are typed:

Model Use

ContactPatchProperties

Contact PATCH — any internal property name plus documented custom fields

EmailActivityCreateProperties

Email object POST; hs_timestamp defaults to now UTC when omitted, because HubSpot requires it on create

EmailActivityUpdateProperties

Email object PATCH; partial fields, no auto-filled timestamp

Validating before the call means a malformed property dict fails in the handler with a Pydantic error rather than as an opaque HubSpot 400.

Call engagements and dispositions

create_call_engagement(client, hubspot_contact_id, CallEngagementInput(…​), extra_properties=…) creates a Call object and associates it with a Contact (v3 create plus v4 associations). CallEngagementInput carries duration, summary HTML, transcript, recording URL, and start timestamp.

disposition_outcomes.py holds the disposition vocabulary and the GUID tables:

class Disposition(StrEnum):
    QUALIFIED_LEAD = "Qualified_Lead"
    AMAZON_SELLER = "Amazon_Seller"
    SALES_HANDOFF = "Sales_Handoff"
    CALLBACK_REQUESTED = "Callback_Requested"
    NOT_QUALIFIED = "Not_Qualified"
    NOT_INTERESTED = "Not_Interested"
    WRONG_NUMBER = "Wrong_Number"
    DO_NOT_CALL = "Do_Not_Call"
    NOT_A_SELLER = "Not_a_Seller"
    HANGUP_EARLY = "Hangup_Early"
    AI_ISSUE = "AI_Issue"
    BUSY = "Busy"
    AI_SCREENING = "AI_Screening"
    GATEKEEPER = "Gatekeeper"
    VOICEMAIL_DETECTED = "Voicemail_Detected"
    UNDETERMINED = "Undetermined"

class NextAction(StrEnum):
    NONE = "None"
    RETARGET_LATER = "Retarget_Later"
    SALES_HANDOFF = "Sales_Handoff"

def get_hubspot_outcome_id(disposition: str, environment: Environment) -> str | None

PRODUCTION_OUTCOME_IDS and PREVIEW_OUTCOME_IDS are separate tables selected by HUBSPOT_ENVIRONMENT. Two things to know: Qualified_Lead, Amazon_Seller, and Sales_Handoff all map to the same GUID (Warm Lead / AE Follow Up), and Voicemail_Detected and Undetermined have no mapping — the caller omits hs_call_disposition when the lookup returns None.

Tasks

Function Purpose

create_crm_task(client, props, contact_id=…, company_id=…)

Create a Task with associations; returns the HubSpot task id

get_crm_task(client, task_id)

Read task properties; raises NotFoundException when deleted

update_crm_task(client, task_id, props)

PATCH a task

create_incoming_email_followup_task(…​)

The specialised task Crm.NewActivity creates for an incoming email

hubspot_owner_id must be the Owner id from GET /crm/v3/owners. The number in a HubSpot settings URL (…​/users/user/60733684/…​) is a user id and is not guaranteed to match. Do not use contact ids, company ids, or email addresses. A wrong, invalid, or out-of-scope id often still creates the task but assigns it to another user — usually the integration user. This is documented at length in tasks.py.

Lists

create_contact_list(), delete_list(), and build_call_list_filter_branch() back the daily per-rep call list. The filterBranch payload is a generic oneOf on the SDK’s ListCreateRequest, so it is hand-built to match HubSpot’s list-filters schema and passed through as a dict.

See list materialisation.

Notes

log_note(client, …) creates a CRM Note associated with a Company and/or Contact in one API call. Morpheus logs a Note whenever a sync updated an existing record, so the timeline shows that automation touched it.

Association type ids

HubSpot-defined ids used for v3 inline associations:

Association Type id

Email → Contact

198

Email → Company

186

Note → Contact

202

Note → Company

190

Verify against the Associations API if an association starts failing — these are portal-level constants that HubSpot documents but can extend.

Communication preferences

apply_subscription_preference_updates() delegates to HubSpotHttpClient.post_communication_preference_subscription_statuses(), hitting POST /communication-preferences/2026-03/statuses/{email}. HubSpot may require one POST per subscription type, so the list is applied item by item.

Reference: HubSpot Communication Preferences.

Tests

cd libs/pyb-hubspot && uv run pytest tests -v

test_contact_search.py, test_contact_properties.py, test_lists.py, test_http_client.py, test_communications_preferences.py — all against a mocked SDK.