SDKs

type Init = RequestInit & { json?: unknown };

export class ContactFollowUp {
  constructor(
    private readonly token: string,
    private readonly base = "https://app.contactfollowup.com",
  ) {}

  private async request<T>(path: string, init: Init = {}): Promise<T> {
    const { json, headers, ...rest } = init;
    const res = await fetch(`${this.base}${path}`, {
      ...rest,
      headers: {
        Authorization: `Bearer ${this.token}`,
        "Content-Type": "application/json",
        Accept: "application/json",
        ...(headers ?? {}),
      },
      body: json !== undefined ? JSON.stringify(json) : init.body,
    });
    if (!res.ok) {
      const body = await res.json().catch(() => ({}));
      throw new ContactFollowUpError(res.status, body.error ?? res.statusText, body);
    }
    return res.status === 204 ? (undefined as T) : ((await res.json()) as T);
  }

  contacts = {
    list: (q?: { q?: string; lifecycle?: string; limit?: number }) => {
      const p = new URLSearchParams(
        Object.entries(q ?? {}).filter(([, v]) => v !== undefined) as [string, string][],
      ).toString();
      return this.request<{ contacts: Contact[] }>(`/api/contacts${p ? "?" + p : ""}`);
    },
    create: (body: ContactCreate) =>
      this.request<Contact>("/api/contacts", { method: "POST", json: body }),
    get: (id: string) => this.request<Contact>(`/api/contacts/${id}`),
    update: (id: string, body: Partial<ContactCreate>) =>
      this.request<Contact>(`/api/contacts/${id}`, { method: "PATCH", json: body }),
    remove: (id: string) =>
      this.request<void>(`/api/contacts/${id}`, { method: "DELETE" }),
  };
}

export class ContactFollowUpError extends Error {
  constructor(public readonly status: number, message: string, public readonly body: unknown) {
    super(message);
  }
}

export interface Contact {
  id: string;
  firstName: string;
  lastName: string;
  email: string | null;
  phone: string | null;
  lifecycle: "LEAD" | "PROSPECT" | "PATIENT" | "INACTIVE" | "CHURNED";
}
export interface ContactCreate {
  firstName: string;
  lastName: string;
  email?: string;
  phone?: string;
  lifecycle?: Contact["lifecycle"];
}

import { ContactFollowUp } from "./hearthnote";

const hn = new ContactFollowUp(process.env.HEARTHNOTE_TOKEN!);

const c = await hn.contacts.create({ firstName: "Avery", lastName: "Liu" });
await hn.contacts.update(c.id, { lifecycle: "PATIENT" });

from __future__ import annotations

import os
from typing import Any, Optional
import requests


class ContactFollowUpError(Exception):
    def __init__(self, status: int, message: str, body: Any) -> None:
        super().__init__(message)
        self.status = status
        self.body = body


class ContactFollowUp:
    def __init__(self, token: Optional[str] = None, base: str = "https://app.contactfollowup.com") -> None:
        self.token = token or os.environ["HEARTHNOTE_TOKEN"]
        self.base = base
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.token}",
            "Accept": "application/json",
            "Content-Type": "application/json",
        })

    def _request(self, method: str, path: str, **kwargs: Any) -> Any:
        url = f"{self.base}{path}"
        res = self.session.request(method, url, timeout=30, **kwargs)
        if not res.ok:
            body = res.json() if "application/json" in res.headers.get("content-type", "") else {}
            raise ContactFollowUpError(res.status_code, body.get("error", res.reason), body)
        return None if res.status_code == 204 else res.json()

    # contacts ---------------------------------------------------------------
    def list_contacts(self, **params: Any) -> dict:
        return self._request("GET", "/api/contacts", params=params)

    def create_contact(self, **body: Any) -> dict:
        return self._request("POST", "/api/contacts", json=body)

    def get_contact(self, contact_id: str) -> dict:
        return self._request("GET", f"/api/contacts/{contact_id}")

    def update_contact(self, contact_id: str, **body: Any) -> dict:
        return self._request("PATCH", f"/api/contacts/{contact_id}", json=body)

    def delete_contact(self, contact_id: str) -> None:
        self._request("DELETE", f"/api/contacts/{contact_id}")

from hearthnote import ContactFollowUp

hn = ContactFollowUp()
c = hn.create_contact(firstName="Avery", lastName="Liu")
hn.update_contact(c["id"], lifecycle="PATIENT")