Authorization | Boosted.ai

Overview

Token Exchange enables your backend to obtain Boosted access tokens for your users without browser redirects or user interaction. This method is designed for backend-to-backend integrations where you embed Boosted features directly in your application.

Authentication Flow:

User is authenticated in your application
User requests access to Boosted features
1. If it is their first time, perform token exchange with sub set to your client ID to obtain a service account token
2. Use the management token to create the account hierarchy (Organization → Workspace → User)
Your backend creates a signed JWT asserting the user’s identity
Your backend sends JWT to Boosted’s token endpoint
Boosted validates JWT signature and returns access tokens
Your backend uses access tokens to call Boosted APIs on user’s behalf

No user redirects, login screens, or consent flow.

Standards:

What You Need to Provide Us

1. Public Signing Key URL (`jwks_uri`)

Provide the public URL where your JWKS (JSON Web Key Set) is hosted. We will fetch your public keys from this endpoint to verify signatures.

How to generate and host:

Generate Keys:

# Generate private key (keep this secret)
openssl genrsa -out private_key.pem 2048
# Extract public key
openssl rsa -in private_key.pem -pubout -out public_key.pem

Convert & Host:
- Convert the public key to JWKS format (JSON).
- Upload the JSON file to a publicly accessible location (e.g., AWS S3, a static site, or a dedicated API endpoint).

Example URL: https://your-domain.com/.well-known/jwks.json

Expected Response (at the URL above):

{
  "keys": [
    {
      "kty": "RSA",
      "alg": "RS256",
      "use": "sig",
      "kid": "your-key-id-1",
      "n": "oO2Re_MzrCqR-1rLFcmuZqf2kYmqjWQax1gYo-cWIGEDX_UIvQ4FtDRx53fPfbl...",
      "e": "AQAB"
    }
  ]
}

What We Will Provide You

1. Client ID

your_client_id_abc123

Used for:

Token exchange request body
JWT iss claim (issuer)
JWT sub claim (for token exchange to use the account-management API)

2. Client Secret

your_client_secret_xyz789

Used for:

Token exchange request body

3. Token Endpoint

https://sandbox.api.boosted.ai/oauth2/token

Send your token exchange requests here.

4. Account Management API Endpoints

https://sandbox.api.boosted.ai/v2/account-management/org
https://sandbox.api.boosted.ai/v2/account-management/workspace
https://sandbox.api.boosted.ai/v2/account-management/user

Create the organizational hierarchy via these endpoints before performing token exchange for users.

5. Audience Value

https://sandbox.api.boosted.ai

Include this in the aud claim of your JWT.

6. JWT Requirements

Your JWT must include these standard claims:

{
  "iss": "your_client_id_abc123",                    // Issuer: Your client ID (who signed this)
  "sub": "user_id_from_api",                         // Subject: Client ID or user identifier (we returned this when you created the user)
  "aud": "https://sandbox.api.boosted.ai",           // Audience: Boosted's server
  "exp": 1234567890,                                 // Expiration: Time issued + 300 seconds (5 min max)
  "jti": "unique-id-12345"                           // JWT ID: Unique for replay prevention
}

7. Available Scopes

You can request specific scopes in your token exchange request. Your partner account will be configured with allowed scopes.

account-management    - Access to management APIs (org, workspace, user creation)
thread                - Access to threads API
analytics             - Access to analytics API

What You Need to Implement

1. Account Setup Flow

The Threads API requires an access token for a user with the thread scope. Before you can generate these tokens, you must set up your accounts using the Account Management API.

Account Hierarchy

The Account Management API uses a three-tier hierarchy:

Organization: Your top-level entity
Workspace: Groups of users within an organization
User: Individual accounts that can access the Threads API

Choosing Your Structure

For most use cases, we recommend starting with a simple structure:

1 Organization
1 Workspace
All users in that single workspace

This approach works well when you don’t need to segment users into different groups. If you have more complex requirements (e.g., separating internal production use from end user production use, or isolating different end user groups), contact us for guidance on structuring your hierarchy.

Step 1: Get Service Account Token

Perform token exchange with your client ID as the subject to obtain a management token:

import time
import uuid
import jwt
import requests
from cryptography.hazmat.primitives import serialization
KID: str = "your-key-id-1"
CLIENT_ID: str = "your_client_id_abc123"
CLIENT_SECRET: str = "your_client_secret_xyz789"
PRIVATE_KEY: str = "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"
BASE_URL: str = "https://sandbox.api.boosted.ai"
key_bytes: bytes = PRIVATE_KEY.encode("utf-8")
private_key = serialization.load_pem_private_key(key_bytes, password=None)
payload: dict = {
    "iss": CLIENT_ID,
    "sub": CLIENT_ID,  # Use client_id for service account
    "aud": "https://sandbox.api.boosted.ai",
    "exp": int(time.time()) + 300,
    "jti": str(uuid.uuid4()),
}
subject_token: str = jwt.encode(payload, private_key, algorithm="RS256", headers={"kid": KID})
token_resp: requests.Response = requests.post(
    f"{BASE_URL}/oauth2/token",
    data={
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
        "subject_token": subject_token,
        "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
        "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
        "scope": "account-management",
    },
)
service_token: str = token_resp.json().get("access_token")

The bash snippets use process substitution <(...) which requires bash. Add #!/bin/bash at the top of your script or run with bash script.sh.

Step 2: Create Organization

org_payload: dict = {"name": "Test Org"}
org_resp: dict = make_request("PUT", "/v2/account-management/org", service_token, org_payload)
org_id: str = org_resp["org"]["org_id"]

Step 3: Create Workspace

workspace_payload: dict = {"org_id": org_id, "name": "Test Workspace"}
ws_resp: dict = make_request("PUT", "/v2/account-management/workspace", service_token, workspace_payload)
workspace_id: str = ws_resp["workspace"]["workspace_id"]

Step 4: Create User

user_payload: dict = {"workspace_id": workspace_id, "name": "Test User"}
user_resp: dict = make_request("PUT", "/v2/account-management/user", service_token, user_payload)
user_id: str = user_resp["user"]["user_id"]
# Store this user_id in your database
# It will be used in the 'sub' claim when performing token exchange

2. User Token Exchange

When a user needs Boosted access, create a JWT with their User ID as the subject and exchange it for an access token:

payload: dict = {
    "iss": CLIENT_ID,
    "sub": user_id,  # User ID from create user response
    "aud": "https://sandbox.api.boosted.ai",
    "exp": int(time.time()) + 300,
    "jti": str(uuid.uuid4()),
}
subject_token: str = jwt.encode(payload, private_key, algorithm="RS256", headers={"kid": KID})
token_resp: requests.Response = requests.post(
    f"{BASE_URL}/oauth2/token",
    data={
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
        "subject_token": subject_token,
        "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
        "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
        "scope": "thread",
    },
)
user_token: str = token_resp.json().get("access_token")

Example Response:

{
  "access_token": "aeea600e-961d-4d2f-bd50-c2eaf6609552",
  "token_type": "Bearer",
  "expires_in": 3600,
  "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
  "scope": "thread"
}

3. Using the Threads API

Use the access token to call Boosted APIs:

# Create a thread
threads_payload: dict = {"thread_name": "Test thread"}
threads_resp: dict = make_request("POST", "/v2/threads", user_token, threads_payload)
thread_id: str = threads_resp["thread_id"]
# List threads
threads_list: dict = make_request("GET", "/v2/threads", user_token)
# Send a message
message_resp: dict = make_request(
    "POST",
    f"/v2/threads/{thread_id}/messages",
    user_token,
    {"message": "Hello from Python client"},
    use_form_urlencoded=True,
)

4. Complete Integration Example

import time
import uuid
import jwt
import requests
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.types import PrivateKeyTypes
# =============================================================================
# CONFIGURATION - Set these values before running
# =============================================================================
KID: str = "SET - KID HERE"
CLIENT_ID: str = "SET - CLIENT ID HERE"
CLIENT_SECRET: str = "SET - CLIENT SECRET HERE"
PRIVATE_KEY: str = "SET - PRIVATE KEY HERE"
BASE_URL: str = "https://sandbox.api.boosted.ai"
# =============================================================================
# API UTILITIES
# =============================================================================
def make_request(
    method: str,
    endpoint: str,
    token: str,
    data: dict | None = None,
    use_form_urlencoded: bool = False,
) -> dict:
    """
    Make an authenticated HTTP request to the Boosted API.
    Args:
        method: HTTP method (GET, POST, PUT, etc.)
        endpoint: API endpoint path
        token: Bearer token for authorization
        data: Request payload
        use_form_urlencoded: If True, send as form data instead of JSON
    Returns:
        Parsed JSON response as a dictionary
    Raises:
        Exception: If the API request fails
    """
    url: str = f"{BASE_URL}{endpoint}"
    headers: dict = {
        "Content-Type": "application/json"
        if not use_form_urlencoded
        else "application/x-www-form-urlencoded",
        "Authorization": f"Bearer {token}",
    }
    print(f"\n[{method}] {endpoint}")
    response: requests.Response
    if use_form_urlencoded:
        response = requests.request(method, url, data=data, headers=headers)
    else:
        response = requests.request(method, url, json=data, headers=headers)
    if response.status_code in [201, 200]:
        print(f"Success ({response.status_code})")
        return response.json()
    else:
        print(f"Failed ({response.status_code})")
        print(response.text)
        raise Exception(f"API Request failed: [{method}] {endpoint}")
def get_auth_token(user: str, scope: str, pkey: PrivateKeyTypes) -> str:
    """
    Exchange a JWT for an access token via OAuth2 token exchange.
    Args:
        user: User ID or client ID to generate token for
        scope: OAuth scope (e.g., "account-management", "thread")
        pkey: RSA private key for signing the JWT
    Returns:
        Access token string
    """
    payload: dict = {
        "iss": CLIENT_ID,
        "sub": user,
        "aud": "https://sandbox.api.boosted.ai",
        "exp": int(time.time()) + 300,
        "jti": str(uuid.uuid4()),
    }
    subject_token: str = jwt.encode(payload, pkey, algorithm="RS256", headers={"kid": KID})
    print("\nExchanging token...")
    token_resp: requests.Response = requests.post(
        f"{BASE_URL}/oauth2/token",
        data={
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
            "subject_token": subject_token,
            "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
            "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
            "scope": scope,
        },
    )
    if token_resp.status_code != 200:
        print(f"Token Exchange Failed: {token_resp.text}")
        exit(1)
    token: str = token_resp.json().get("access_token")
    print(f"Auth Token Acquired: {token}")
    return token
# =============================================================================
# ACCOUNT MANAGEMENT
# =============================================================================
def create_user(token: str) -> str:
    """
    Create a new organization, workspace, and user.
    Args:
        token: Service account token with account-management scope
    Returns:
        The created user's ID
    """
    org_payload: dict = {"name": "Test Org"}
    org_resp: dict = make_request("PUT", "/v2/account-management/org", token, org_payload)
    print(org_resp)
    org_id: str = org_resp["org"]["org_id"]
    print(f"Created Org ID: {org_id}")
    workspace_payload: dict = {"org_id": org_id, "name": "Test Workspace"}
    ws_resp: dict = make_request(
        "PUT", "/v2/account-management/workspace", token, workspace_payload
    )
    print(ws_resp)
    workspace_id: str = ws_resp["workspace"]["workspace_id"]
    print(f"Created Workspace ID: {workspace_id}")
    user_payload: dict = {"workspace_id": workspace_id, "name": "Test User"}
    user_resp: dict = make_request("PUT", "/v2/account-management/user", token, user_payload)
    print(user_resp)
    user_id: str = user_resp["user"]["user_id"]
    print(f"Created User ID: {user_id}")
    return user_id
# =============================================================================
# THREAD OPERATIONS
# =============================================================================
def test_thread_operations(token: str) -> None:
    """
    Test thread API operations: create, list, and send message.
    Args:
        token: User token with thread scope
    """
    threads_payload: dict = {"thread_name": "Test thread"}
    threads_resp: dict = make_request("POST", "/v2/threads", token, threads_payload)
    print(threads_resp)
    threads_list: dict = make_request("GET", "/v2/threads", token)
    print(threads_list)
    thread_id: str = threads_resp["thread_id"]
    message_resp: dict = make_request(
        "POST",
        f"/v2/threads/{thread_id}/messages",
        token,
        {"message": "string"},
        True,
    )
    print(message_resp)
# =============================================================================
# MAIN
# =============================================================================
if __name__ == "__main__":
    key_bytes: bytes = PRIVATE_KEY.encode("utf-8")
    private_key: PrivateKeyTypes = serialization.load_pem_private_key(key_bytes, password=None)
    print("Making Service Token")
    service_token: str = get_auth_token(CLIENT_ID, "account-management", private_key)
    user_id: str = create_user(service_token)
    print("Making User Token")
    user_token: str = get_auth_token(user_id, "thread", private_key)
    test_thread_operations(user_token)

5. Token Expiration Handling

Access tokens expire after 1 hour. When expired, perform another token exchange:

def get_boosted_token_with_cache(user_id: str, scope: str, private_key) -> str:
    """Get token with expiry check
    Args:
        user_id: The User ID from create user API response
        scope: OAuth scope to request
        private_key: RSA private key for signing
    """
    cache_key = f"boosted_token:{user_id}:{scope}"
    # Check cache
    cached = cache.get(cache_key)
    if cached and cached['expires_at'] > time.time():
        return cached['token']
    # Token expired or not cached - perform exchange
    access_token = get_auth_token(user_id, scope, private_key)
    # Cache with 55 min TTL (5 min buffer before actual expiry)
    cache.set(cache_key, {
        'token': access_token,
        'expires_at': time.time() + 3300  # 55 minutes
    })
    return access_token

Token Exchange does not return refresh tokens. When the access token expires, create a new JWT and perform another token exchange.

Table of Contents