The tools businesses actually run

An AI feature delivers value only when it's wired into the systems where work happens:

• CRM — Salesforce, HubSpot, Pipedrive: leads, contacts, deals. (Your Eloqua/CRM background lives here.)
• Helpdesk / ticketing — Zendesk, Intercom, Freshdesk: support tickets and conversations.
• Data hubs / warehouses — Snowflake, BigQuery, Postgres: the analytical source of truth.
• Comms — Slack, Teams, email: where humans get notified and act.

Each exposes a REST API (and usually webhooks) — so everything from 9.1–9.2 applies directly. But each also has its own auth, data model, and rate limits, and that friction is most of the real work.

The connector reality

Before writing a line, know what every business-tool integration throws at you:

• Auth is usually OAuth — not a simple API key. The user authorizes your app; you get access + refresh tokens and must refresh them (this is exactly the Microsoft-Graph OAuth pain that shows up in real N8N work — the token/consent must be for the right account).
• Rate limits are real and per-tool — Salesforce, Zendesk, etc. cap requests; hit them and you get 429s. Handle with the backoff + queue patterns from 9.2.
• Data models are opinionated — Salesforce's Lead/Contact/Opportunity objects, custom fields, required fields. You map to their schema, not yours.
• Sync direction & webhooks — read via API, react via webhook (new lead, updated ticket). Many tools push events you subscribe to rather than making you poll.

Reading and writing a business system

The core loop is: read context → run AI → write back. Here's the shape against a CRM (the SDK/endpoints vary by tool, the pattern doesn't):

def enrich_lead(lead_id: str):
    lead = crm.get(f"/leads/{lead_id}", auth=oauth_token())   # READ the record (OAuth-authenticated)
    summary = llm(f"Summarize and score this lead:\n{lead}")  # run AI on the business data
    crm.patch(f"/leads/{lead_id}", auth=oauth_token(), json={ # WRITE the result back into THEIR fields
        "ai_summary__c": summary.text,          # a custom field (Salesforce '__c' suffix) you created
        "ai_score__c": summary.score,           # map your output onto their schema
    })

The AI part is one line; the work is authenticating, reading the right record, and writing to the right fields in the tool's data model. That's the connector.

The enrich-on-event pattern

The highest-value connector pattern combines all of Track 9: a webhook fires on a business event, you queue it, a worker runs AI and writes back to the source system — reliably, idempotently:

# 1. Webhook: the CRM notifies you a new lead was created (9.1) — verify + enqueue + ack fast.
@app.post("/webhooks/new-lead")
async def on_new_lead(request: Request):
    event = await verified_event(request)                 # verify signature (9.1)
    queue.put({"id": event["id"], "lead_id": event["data"]["id"]})   # enqueue (9.2)
    return {"queued": True}                                # ack fast

# 2. Worker: process off the request path, idempotently, writing back to the CRM.
def worker():
    while True:
        msg = queue.get()
        if already_processed(msg["id"]):                  # idempotency (9.1/9.2): a re-delivered event is a no-op
            queue.ack(msg); continue
        with_retry(lambda: enrich_lead(msg["lead_id"]))   # AI enrich + write-back, retrying transient 429s (9.2)
        mark_processed(msg["id"]); queue.ack(msg)

New lead → AI summarizes and scores it → the score appears on the sales rep's CRM record automatically. That's a complete, production-shaped AI integration — and it's built entirely from Track 9's primitives plus the AI from Tracks 1–5.

Data mapping

Business systems have their own schema, and your AI output must fit it. Mapping is unglamorous and where integrations quietly break:

• Field mapping — AI category → the tool's picklist value (and their picklist may not contain your value — handle the mismatch).
• Required fields & validation — the tool rejects a write missing a required field or violating a rule; validate before writing (Track 4 argument validation applies).
• Types & formats — dates, currencies, enums must match the tool's expectations.
• Idempotent writes — use the tool's upsert/external-id features so a retried write updates rather than duplicates (9.2 idempotency, tool-native).

CATEGORY_TO_CRM = {"billing": "Billing_Issue", "tech": "Technical_Support"}   # map YOUR labels to THEIR picklist
def to_crm_fields(ai_result: dict) -> dict:
    category = CATEGORY_TO_CRM.get(ai_result["category"], "Other")   # fall back safely on an unmapped value
    return {"Case_Type__c": category, "Priority__c": ai_result["priority"].upper()}   # their field names + formats

Choosing the connector approach

The code-vs-no-code judgment (Track 4) applied to connectors:

Approach	Best when
iPaaS / no-code (n8n, Zapier, Make)	standard tools, standard events, non-engineer maintainable, fast to ship — most connectors
Native SDK / custom code	complex logic, high volume, unusual data mapping, or the tool's own API is needed for something the connector can't do
Hybrid (Track 4)	no-code owns the connector/trigger; your deployed code service does the AI — often the best of both

For most business-tool connectors, an iPaaS platform already has the connector built (auth, pagination, rate limits handled) — reinventing it in code is wasted effort. Reach for custom code when the logic or volume outgrows the no-code tool, and prefer the hybrid: let n8n/Zapier own the Salesforce/Zendesk connection and call your code service for the AI. This is exactly the pattern from the Code-vs-No-code and Discovery topics, now concrete.

Pitfalls

• Underestimating auth. OAuth refresh and wrong-account consent break connectors in production — budget for it.
• Ignoring rate limits. Per-tool caps cause 429s under load — use backoff + queues (9.2).
• Mapping to your schema, not theirs. You must write to the tool's fields/picklists/required-fields, or writes fail.
• Non-idempotent writes. Retries duplicate records — use the tool's upsert/external-id (9.2 idempotency).
• Coding a connector that already exists. iPaaS platforms ship maintained connectors — don't rebuild them; use the hybrid.

Recap

• AI delivers value inside the business's tools — CRM, helpdesk, data hubs, Slack — each via REST APIs + webhooks (Track 9.1–9.2).
• The connector reality is OAuth, per-tool rate limits, and opinionated data models — auth and limits are the demo-to-production gap, not the AI.
• The enrich-on-event pattern (webhook → queue → idempotent worker → write-back) is a complete, production- shaped AI integration built from Track 9's primitives.
• Map AI output to the tool's schema (fields, picklists, required fields, formats) and make writes idempotent with upsert/external-id.
• Choose iPaaS for standard connectors, custom code for complex/high-volume, and the hybrid (no-code connector + code AI service) most often.
• Worked examples map to the exercises: read→AI→write-back (Ex 1), enrich-on-event pipeline (Ex 2), data mapping (Ex 3), connector-approach choice (Ex 4).

def enrich_lead(lead_id):
    token = oauth_token()                                  # OAuth (with refresh) — the real work
    lead = crm.get(f"/leads/{lead_id}", auth=token)        # READ (counts against the rate limit)
    result = llm(f"Summarize and score this lead:\n{lead}")
    crm.patch(f"/leads/{lead_id}", auth=token, json={      # WRITE back to THEIR fields
        "ai_summary__c": result.text,
        "ai_score__c": result.score,
    })

@app.post("/webhooks/new-ticket")
async def on_new_ticket(request):
    event = await verified_event(request)                  # verify signature (9.1)
    queue.put({"id": event["id"], "ticket_id": event["data"]["id"]})
    return {"queued": True}                                # ack fast

def worker():
    while True:
        msg = queue.get()
        if already_processed(msg["id"]):                   # idempotency — re-delivered event is a no-op
            queue.ack(msg); continue
        with_retry(lambda: classify_and_write(msg["ticket_id"]))   # retry transient 429s (9.2)
        mark_processed(msg["id"]); queue.ack(msg)

def classify_and_write(ticket_id):
    t = helpdesk.get(f"/tickets/{ticket_id}", auth=oauth_token())
    result = my_classifier(t["body"])
    helpdesk.patch(f"/tickets/{ticket_id}", auth=oauth_token(),
                   json=to_helpdesk_fields(result))        # map to THEIR schema (Ex 3)

CATEGORY_MAP = {"billing": "Billing_Issue", "tech": "Technical_Support"}

def to_helpdesk_fields(ai):
    category = CATEGORY_MAP.get(ai["category"], "Other")   # unmapped -> safe fallback, not a failed write
    fields = {
        "case_type": category,
        "priority": ai["priority"].upper(),                # match THEIR expected format (enum casing)
        "status": "open",                                  # a REQUIRED field the tool rejects writes without
    }
    return fields