Adding a new event

Adding an event never means editing handler.py. Five steps: model, handler, registry, trigger, sample.

1. Define the model

Cross-function event — a producer in one function or the web application, a consumer in another. Add it to libs/morpheus-common/morpheus_common/events/models.py and export it from events/init.py and morpheus_common/init.py.

Function-local event — no other component constructs it. Put it in that function’s models/.

class ContactArchiveEvent(BaseModel):
    """Archive a HubSpot contact and mark the local row inactive."""

    model_config = ConfigDict(str_strip_whitespace=True)

    event_type: str = "Contact.Archive"          # default = the routing key
    subscription_id: str
    hubspot_contact_id: str = Field(..., min_length=1)
    reason: str | None = None

    @model_validator(mode="after")
    def _require_reason_when_permanent(self) -> "ContactArchiveEvent":
        ...
        return self

Conventions:

event_type defaults to the routing key, so a producer gets it for free.
ConfigDict(str_strip_whitespace=True) for domain events; ConfigDict(extra="ignore") for anything mirroring an external payload.
Cross-field requirements go in a model_validator(mode="after") — rejected at the boundary, never mid-handler.
Field descriptions matter; they are the contract a producer reads.

2. Write the handler

events/<name>.py, extending the function’s base class.

"""Handler for ``Contact.Archive`` — archive a HubSpot contact."""

from __future__ import annotations

import logging
from typing import Any

from morpheus_common.events.models import ContactArchiveEvent
from morpheus_data_layer.repositories import ContactRepository
from morpheus_data_layer.session import get_session

from events.base import BaseEventHandler
from models.config import settings
from models.secrets import load_secrets
from utils.hubspot_client import get_hubspot_client

logger = logging.getLogger(__name__)

_contact_repo = ContactRepository()          # module-level; repositories are stateless


class ContactArchiveHandler(BaseEventHandler):
    def handle(self, event: ContactArchiveEvent, context: Any) -> None:
        _ = context

        if event.subscription_id != str(settings.subscription_id):
            logger.error(
                "Contact.Archive subscription mismatch event=%r lambda=%r",
                event.subscription_id, str(settings.subscription_id),
            )
            return                                    # bad data → drop, do not retry

        secrets = load_secrets()
        client = get_hubspot_client(secrets)

        archive_contact(client, event.hubspot_contact_id)   # raises → DLQ

        with get_session() as session:
            _contact_repo.update_by_hubspot_id(
                session, event.hubspot_contact_id, {"isActive": False},
            )

        logger.info(
            "Contact.Archive completed hubspot_contact_id=%r reason=%r",
            event.hubspot_contact_id, event.reason,
        )

The rules this example follows:

Guard first. Tenant mismatch, missing identifiers, empty payloads — log at ERROR and return. The message is deleted; retrying cannot fix bad data.
Let transient failures raise. A HubSpot 5xx or a database error propagates so the message reaches the DLQ.
Repositories at module level, sessions in the handler. Repositories never open sessions and never commit.
get_session() owns the transaction, and it commits on clean exit.
No HubSpot SDK imports. Every CRM call goes through pyb_hubspot.
End with a summary line so one log line tells the whole story.
Log identifiers, not payloads.

Multi-row handlers

For a handler that iterates, wrap each row so one failure does not abort the batch, and count outcomes:

updated = skipped = 0
for row in rows:
    try:
        ...
        updated += 1
    except Exception:
        logger.error("… failed id=%r", row["id"], exc_info=True)
        skipped += 1
logger.info("Job completed updated=%d skipped=%d total=%d", updated, skipped, len(rows))

3. Register the routing key

Dataclass registry (scraper, scorer, email-agent, phone-agent):

def build_registry() -> EventRegistry:
    registry = EventRegistry()
    registry.register("Contact.Archive", ContactArchiveHandler, ContactArchiveEvent)
    return registry

Dict registry (morpheus-hubspot):

event_registry: dict = {
    …
    "Contact.Archive": (ContactArchiveHandler(), ContactArchiveEvent),
}

Scheduled handlers in morpheus-hubspot go in scheduled_registry instead, keyed by detail-type.

The registry key is the wire contract. It must match the event_type a producer sends, or the detail-type in a rule’s Input, exactly. A mismatch is a silent no-op: the dispatcher logs "no handler registered" and the invocation succeeds.

4. Add the trigger

SQS — the queue trigger already exists on the function; nothing to add. The producer sets the routing key.

Scheduled — add an Events entry in lambdas/template.yaml:

Events:
  ContactArchiveSweep:
    Type: Schedule
    Properties:
      Name: !Sub "${AWS::StackName}-contact-archive-sweep"
      Description: "Archive contacts flagged for removal — daily"
      Schedule: cron(0 5 * * ? *)
      Enabled: true
      Input: '{"detail-type":"Contact.ArchiveSweep","source":"morpheus.hubspot","detail":{}}'

New queue — define the queue, a dead-letter queue, and the Events entry, keeping the function Timeout below the queue’s VisibilityTimeout. See Queues.

Check the layer list too: a handler needing database access requires MorpheusDataLayer, an agent requires MorpheusLlmLayer, HubSpot requires PybHubspotLayer, BigQuery requires MorpheusBigQueryLayer.

5. Add a sample and a test

Sample payload in <function>/events/samples/<name>.json — the inner message body, not the full Lambda event. Use REPLACE_* placeholders for ids.

{
  "event_type": "Contact.Archive",
  "subscription_id": "b191a905-5d19-4fa0-941b-d8eb81c0c477",
  "hubspot_contact_id": "REPLACE_HUBSPOT_CONTACT_ID",
  "reason": "duplicate record"
}

Then a test. Highest-value targets first: the decision functions, the guard clauses, and the model validators. See Testing Lambdas.

6. Document it

Add a row to Event catalog.
Add the schema to the relevant queue page, or a row to Scheduled events.
If it encodes a business rule, add or extend a page under Business logic.
If it adds a schedule, update EventBridge schedules.

Deploy

cd lambdas
make deploy-hubspot        # re-exports requirements.txt, builds, deploys

Then verify:

aws lambda invoke \
  --function-name morpheus-sam-hubspot-sync \
  --payload "$(cat morpheus_hubspot/events/samples/contact_archive.json | jq -c '{Records:[{messageId:"t1",eventSource:"aws:sqs",body:(.|tostring)}]}')" \
  --cli-binary-format raw-in-base64-out /dev/stdout

Checklist

Model defined, with validators, in the right place
Model exported from init.py if it is cross-function
Handler extends the function’s base class
Guards return without raising; transient failures raise
Repositories at module level; get_session() owns the transaction
Registered with a key matching the wire contract exactly
Trigger added, with Timeout < VisibilityTimeout if it is a new queue
Required layers present on the function
Sample payload added
Test covering the decision logic
Documentation updated