How to Sync Stripe Payments to HubSpot with Python (Complete Tutorial)

I've built this integration three times now. Once at a startup where the CEO kept asking "did that enterprise deal actually start paying?" and nobody could answer without logging into Stripe. Once for a RevOps team that was manually copying subscription statuses into HubSpot custom properties every Monday. And once for myself, because I got tired of the two systems pretending the other didn't exist.

The core problem is simple: Stripe knows about payments. HubSpot knows about deals and contacts. Neither tells the other when something important happens. A customer churns in Stripe and their HubSpot deal sits at "Closed Won" for months. A payment fails and the CSM doesn't find out until the customer complains. Your CRM revenue report and your Stripe revenue report disagree by 15% and nobody can explain why.

Here's how to fix it with a Python webhook listener that takes about an afternoon to build.

What we're building

A Flask webhook endpoint that listens to Stripe events and updates HubSpot in real time. When it's done, you'll have:

New Stripe customers automatically matched to HubSpot contacts
Subscription creates/updates reflected as deal property changes
Payment failures surfaced as HubSpot timeline events
Churns automatically moving deals to a "Churned" pipeline stage
MRR and subscription status as live custom properties on the contact

Prerequisites

You need a Stripe account with webhook access, a HubSpot account with API access (Marketing Hub Starter or above), and Python 3.9+. You'll also need these packages:

pip install flask stripe hubspot-api-client python-dotenv

Step 1: Set up HubSpot custom properties

Before writing any code, create these custom properties in HubSpot (Settings → Properties → Contact Properties):

# create_hubspot_properties.py
import os
from dotenv import load_dotenv
from hubspot import HubSpot
from hubspot.crm.properties import SimplePublicObjectInput
 
load_dotenv()
client = HubSpot(access_token=os.environ["HUBSPOT_ACCESS_TOKEN"])
 
properties = [
    {
        "name": "stripe_customer_id",
        "label": "Stripe Customer ID",
        "type": "string",
        "fieldType": "text",
        "groupName": "contactinformation",
    },
    {
        "name": "stripe_subscription_status",
        "label": "Stripe Subscription Status",
        "type": "enumeration",
        "fieldType": "select",
        "groupName": "contactinformation",
        "options": [
            {"label": "Active", "value": "active"},
            {"label": "Past Due", "value": "past_due"},
            {"label": "Canceled", "value": "canceled"},
            {"label": "Trialing", "value": "trialing"},
            {"label": "Unpaid", "value": "unpaid"},
        ],
    },
    {
        "name": "stripe_mrr",
        "label": "Stripe MRR (cents)",
        "type": "number",
        "fieldType": "number",
        "groupName": "contactinformation",
    },
    {
        "name": "stripe_last_payment_date",
        "label": "Last Stripe Payment",
        "type": "date",
        "fieldType": "date",
        "groupName": "contactinformation",
    },
]
 
for prop in properties:
    try:
        client.crm.properties.core_api.create(
            object_type="contacts",
            simple_public_object_input=SimplePublicObjectInput(**prop),
        )
        print(f"Created: {prop['name']}")
    except Exception as e:
        print(f"Skipped {prop['name']}: {e}")

Run this once. It creates the fields your sync will write into.

Step 2: The webhook listener

This is the core of the integration. A Flask app that receives Stripe webhook events and dispatches them to handlers:

# webhook_server.py
import os
import json
import stripe
from flask import Flask, request, jsonify
from dotenv import load_dotenv
from hubspot import HubSpot
from hubspot.crm.contacts import (
    SimplePublicObjectInput,
    PublicObjectSearchRequest,
    Filter,
    FilterGroup,
)
 
load_dotenv()
 
app = Flask(__name__)
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
webhook_secret = os.environ["STRIPE_WEBHOOK_SECRET"]
hubspot = HubSpot(access_token=os.environ["HUBSPOT_ACCESS_TOKEN"])
 
 
def find_hubspot_contact(email):
    """Find a HubSpot contact by email address."""
    filter = Filter(property_name="email", operator="EQ", value=email)
    filter_group = FilterGroup(filters=[filter])
    search_request = PublicObjectSearchRequest(
        filter_groups=[filter_group],
        properties=["email", "stripe_customer_id"],
    )
    results = hubspot.crm.contacts.search_api.do_search(
        public_object_search_request=search_request
    )
    if results.total > 0:
        return results.results[0]
    return None
 
 
def update_hubspot_contact(contact_id, properties):
    """Update a HubSpot contact with new properties."""
    hubspot.crm.contacts.basic_api.update(
        contact_id=contact_id,
        simple_public_object_input=SimplePublicObjectInput(
            properties=properties
        ),
    )
 
 
def handle_customer_subscription_created(event):
    """New subscription → update HubSpot contact with status and MRR."""
    subscription = event["data"]["object"]
    customer = stripe.Customer.retrieve(subscription["customer"])
 
    if not customer.email:
        return
 
    contact = find_hubspot_contact(customer.email)
    if not contact:
        return
 
    mrr = sum(
        item["price"]["unit_amount"] * item["quantity"]
        for item in subscription["items"]["data"]
    )
 
    update_hubspot_contact(contact.id, {
        "stripe_customer_id": customer.id,
        "stripe_subscription_status": subscription["status"],
        "stripe_mrr": str(mrr),
    })
 
 
def handle_invoice_paid(event):
    """Successful payment → update last payment date."""
    invoice = event["data"]["object"]
    if not invoice["customer_email"]:
        return
 
    contact = find_hubspot_contact(invoice["customer_email"])
    if not contact:
        return
 
    from datetime import datetime
    payment_date = datetime.fromtimestamp(
        invoice["status_transitions"]["paid_at"]
    ).strftime("%Y-%m-%d")
 
    update_hubspot_contact(contact.id, {
        "stripe_last_payment_date": payment_date,
        "stripe_subscription_status": "active",
    })
 
 
def handle_invoice_payment_failed(event):
    """Failed payment → mark as past_due, create timeline event."""
    invoice = event["data"]["object"]
    if not invoice["customer_email"]:
        return
 
    contact = find_hubspot_contact(invoice["customer_email"])
    if not contact:
        return
 
    update_hubspot_contact(contact.id, {
        "stripe_subscription_status": "past_due",
    })
 
    # Create a note so the CSM sees it immediately
    from hubspot.crm.objects.notes import SimplePublicObjectInput as NoteInput
    hubspot.crm.objects.notes.basic_api.create(
        simple_public_object_input=NoteInput(
            properties={
                "hs_note_body": (
                    f"⚠️ Payment failed for invoice {invoice['id']}. "
                    f"Amount: ${invoice['amount_due'] / 100:.2f}. "
                    f"Attempt #{invoice['attempt_count']}."
                ),
                "hs_timestamp": str(int(invoice["created"] * 1000)),
            },
            associations=[{
                "to": {"id": contact.id},
                "types": [{"associationCategory": "HUBSPOT_DEFINED",
                           "associationTypeId": 202}],
            }],
        )
    )
 
 
def handle_subscription_deleted(event):
    """Subscription canceled → update status, zero out MRR."""
    subscription = event["data"]["object"]
    customer = stripe.Customer.retrieve(subscription["customer"])
 
    if not customer.email:
        return
 
    contact = find_hubspot_contact(customer.email)
    if not contact:
        return
 
    update_hubspot_contact(contact.id, {
        "stripe_subscription_status": "canceled",
        "stripe_mrr": "0",
    })
 
 
EVENT_HANDLERS = {
    "customer.subscription.created": handle_customer_subscription_created,
    "customer.subscription.updated": handle_customer_subscription_created,
    "invoice.paid": handle_invoice_paid,
    "invoice.payment_failed": handle_invoice_payment_failed,
    "customer.subscription.deleted": handle_subscription_deleted,
}
 
 
@app.route("/webhooks/stripe", methods=["POST"])
def stripe_webhook():
    payload = request.get_data()
    sig_header = request.headers.get("Stripe-Signature")
 
    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, webhook_secret
        )
    except (ValueError, stripe.error.SignatureVerificationError):
        return jsonify({"error": "Invalid signature"}), 400
 
    handler = EVENT_HANDLERS.get(event["type"])
    if handler:
        try:
            handler(event)
        except Exception as e:
            # Log but don't fail — Stripe retries on 5xx
            print(f"Handler error for {event['type']}: {e}")
            return jsonify({"error": "Handler failed"}), 500
 
    return jsonify({"received": True}), 200
 
 
if __name__ == "__main__":
    app.run(port=4242)

Step 3: Deploy and register the webhook

For local development, use ngrok:

ngrok http 4242

Then register the webhook in your Stripe Dashboard (Developers → Webhooks → Add endpoint). Set the URL to your ngrok URL + /webhooks/stripe and select these events:

customer.subscription.created
customer.subscription.updated
customer.subscription.deleted
invoice.paid
invoice.payment_failed

For production, deploy this to Railway, Render, or AWS Lambda behind an API Gateway. The handler is stateless — it scales horizontally with no issues.

Step 4: Backfill existing data

Your webhook only catches future events. For existing subscriptions, run a one-time backfill:

# backfill_stripe_to_hubspot.py
import os
import stripe
from dotenv import load_dotenv
from webhook_server import find_hubspot_contact, update_hubspot_contact
 
load_dotenv()
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
 
synced = 0
not_found = 0
 
for subscription in stripe.Subscription.list(status="active", limit=100).auto_paging_iter():
    customer = stripe.Customer.retrieve(subscription.customer)
    if not customer.email:
        continue
 
    contact = find_hubspot_contact(customer.email)
    if not contact:
        not_found += 1
        continue
 
    mrr = sum(
        item.price.unit_amount * item.quantity
        for item in subscription["items"]["data"]
    )
 
    update_hubspot_contact(contact.id, {
        "stripe_customer_id": customer.id,
        "stripe_subscription_status": subscription.status,
        "stripe_mrr": str(mrr),
    })
    synced += 1
 
print(f"Synced: {synced}, Not found in HubSpot: {not_found}")

What this doesn't handle (and what to do about it)

This is a solid foundation, but a production system needs more:

Identity matching beyond email. Not every Stripe customer has the same email as their HubSpot contact. Company domains help, but you'll hit cases where you need fuzzy matching or a manual mapping table.

Rate limiting. HubSpot's API has rate limits (100 calls per 10 seconds on Starter plans). If you process a batch of Stripe events simultaneously, you'll get throttled. Add exponential backoff or a queue.

Retry logic. If HubSpot is down when a Stripe event arrives, you lose that update. In production, push events to a queue (Redis, SQS) and process them with retries.

Multi-subscription customers. A customer with two subscriptions needs their MRR summed, not overwritten. The handler above overwrites — you'd need to query all active subscriptions for that customer on each event.

Historical reconciliation. The webhook tells you about changes going forward. It doesn't tell you about the 47 deals in HubSpot from last year that never matched to any Stripe subscription. You need a scheduled reconciliation job that scans for mismatches.

The maintenance tax

I want to be honest about this: the code above works, and it'll run for months without issues. But it's also something you now own. When HubSpot changes their API version, you update it. When you add a new plan in Stripe, you might need to adjust the MRR calculation. When someone on the team asks "can we also sync the plan name?" you're back in this code.

This is the build-vs-buy tradeoff. Building gives you total control and zero monthly cost. But it costs engineering time on an ongoing basis, and revenue operations isn't a solved-once problem — the edge cases multiply as you scale.

Tools like Fastero let you set up Stripe-to-HubSpot cross-source joins without writing webhook handlers. You define the reconciliation query — "show me closed-won deals with no matching Stripe subscription" — and it runs continuously, alerting you in Slack when something drifts. No deployment, no rate-limit handling, no identity matching code to maintain. Whether that's worth it depends on how much you value your engineering time versus your operational budget.

The code, all in one place

The full working example (webhook server + backfill script + property setup) handles the five most common Stripe events that RevOps teams care about. Fork it, deploy it, and you'll have real-time billing data in HubSpot by end of day. Then decide whether you want to maintain it forever or let something else handle the ongoing reconciliation.

Try Fastero free — connect your CRM and billing data, get live revenue dashboards, and set up alerts that catch leaks before your next board meeting. No credit card required.

Last updated: July 2026.