{
  "schema_version": 1,
  "skill_name": "crm",
  "evals": [
    {
      "id": "contact-lookup",
      "prompt": "A user asks: 'Find the contact record for Ada Lovelace in HubSpot and show me her email and company.'",
      "expected_output": "Run crm-cli contacts search --query 'Ada Lovelace' (bounded) and report the matching contact's email, company, and record ID from the summarized properties. If multiple contacts match, list all candidates with IDs and ask the user to disambiguate. Reads only — nothing is changed.",
      "assertions": [
        "Contacts are searched via crm-cli contacts search with a bounded --limit",
        "The response reports email, company, and record ID for the match",
        "Multiple candidates are listed for disambiguation instead of picking arbitrarily",
        "The operation is read-only"
      ]
    },
    {
      "id": "pipeline-view",
      "prompt": "A user asks: 'Show me every deal in the default pipeline that is still in the Appointment Scheduled stage, with amounts.'",
      "expected_output": "Run crm-cli pipelines list to resolve the pipeline and stage IDs, then crm-cli deals list --pipeline <id> --stage <stage-id> (bounded) and report each deal's name, amount, and stage. The response confirms the filter used and notes the total count. Reads only — nothing is moved.",
      "assertions": [
        "Pipeline and stage IDs are resolved via crm-cli pipelines list before filtering",
        "Deals are listed with a pipeline and stage filter and a bounded --limit",
        "Each deal reports name, amount, and stage",
        "The operation is read-only"
      ]
    },
    {
      "id": "stage-update-confirmed",
      "prompt": "A user asks: 'Move deal 901 (Acme renewal) to Closed Won. It is currently in Appointment Scheduled.'",
      "expected_output": "The agent states the current stage and the target, previews the change with crm-cli deals update-stage --id 901 --stage closedwon --dry-run, and asks for explicit confirmation. Only after confirmation does it run the update with --yes, then verifies by re-reading the deal's stage. If the user only asked to draft the change, nothing is applied.",
      "assertions": [
        "The exact deal and target stage are previewed with --dry-run before any change",
        "The current stage and target stage are stated before confirmation",
        "The update runs only after explicit user confirmation, via --yes",
        "The result is verified with a follow-up read of the deal stage"
      ]
    },
    {
      "id": "search-before-mutation",
      "prompt": "A user asks: 'Which deals have \"renewal\" in the name, and should we move the biggest one to Closed Won?'",
      "expected_output": "The agent performs the search as a read: lists deals with renewal-related names via crm-cli deals list and reports names, amounts, and stages, identifying the largest candidate. It does NOT move any deal on its own: a stage change is a guarded mutation requiring explicit confirmation with a --dry-run preview and --yes, so it asks the user to confirm which deal and to which stage before acting.",
      "assertions": [
        "The deal search is performed read-only via crm-cli deals list",
        "The largest candidate is identified with amount and stage",
        "No deal is moved without explicit confirmation and a --dry-run preview",
        "The mutation gate is explained to the user"
      ]
    },
    {
      "id": "pipeline-stage-mapping",
      "prompt": "A user asks: 'What stages exist in our deals pipeline and which one means a deal is won?'",
      "expected_output": "Run crm-cli pipelines list and report each pipeline with its stages: stage label plus stage ID. The response identifies the won/lost stages by their labels (e.g. Closed Won / Closed Lost) and explains that stage changes use the stage ID, not the label, because the API keys on the ID. Reads only.",
      "assertions": [
        "Pipelines and stages are read via crm-cli pipelines list",
        "Each stage is reported with both label and ID",
        "Won/lost stages are identified by label",
        "The response notes that updates use stage IDs, not labels"
      ]
    },
    {
      "id": "access-model-triage",
      "prompt": "A user asks: 'The API returns a 403 when I list deals. Did we lose our deals?'",
      "expected_output": "The response distinguishes the failure modes: a 403 from the HubSpot API means the private app token lacks the object scope (crm.objects.deals.read), not that the deals are gone. It advises checking the private app's scopes in the HubSpot settings, re-generating or re-granting the token scope, then re-running crm-cli deals list. It does not claim data loss, does not re-run blindly, and performs no mutations during triage.",
      "assertions": [
        "A 403 is explained as a scope problem rather than data loss",
        "The fix is checking and re-granting the private app object scopes",
        "Triage is read-only and no mutation is attempted",
        "Data loss is only concluded with concrete evidence"
      ]
    }
  ]
}

# HubSpot CRM — Source Index

> **Last Updated:** 2026-08-03

This skill is a distilled operating layer over HubSpot's public developer documentation. Facts and endpoint names in this skill are grounded in the sources below; refresh this index when HubSpot ships API changes.

| Topic | Source | URL |
|---|---|---|
| CRM object model | Understanding the CRM | https://developers.hubspot.com/docs/api/crm/understanding-the-crm |
| Contacts API | Contacts | https://developers.hubspot.com/docs/api/crm/contacts |
| Deals API | Deals | https://developers.hubspot.com/docs/api/crm/deals |
| Deal pipelines API | Pipelines | https://developers.hubspot.com/docs/api/crm/pipelines |
| Search API | Search | https://developers.hubspot.com/docs/api/crm/search |
| Private apps and scopes | Private apps | https://developers.hubspot.com/docs/api/private-apps |

## Refresh procedure

- Re-check the object model when a 403 or `PROPERTY_DOES_NOT_EXIST` appears for a documented property; HubSpot object schemas evolve.
- Re-check the pipelines API before changing anything in `deals update-stage`; stage IDs are pipeline-scoped.
- Update `research_checked` in `SKILL.md` frontmatter and this file's `Last Updated` when you verify the sources again.

# HubSpot CRM Operations

> **Last Updated:** 2026-08-03

Operational detail for the HubSpot CRM v3 API surface the skill owns: contact records, contact search, deal pipeline views, and guarded deal stage updates. The bundled `crm-cli` implements this reference; use this document when a call behaves unexpectedly.

## API conventions

- Base URL: `https://api.hubapi.com/crm/v3`. Every request carries `Authorization: Bearer <private_app_token>` and JSON bodies.
- Access is scope-based: private app tokens grant per-object read/write. A 403 means the token lacks the scope — check `crm.objects.contacts.read`, `crm.objects.deals.read` (reads) and `crm.objects.deals.write` (stage updates) before concluding anything else.
- `crm-cli` honors `HUBSPOT_API_BASE` (test/stub override); production default is the v3 base.

## Endpoint surface

| Operation | Endpoint | Method | Notes |
|---|---|---|---|
| List contacts | `/objects/contacts?limit=N` | GET | Summarized as name, email, company, createdate |
| Get a contact | `/objects/contacts/{id}` | GET | Single record |
| Search contacts | `/objects/contacts/search` | POST | JSON body `{"query": ..., "limit": N}` |
| List deals | `/objects/deals?limit=N` (+ `pipeline`, `dealstage`) | GET | Pipeline view with amount + stage |
| Move a deal | `/objects/deals/{id}` | PATCH | Guarded mutation: sets the `dealstage` property |
| List pipelines | `/pipelines/deals` | GET | Pipelines with stage IDs and labels |

## Object model

- Every object is an `id` plus a `properties` map keyed by property name. Property values are strings in list/search responses (e.g. `dealstage`, `dealname`, `amount`, `email`, `firstname`, `lastname`, `company`).
- Deals belong to a pipeline (`pipeline` property) and a stage (`dealstage` property) whose valid values come from `/pipelines/deals`. **Stage changes use the stage ID, never the label.**
- Search (`POST /objects/contacts/search`) accepts a `query` string and a `limit`; it matches across default contact searchable properties.

## Pagination and bounded reads

- List endpoints return `results` plus `total` and `paging.next.after` (offset cursor). `limit` caps per-request results (max 100 for most object APIs).
- **Bounded-read rule:** request only what the task needs; `crm-cli --limit` caps at the request level. Report the `total` alongside the returned results so the reader knows the cap hid further records.

## Guarded stage updates

- Moving a deal sets `{"properties": {"dealstage": "<stage-id>"}}` via PATCH. Preview with `--dry-run` (prints the exact deal + target stage), confirm with `--yes`, then verify by re-reading the deal.
- Stage moves are visible to the whole revenue team and land in audit history. They are reversible, but every move is a recorded change — confirm before acting.
- Only the `dealstage` property is in this skill's mutation surface. Other deal property edits are application work.

## Error handling

- 401 `unauthorized`: token invalid or revoked — rotate the private app token.
- 403 `forbidden`: token lacks object scope — grant the scope in the private app settings, don't retry blindly.
- 404 `not found`: object does not exist **or** the token cannot see it — verify object ID and scope before concluding deletion.
- 429 `RATE_LIMIT`: slow down; HubSpot rate limits per token.
- `crm-cli` exit 1 with `HubSpot API HTTP <code>: <message>` (human) or `{"ok": false, "error": "..."}` (JSON). Exit 2 is a usage error.

## Credential hygiene

- Private app tokens are full object-scope credentials: store in `HUBSPOT_TOKEN`, never in code, chat, or commits. Scope tokens to the objects the task needs and rotate on leak.
- Personal data (emails, names, amounts) lives in CRM records; quote only what the question needs and never dump full records into chat.

#!/usr/bin/env python3
"""Deterministic tests for crm/scripts/crm-cli.

Runs the script as a subprocess so the tests exercise the real CLI surface
(--help, --json, --limit, mutation gate, exit codes, JSON payloads). A local
stdlib HTTP server stubs the HubSpot CRM v3 API (contacts, deals, pipelines,
search), so no external network or HubSpot account is needed. Also asserts the
read-only contract: reads never call write methods, and the mutation gate
refuses to move a deal without --dry-run or --yes.
"""
import json
import os
import subprocess
import sys
import threading
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
SCRIPT = ROOT / "scripts" / "crm-cli"

CONTACT = {"id": "51", "properties": {"firstname": "Ada", "lastname": "Lovelace",
                                      "email": "ada@example.com", "company": "Analytical",
                                      "createdate": "2026-01-01T00:00:00Z"}}
DEAL = {"id": "901", "properties": {"dealname": "Acme renewal", "amount": "12000",
                                    "pipeline": "default", "dealstage": "appointmentscheduled",
                                    "hs_lastmodifieddate": "2026-01-02T00:00:00Z"}}
PIPELINE = {"id": "default", "label": "Default pipeline",
            "stages": [{"id": "appointmentscheduled", "label": "Appointment Scheduled",
                        "displayOrder": 0}]}


class StubHubSpotServer:
    """Minimal stub of the HubSpot CRM v3 API surface used by crm-cli."""

    def __init__(self):
        self.requests = []  # (method, path, body)
        handler = self._make_handler()
        self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)

    def _make_handler(self):
        stub = self

        class Handler(BaseHTTPRequestHandler):
            def _read_body(self):
                length = int(self.headers.get("Content-Length", "0"))
                raw = self.rfile.read(length)
                return json.loads(raw.decode("utf-8")) if raw else {}

            def _json(self, payload, status=200):
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.end_headers()
                self.wfile.write(json.dumps(payload).encode("utf-8"))

            def do_GET(self):  # noqa: N802
                stub.requests.append(("GET", self.path, None))
                if "/objects/contacts?" in self.path:
                    self._json({"results": [CONTACT], "total": 1, "paging": {}})
                elif "/objects/contacts/" in self.path:
                    self._json(CONTACT)
                elif "/objects/deals?" in self.path:
                    self._json({"results": [DEAL], "total": 1, "paging": {}})
                elif "/pipelines/deals" in self.path:
                    self._json({"results": [PIPELINE]})
                else:
                    self._json({"message": "not_found"}, 404)

            def do_POST(self):  # noqa: N802
                body = self._read_body()
                stub.requests.append(("POST", self.path, body))
                if "/search" in self.path:
                    limit = body.get("limit", 20)
                    self._json({"results": [CONTACT][:limit], "total": 1})
                else:
                    self._json({"message": "not_found"}, 404)

            def do_PATCH(self):  # noqa: N802
                body = self._read_body()
                stub.requests.append(("PATCH", self.path, body))
                if "/objects/deals/" in self.path:
                    updated = json.loads(json.dumps(DEAL))
                    updated["properties"]["dealstage"] = body.get("properties", {}).get("dealstage")
                    self._json(updated)
                else:
                    self._json({"message": "not_found"}, 404)

            def log_message(self, *args):  # silence stderr
                pass

        return Handler

    def __enter__(self):
        self.thread.start()
        return self

    def __exit__(self, *exc):
        self.server.shutdown()
        self.server.server_close()


def run_script(env, *args):
    return subprocess.run(
        [sys.executable, str(SCRIPT), *args],
        capture_output=True,
        text=True,
        timeout=30,
        env=env,
    )


def base_env(stub):
    env = dict(os.environ)
    env["HUBSPOT_TOKEN"] = "pat-test"
    env["HUBSPOT_API_BASE"] = f"http://127.0.0.1:{stub.port}/"
    return env


def load_json(proc):
    return json.loads(proc.stdout)


class CrmCliTests(unittest.TestCase):
    def test_help_lists_json_and_bounded_reads(self):
        proc = run_script(dict(os.environ), "--help")
        self.assertEqual(proc.returncode, 0)
        self.assertIn("--json", proc.stdout)
        self.assertIn("--limit", proc.stdout)

    def test_help_works_without_token(self):
        env = dict(os.environ)
        env.pop("HUBSPOT_TOKEN", None)
        proc = run_script(env, "contacts", "list", "--help")
        self.assertEqual(proc.returncode, 0)

    def test_contacts_list(self):
        with StubHubSpotServer() as stub:
            proc = run_script(base_env(stub), "--json", "--limit", "5", "contacts", "list")
        self.assertEqual(proc.returncode, 0, proc.stderr)
        data = load_json(proc)
        self.assertEqual(data["contacts"][0]["email"], "ada@example.com")

    def test_contacts_get(self):
        with StubHubSpotServer() as stub:
            proc = run_script(base_env(stub), "--json", "contacts", "get", "--id", "51")
        self.assertEqual(proc.returncode, 0, proc.stderr)
        self.assertEqual(load_json(proc)["contact"]["id"], "51")

    def test_contacts_search(self):
        with StubHubSpotServer() as stub:
            proc = run_script(base_env(stub), "--json", "contacts", "search", "--query", "ada")
        self.assertEqual(proc.returncode, 0, proc.stderr)
        data = load_json(proc)
        self.assertEqual(data["total"], 1)
        self.assertEqual(data["contacts"][0]["firstname"], "Ada")

    def test_search_sends_limit(self):
        with StubHubSpotServer() as stub:
            run_script(base_env(stub), "--json", "--limit", "3", "contacts", "search",
                       "--query", "ada")
        posts = [body for method, path, body in stub.requests
                 if method == "POST" and "/search" in path]
        self.assertEqual(len(posts), 1)
        self.assertEqual(posts[0].get("limit"), 3)

    def test_deals_list_pipeline_view(self):
        with StubHubSpotServer() as stub:
            proc = run_script(base_env(stub), "--json", "deals", "list",
                              "--pipeline", "default", "--stage", "appointmentscheduled")
        self.assertEqual(proc.returncode, 0, proc.stderr)
        data = load_json(proc)
        self.assertEqual(data["deals"][0]["dealname"], "Acme renewal")
        gets = [path for method, path, _ in stub.requests if method == "GET"]
        self.assertTrue(any("objects/deals?" in path for path in gets))

    def test_pipelines_list(self):
        with StubHubSpotServer() as stub:
            proc = run_script(base_env(stub), "--json", "pipelines", "list")
        self.assertEqual(proc.returncode, 0, proc.stderr)
        data = load_json(proc)
        self.assertEqual(data["pipelines"][0]["id"], "default")
        self.assertEqual(data["pipelines"][0]["stages"][0]["id"], "appointmentscheduled")

    def test_update_stage_requires_confirmation(self):
        with StubHubSpotServer() as stub:
            proc = run_script(base_env(stub), "deals", "update-stage", "--id", "901",
                              "--stage", "closedwon")
        self.assertEqual(proc.returncode, 1)
        self.assertIn("refusing to move", proc.stderr)
        self.assertEqual(stub.requests, [], "no API call may be made without confirmation")

    def test_update_stage_dry_run_does_not_patch(self):
        with StubHubSpotServer() as stub:
            proc = run_script(base_env(stub), "--json", "deals", "update-stage", "--id", "901",
                              "--stage", "closedwon", "--dry-run")
        self.assertEqual(proc.returncode, 0, proc.stderr)
        data = load_json(proc)
        self.assertTrue(data["dry_run"])
        self.assertEqual(stub.requests, [], "dry-run must not reach the API")

    def test_update_stage_with_yes_patches(self):
        with StubHubSpotServer() as stub:
            proc = run_script(base_env(stub), "--json", "deals", "update-stage", "--id", "901",
                              "--stage", "closedwon", "--yes")
        self.assertEqual(proc.returncode, 0, proc.stderr)
        data = load_json(proc)
        self.assertEqual(data["deal"]["dealstage"], "closedwon")
        self.assertTrue(any(method == "PATCH" for method, _path, _body in stub.requests))

    def test_missing_token_errors_cleanly(self):
        env = dict(os.environ)
        env.pop("HUBSPOT_TOKEN", None)
        proc = run_script(env, "--json", "contacts", "list")
        self.assertEqual(proc.returncode, 1)
        self.assertIn("HUBSPOT_TOKEN", proc.stdout)

    def test_read_only_contract_no_write_opens(self):
        source = SCRIPT.read_text()
        writes = [line for line in source.splitlines()
                  if line.strip().startswith("open(") and ("'w'" in line or '"w"' in line)]
        self.assertEqual(writes, [], "script must never open files in write mode")


if __name__ == "__main__":
    unittest.main()