HubSpot Service Hub

HubSpot Service Hub uses Tickets and Conversations as its core objects. This guide walks through connecting every Simpli service to HubSpot using workflow automation, webhooks, and the HubSpot API v3.

HubSpot remains your system of record. Simpli adds AI capabilities alongside it — auto-classifying tickets, drafting responses, scoring conversations, and tracking customer health.

Prerequisites

HubSpot Service Hub (Professional or Enterprise plan required for workflow automation with webhooks)
A Private App with an API token — create one at Settings > Integrations > Private Apps
AI capabilities deployed and running
A middleware endpoint — your own server or serverless function that sits between HubSpot and Simpli

Architecture

┌──────────┐    workflow webhook    ┌──────────────┐     REST API      ┌─────────────────┐
│ HubSpot  │ ────────────────────► │  Middleware   │ ────────────────► │ Simpli Services  │
│ Service  │ ◄──────────────────── │  (your code)  │ ◄──────────────── │ Triage, Reply... │
│ Hub      │    HubSpot API v3     └──────────────┘    AI results      └─────────────────┘
└──────────┘

HubSpot workflows trigger webhook actions on ticket and conversation events. Your middleware receives these, calls the relevant Simpli APIs, and pushes results back to HubSpot via its API v3.

Authentication

HubSpot Private Apps use a Bearer token in the Authorization header.

import httpx

HUBSPOT_TOKEN = "your-private-app-token"
HUBSPOT_BASE = "https://api.hubapi.com"
HUBSPOT_HEADERS = {
    "Authorization": f"Bearer {HUBSPOT_TOKEN}",
    "Content-Type": "application/json",
}

When creating your Private App, grant these scopes:

tickets (read and write)
crm.objects.contacts.read and crm.objects.contacts.write
conversations.read
content (for Knowledge Base access)

All API examples below use these constants.

Key concepts

HubSpot organizes service data differently from traditional helpdesks:

Tickets live in a pipeline with configurable pipeline stages (e.g., New, Waiting on contact, Waiting on us, Closed).
Conversations are the messaging thread (email, chat, etc.) associated with a ticket.
Contacts are your customers. Tickets are associated with contacts via the CRM associations API.
Custom properties on tickets and contacts store additional data. Properties must be created before use.
Workflows automate actions based on ticket or contact events. The webhook action sends data to external URLs.

Custom properties

Before integrating, create custom properties in Settings > Properties.

Ticket properties

Property name	Internal name	Type	Purpose
AI Category	`ai_category`	Dropdown	Triage classification category
AI Urgency	`ai_urgency`	Dropdown	Urgency level from Triage
AI Confidence	`ai_confidence`	Number	Classification confidence (0-100)
QA Score	`qa_score`	Number	Overall QA evaluation score (0-100)

For dropdown properties, add options: billing, technical, account, feature_request, general for AI Category and low, medium, high, critical for AI Urgency.

Contact properties

Property name	Internal name	Type	Purpose
Sentiment Score	`sentiment_score`	Dropdown	`positive`, `neutral`, `negative`
Escalation Risk	`escalation_risk`	Number	Risk score from 0 to 100

Create these via the API or the HubSpot UI.

Per-service integration

Triage — auto-classify and route tickets

When: A new ticket is created in HubSpot.

Flow: A workflow triggers on ticket creation and fires a webhook to your middleware. Middleware calls Triage /classify and /route, then updates the HubSpot ticket properties and moves it to the appropriate pipeline stage.

TRIAGE_URL = "http://localhost:8001"

async def classify_and_route(ticket_id: str, subject: str, description: str):
    async with httpx.AsyncClient() as client:
        # Classify the ticket
        classify_resp = await client.post(f"{TRIAGE_URL}/classify", json={
            "subject": subject,
            "body": description,
        })
        classification = classify_resp.json()

        # Route based on classification
        route_resp = await client.post(f"{TRIAGE_URL}/route", json={
            "ticket_id": ticket_id,
            "category": classification["category"],
            "urgency": classification["urgency"],
            "sentiment": classification["sentiment"],
        })
        routing = route_resp.json()

        # Map urgency to HubSpot priority
        priority_map = {"low": "LOW", "medium": "MEDIUM", "high": "HIGH", "critical": "HIGH"}

        # Update the HubSpot ticket
        await client.patch(
            f"{HUBSPOT_BASE}/crm/v3/objects/tickets/{ticket_id}",
            headers=HUBSPOT_HEADERS,
            json={
                "properties": {
                    "ai_category": classification["category"],
                    "ai_urgency": classification["urgency"],
                    "ai_confidence": str(round(classification["confidence"] * 100)),
                    "hs_ticket_priority": priority_map.get(
                        classification["urgency"], "MEDIUM"
                    ),
                }
            },
        )

        # Assign to owner based on routing
        if routing.get("agent_id"):
            await client.patch(
                f"{HUBSPOT_BASE}/crm/v3/objects/tickets/{ticket_id}",
                headers=HUBSPOT_HEADERS,
                json={
                    "properties": {
                        "hubspot_owner_id": routing["agent_id"],
                    }
                },
            )

    return {"classification": classification, "routing": routing}

HubSpot ticket properties use string values even for numbers. The hs_ticket_priority property accepts LOW, MEDIUM, or HIGH.

Reply — AI draft responses

When: After Triage classifies a ticket (chain this in the same workflow).

Flow: Middleware calls Reply /api/v1/draft and creates a note on the ticket with the draft. Agents see the note and can use the draft as a starting point for their reply.

REPLY_URL = "http://localhost:8002"

async def draft_reply(ticket_id: str, conversation: list[dict]):
    async with httpx.AsyncClient() as client:
        # Generate a draft
        draft_resp = await client.post(f"{REPLY_URL}/api/v1/draft", json={
            "ticket_id": ticket_id,
            "conversation": conversation,
            "style": "friendly",
            "language": "en",
        })
        draft = draft_resp.json()

        # Create a note (engagement) associated with the ticket
        note_resp = await client.post(
            f"{HUBSPOT_BASE}/crm/v3/objects/notes",
            headers=HUBSPOT_HEADERS,
            json={
                "properties": {
                    "hs_timestamp": str(int(__import__("time").time() * 1000)),
                    "hs_note_body": (
                        f"<strong>AI Draft Response</strong> "
                        f"(confidence: {draft['confidence']:.0%})<br><br>"
                        f"{draft['draft']}<br><br>"
                        f"<em>Copy and edit this draft before sending.</em>"
                    ),
                },
            },
        )
        note_id = note_resp.json()["id"]

        # Associate the note with the ticket
        await client.put(
            f"{HUBSPOT_BASE}/crm/v3/objects/notes/{note_id}/associations/tickets/{ticket_id}/note_to_ticket",
            headers=HUBSPOT_HEADERS,
        )

    return draft

HubSpot notes support HTML formatting. The note is visible on the ticket timeline to all agents.

Sentiment — customer health tracking

When: A conversation message is received from a customer.

Flow: A workflow triggers on new conversation activity. Middleware calls Sentiment /analyze and updates a custom property on the associated contact.

SENTIMENT_URL = "http://localhost:8004"

async def analyze_message(ticket_id: str, contact_id: str, message_text: str):
    async with httpx.AsyncClient() as client:
        # Analyze sentiment
        sentiment_resp = await client.post(f"{SENTIMENT_URL}/analyze", json={
            "customer_id": contact_id,
            "text": message_text,
            "channel": "web",
        })
        sentiment = sentiment_resp.json()

        # Update contact properties
        await client.patch(
            f"{HUBSPOT_BASE}/crm/v3/objects/contacts/{contact_id}",
            headers=HUBSPOT_HEADERS,
            json={
                "properties": {
                    "sentiment_score": sentiment["label"],
                    "escalation_risk": str(
                        round(sentiment["escalation_risk"] * 100)
                    ),
                },
            },
        )

        # If high escalation risk, bump ticket priority
        if sentiment["escalation_risk"] >= 0.5:
            await client.patch(
                f"{HUBSPOT_BASE}/crm/v3/objects/tickets/{ticket_id}",
                headers=HUBSPOT_HEADERS,
                json={
                    "properties": {
                        "hs_ticket_priority": "HIGH",
                    }
                },
            )

    return sentiment

Build a HubSpot active list filtering on escalation_risk > 50 to surface at-risk contacts for your support lead.

QA — conversation scoring

When: A ticket is moved to the Closed pipeline stage.

Flow: A workflow triggers when the ticket pipeline stage changes to Closed. Middleware pulls the conversation thread, sends it to QA /evaluate, and updates the ticket with the score.

QA_URL = "http://localhost:8003"

async def score_conversation(ticket_id: str):
    async with httpx.AsyncClient() as client:
        # Get the conversation thread ID associated with the ticket
        assoc_resp = await client.get(
            f"{HUBSPOT_BASE}/crm/v3/objects/tickets/{ticket_id}/associations/conversations",
            headers=HUBSPOT_HEADERS,
        )
        associations = assoc_resp.json().get("results", [])

        if not associations:
            return None

        thread_id = associations[0]["id"]

        # Pull conversation messages
        thread_resp = await client.get(
            f"{HUBSPOT_BASE}/conversations/v3/conversations/threads/{thread_id}/messages",
            headers=HUBSPOT_HEADERS,
        )
        thread_messages = thread_resp.json().get("results", [])

        # Convert to Simpli message format
        messages = []
        agent_id = ""
        for msg in thread_messages:
            sender_type = msg.get("senders", [{}])[0].get("actorId", "")
            # HubSpot distinguishes VISITOR from AGENT senders
            role = "customer" if "VISITOR" in str(msg.get("type", "")) else "agent"
            messages.append({
                "role": role,
                "content": msg.get("text", msg.get("richText", "")),
            })
            if role == "agent":
                agent_id = str(sender_type)

        # Evaluate the conversation
        qa_resp = await client.post(f"{QA_URL}/evaluate", json={
            "conversation_id": ticket_id,
            "agent_id": agent_id,
            "messages": messages,
        })
        evaluation = qa_resp.json()

        # Update ticket with QA score
        await client.patch(
            f"{HUBSPOT_BASE}/crm/v3/objects/tickets/{ticket_id}",
            headers=HUBSPOT_HEADERS,
            json={
                "properties": {
                    "qa_score": str(round(evaluation["overall_score"] * 100)),
                }
            },
        )

        # Also create a detailed note
        dimensions = evaluation.get("dimensions", {})
        dimension_lines = "".join(
            f"<li><strong>{k.title()}</strong>: {v:.0%}</li>"
            for k, v in dimensions.items()
        )
        note_resp = await client.post(
            f"{HUBSPOT_BASE}/crm/v3/objects/notes",
            headers=HUBSPOT_HEADERS,
            json={
                "properties": {
                    "hs_timestamp": str(int(__import__("time").time() * 1000)),
                    "hs_note_body": (
                        f"<strong>QA Score: {evaluation['overall_score']:.0%}</strong>"
                        f"<ul>{dimension_lines}</ul>"
                        f"<p><strong>Coaching notes:</strong> "
                        f"{', '.join(evaluation.get('coaching_notes', []))}</p>"
                    ),
                },
            },
        )
        note_id = note_resp.json()["id"]

        await client.put(
            f"{HUBSPOT_BASE}/crm/v3/objects/notes/{note_id}/associations/tickets/{ticket_id}/note_to_ticket",
            headers=HUBSPOT_HEADERS,
        )

    return evaluation

KB — article sync

Sync your HubSpot Knowledge Base articles into Simpli KB for semantic search and gap analysis.

KB_URL = "http://localhost:8006"

async def sync_hubspot_articles():
    async with httpx.AsyncClient() as client:
        offset = 0
        limit = 50

        while True:
            resp = await client.get(
                f"{HUBSPOT_BASE}/cms/v3/blogs/posts",
                headers=HUBSPOT_HEADERS,
                params={
                    "offset": offset,
                    "limit": limit,
                    "state": "PUBLISHED",
                    "content_group_id": "your-kb-content-group-id",
                },
            )
            data = resp.json()
            results = data.get("results", [])

            if not results:
                break

            for article in results:
                await client.post(f"{KB_URL}/articles", json={
                    "title": article["name"],
                    "content": article.get("postBody", ""),
                    "tags": [
                        tag["name"]
                        for tag in article.get("tagIds", [])
                    ],
                    "source": "hubspot-kb",
                })

            offset += limit
            if offset >= data.get("total", 0):
                break

        # Check for content gaps
        gaps_resp = await client.get(f"{KB_URL}/gaps")
        return gaps_resp.json()

Replace your-kb-content-group-id with the content group ID of your Knowledge Base. Find it in Settings > Website > Blog > Knowledge Base or via the API.

Workflow configuration

1. Create the ticket classification workflow

In Automation > Workflows, create a new ticket-based workflow:

Trigger: Ticket is created
Action: Send a webhook
- Method: POST
- URL: https://your-server.example.com/hubspot/ticket-created
- Request body: Include hs_object_id (ticket ID), subject, and content properties

2. Create the QA scoring workflow

Create another ticket-based workflow:

Trigger: Ticket pipeline stage changes to "Closed"
Action: Send a webhook
- Method: POST
- URL: https://your-server.example.com/hubspot/ticket-closed
- Request body: Include hs_object_id

3. Create the sentiment workflow

For conversation-based sentiment analysis, create a contact-based workflow:

Trigger: Contact has a new conversation (or use ticket-based with "New message received")
Action: Send a webhook
- Method: POST
- URL: https://your-server.example.com/hubspot/message-received
- Request body: Include contact ID and message content

Webhook action format

HubSpot workflow webhooks send a JSON payload. Your middleware receives something like:

{
  "hs_object_id": "12345",
  "subject": "Cannot access my account",
  "content": "I have been trying to log in for the past hour..."
}

Parse the payload in your middleware and call the appropriate Simpli service.

Pipeline and stage configuration

HubSpot tickets move through pipeline stages. A common setup:

Stage	Internal value	Simpli action
New	`1`	Triage classifies and routes
Waiting on contact	`2`	Sentiment analyzes replies
Waiting on us	`3`	Reply generates drafts
Closed	`4`	QA evaluates the conversation

Map your pipeline stages to Simpli workflows in the HubSpot workflow triggers.

Rate limits and best practices

HubSpot API rate limits depend on your plan and authentication method. Private Apps typically allow 100 requests per 10 seconds.

Handle 429 responses. Back off when rate-limited:

resp = await client.patch(url, headers=HUBSPOT_HEADERS, json=data)
if resp.status_code == 429:
    retry_after = int(resp.headers.get("Retry-After", 10))
    await asyncio.sleep(retry_after)
    resp = await client.patch(url, headers=HUBSPOT_HEADERS, json=data)

Use batch endpoints. When updating multiple tickets, use POST /crm/v3/objects/tickets/batch/update instead of individual PATCH requests.
Create properties before use. Unlike some platforms, HubSpot requires custom properties to exist before you can set values on them. Create them via the UI or API during setup.
Keep middleware idempotent. HubSpot may retry webhook actions on failure. Use the ticket ID as an idempotency key.
Scope your Private App tightly. Only grant the scopes your integration needs. Review scopes in Settings > Integrations > Private Apps.

Next steps

Integration Overview -- Architecture and other platform guides
Generic Webhook -- Build a platform-agnostic integration

On this page