morpheus-common

Import prefix morpheus_common. Layer MorpheusCommonLayer — always first in a function’s layer list, because every handler calls configure_sentry() at cold start.

Source: libs/morpheus-common/morpheus_common/.

Public surface

from morpheus_common import configure, load_raw_secret
from morpheus_common.events import (
    SellerSyncEvent, ContactSyncPayload, ContactUpdateEvent, ContactUnsubscribeEvent,
    ContactResolveHubSpotIdsEvent, CrmNewActivityEvent, CrmUpdateActivityEvent,
    CrmTaskCreateEvent, CrmDialerCallEndedEvent, CrmHubSpotWebhookEvent,
    HubSpotWebhookInnerEvent, FollowUpTaskSyncEvent, SubscriptionPreferenceUpdate,
    CrmCustomEventEnvelope, CrmCustomEventAgentEmail, CustomEventKind,
    custom_event_to_hubspot_properties, hubspot_event_name, parse_crm_custom_event,
    publish_event,
)

CrmDealUpdateEvent is defined in events/models.py but is not re-exported from events/init.py; import it from morpheus_common.events.models directly, as morpheus_hubspot/registry.py does.

events/models.py — the event contracts

Pydantic v2 models defining every message crossing a component boundary. They are the platform’s real interface, so a change here is a wire-format change.

Full schemas: CrmSyncQueue events.

Conventions:

  • event_type defaults to the routing key, so constructing the model gives the producer the key for free.

  • ConfigDict(str_strip_whitespace=True) on domain events; ConfigDict(extra="ignore") on anything mirroring an external payload.

  • Cross-field requirements live in a model_validator(mode="after"): ContactUnsubscribeEvent requires an id or an email, CrmTaskCreateEvent requires a contact identifier, CrmDealUpdateEvent requires a deal or contact id, ContactUpdateEvent requires non-empty properties.

  • AliasChoices where an external system uses different casing — HubSpotWebhookInnerEvent accepts both subscriptionType and subscription_type.

  • Field descriptions are written for producers, including the traps: the CrmNewActivityEvent.hubspot_owner_id description spells out that it must be the CRM owner id, not a settings URL user id.

The module docstring documents both sides of the contract:

# producer
event = SellerSyncEvent(seller_id="ATVPDKIKX0DER", …, contacts=[ContactSyncPayload(…)])
publish_event(queue_url=settings.crm_sync_queue_url, event=event, group_id=seller_id)

# consumer
event = SellerSyncEvent.model_validate_json(record["body"])

events/custom_events.py — the behavioural-event whitelist

Only whitelisted custom_event_kind values are accepted, each with a dedicated model, so a payload is validated before any HubSpot call.

class CustomEventKind(StrEnum):
    AGENT_EMAIL = "agent_email"

_DEFAULT_HUBSPOT_EVENT_NAMES = {"agent_email": "pe9298797_agentemail"}

def hubspot_event_name(kind, *, override=None) -> str: ...
def parse_crm_custom_event(data: object) -> CrmCustomEventAgentEmail: ...
def custom_event_to_hubspot_properties(event) -> dict[str, str]: ...

The HubSpot event name follows the portal-qualified pe{hubId}_{internal_name} pattern. hubspot_event_name_override on the event allows a per-message override.

custom_event_to_hubspot_properties() maps the validated model to HubSpot’s properties dict — status always, campaign and link_url only when set. The whitelist plus this explicit mapping is the security boundary: an unknown kind fails validation, and an unmapped field never reaches HubSpot.

events/publisher.py — publish_event()

def publish_event(*, queue_url: str, event: BaseModel, group_id: str,
                  dedup_id: str | None = None) -> None:

Wraps boto3 SQS send_message with FIFO defaults and Pydantic serialisation, keeping producers free of boto3 boilerplate. The client is a module-level singleton pinned to us-east-1.

dedup_id defaults to sha256 of the serialised body. Logging is DEBUG level with the event type, group id, and queue — never the body.

Group and deduplication key choices: Publishing events.

FIFO only. MessageGroupId and MessageDeduplicationId are invalid on a standard queue, which is why phone-agent has its own publish_to_standard_queue() for EmailAgentQueue.

secrets.py — load_raw_secret()

@lru_cache(maxsize=4)
def load_raw_secret(secret_arn: str) -> dict:

One Secrets Manager call per ARN per container lifetime, regardless of how many modules ask. Returns the parsed JSON as a plain dict keyed by the exact Secrets Manager keys ("DATABASE_URL", "HUBSPOT_CLIENT_ID", …).

maxsize=4 covers the common single-ARN case with room to spare.

Raises RuntimeError when the secret exists but has no string value; json.JSONDecodeError on invalid JSON; botocore ClientError on any AWS error.

Each function wraps it in its own typed loader, mapping only the keys it needs:

@lru_cache(maxsize=1)
def load_secrets() -> PhoneAgentSecrets:
    data = load_raw_secret(settings.agentcore_secret_arn)
    return PhoneAgentSecrets.model_validate({
        "database_url":       data.get("DATABASE_URL", ""),
        "twilio_account_sid": data.get("TWILIO_ACCOUNT_SID", ""),
        …
    })

Two layers of caching, and extra="ignore" on every secrets model, so an unrelated key can be added to the shared secret without touching any function.

The cache is also why a rotated secret takes effect only on the next cold start. See Configuration and secrets.

sentry.py — configure()

def configure() -> None:

Reads everything from the environment and does nothing when SENTRY_DSN is empty or unset — which is the current production state.

Variable Effect

SENTRY_DSN

Empty or unset disables Sentry entirely

SENTRY_ENVIRONMENT

Environment tag; defaults to production (the SAM template sets prod)

SENTRY_RELEASE

Optional release identifier for correlating errors to deploys

SENTRY_TRACES_SAMPLE_RATE

Documented, but the traces_sample_rate line is currently commented out

Integrations when enabled:

  • AwsLambdaIntegration(timeout_warning=True) — captures unhandled handler exceptions and flushes synchronously before the runtime freezes.

  • LoggingIntegration(level=INFO, event_level=ERROR) — INFO+ records become breadcrumbs, ERROR+ records become events.

  • send_default_pii=False — strips IP addresses and user agents.

Two safety properties: it is idempotent (safe to call more than once), and initialisation failure is caught and logged at WARNING rather than breaking the cold start. An observability failure must never take down a function.

LoggingIntegration forwards ERROR-level records as events, so any PII placed in an error message reaches Sentry despite send_default_pii=False. The morpheus-hubspot dispatcher logs full message bodies on failure — worth resolving before enabling a DSN. See PII.

Tests

cd libs/morpheus-common && uv run pytest tests -v

test_contact_update_event.py, test_contact_unsubscribe_event.py, test_contact_resolve_hubspot_ids_event.py, test_crm_hubspot_webhook_event.py, test_custom_events.py — all validation-focused, asserting that the model rejects what it should. Since these models are the wire contract, that is the right place to spend test effort.