Connect with OAuth

curl https://app.bloomtext.com/oauth/token \
  -u "$BLOOMTEXT_CLIENT_ID:$BLOOMTEXT_CLIENT_SECRET" \
  -d grant_type=authorization_code \
  -d code=SplxlOBeZQQYbYS6WxSbIA \
  -d redirect_uri=https://ehr.example.com/bloomtext/callback \
  -d code_verifier=dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk

import crypto from 'node:crypto'
 
const AUTHORIZE_URL = 'https://app.bloomtext.com/oauth/authorize'
const TOKEN_URL = 'https://app.bloomtext.com/oauth/token'
const REDIRECT_URI = 'https://ehr.example.com/bloomtext/callback'
 
// Step 1: build the authorize URL. Store state and verifier with the session.
export function startAuthorization(session) {
  session.state = crypto.randomBytes(16).toString('base64url')
  session.verifier = crypto.randomBytes(32).toString('base64url')
  const challenge = crypto.createHash('sha256').update(session.verifier).digest('base64url')
 
  const url = new URL(AUTHORIZE_URL)
  url.search = new URLSearchParams({
    response_type: 'code',
    client_id: process.env.BLOOMTEXT_CLIENT_ID,
    redirect_uri: REDIRECT_URI,
    scope: 'messages.read offline_access',
    state: session.state,
    code_challenge: challenge,
    code_challenge_method: 'S256',
  })
  return url.toString()
}
 
// Step 4: exchange the code from the redirect for tokens.
export async function finishAuthorization(session, query) {
  if (query.error) throw new Error(`Not connected: ${query.error_description ?? query.error}`)
  if (query.state !== session.state) throw new Error('State mismatch')
 
  const credentials = Buffer.from(`${process.env.BLOOMTEXT_CLIENT_ID}:${process.env.BLOOMTEXT_CLIENT_SECRET}`).toString('base64')
  const response = await fetch(TOKEN_URL, {
    method: 'POST',
    headers: { Authorization: `Basic ${credentials}` },
    body: new URLSearchParams({
      grant_type: 'authorization_code',
      code: query.code,
      redirect_uri: REDIRECT_URI,
      code_verifier: session.verifier,
    }),
  })
  if (!response.ok) throw new Error(`Token exchange failed: ${response.status}`)
  return response.json() // { access_token, refresh_token, expires_in, scope, token_type }
}

import base64
import hashlib
import os
import secrets
from urllib.parse import urlencode
 
import requests
 
AUTHORIZE_URL = "https://app.bloomtext.com/oauth/authorize"
TOKEN_URL = "https://app.bloomtext.com/oauth/token"
REDIRECT_URI = "https://ehr.example.com/bloomtext/callback"
 
 
def start_authorization(session: dict) -> str:
    """Step 1: build the authorize URL. Store state and verifier with the session."""
    session["state"] = secrets.token_urlsafe(16)
    session["verifier"] = secrets.token_urlsafe(32)
    digest = hashlib.sha256(session["verifier"].encode()).digest()
    challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
 
    return AUTHORIZE_URL + "?" + urlencode({
        "response_type": "code",
        "client_id": os.environ["BLOOMTEXT_CLIENT_ID"],
        "redirect_uri": REDIRECT_URI,
        "scope": "messages.read offline_access",
        "state": session["state"],
        "code_challenge": challenge,
        "code_challenge_method": "S256",
    })
 
 
def finish_authorization(session: dict, query: dict) -> dict:
    """Step 4: exchange the code from the redirect for tokens."""
    if "error" in query:
        raise RuntimeError(f"Not connected: {query.get('error_description', query['error'])}")
    if query["state"] != session["state"]:
        raise RuntimeError("State mismatch")
 
    response = requests.post(
        TOKEN_URL,
        auth=(os.environ["BLOOMTEXT_CLIENT_ID"], os.environ["BLOOMTEXT_CLIENT_SECRET"]),
        data={
            "grant_type": "authorization_code",
            "code": query["code"],
            "redirect_uri": REDIRECT_URI,
            "code_verifier": session["verifier"],
        },
    )
    response.raise_for_status()
    return response.json()  # access_token, refresh_token, expires_in, scope, token_type