Authorization

Authenticate with the Alfa API using OAuth Token Exchange.

Authorization

Authenticate with the Alfa API using OAuth Token Exchange.

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):

1 {
2   "keys": [
3     {
4       "kty": "RSA",
5       "alg": "RS256",
6       "use": "sig",
7       "kid": "your-key-id-1",
8       "n": "oO2Re_MzrCqR-1rLFcmuZqf2kYmqjWQax1gYo-cWIGEDX_UIvQ4FtDRx53fPfbl...",
9       "e": "AQAB"
10     }
11   ]
12 }

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:

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

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:

1 import time
2 import uuid
3 
4 import jwt
5 import requests
6 from cryptography.hazmat.primitives import serialization
7 
8 KID: str = "your-key-id-1"
9 CLIENT_ID: str = "your_client_id_abc123"
10 CLIENT_SECRET: str = "your_client_secret_xyz789"
11 PRIVATE_KEY: str = "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"
12 
13 BASE_URL: str = "https://sandbox.api.boosted.ai"
14 
15 key_bytes: bytes = PRIVATE_KEY.encode("utf-8")
16 private_key = serialization.load_pem_private_key(key_bytes, password=None)
17 
18 payload: dict = {
19     "iss": CLIENT_ID,
20     "sub": CLIENT_ID,  # Use client_id for service account
21     "aud": "https://sandbox.api.boosted.ai",
22     "exp": int(time.time()) + 300,
23     "jti": str(uuid.uuid4()),
24 }
25 
26 subject_token: str = jwt.encode(payload, private_key, algorithm="RS256", headers={"kid": KID})
27 
28 token_resp: requests.Response = requests.post(
29     f"{BASE_URL}/oauth2/token",
30     data={
31         "client_id": CLIENT_ID,
32         "client_secret": CLIENT_SECRET,
33         "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
34         "subject_token": subject_token,
35         "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
36         "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
37         "scope": "account-management",
38     },
39 )
40 
41 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

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

Step 3: Create Workspace

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

Step 4: Create User

1 user_payload: dict = {"workspace_id": workspace_id, "name": "Test User"}
2 user_resp: dict = make_request("PUT", "/v2/account-management/user", service_token, user_payload)
3 user_id: str = user_resp["user"]["user_id"]
4 
5 # Store this user_id in your database
6 # 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:

1 payload: dict = {
2     "iss": CLIENT_ID,
3     "sub": user_id,  # User ID from create user response
4     "aud": "https://sandbox.api.boosted.ai",
5     "exp": int(time.time()) + 300,
6     "jti": str(uuid.uuid4()),
7 }
8 
9 subject_token: str = jwt.encode(payload, private_key, algorithm="RS256", headers={"kid": KID})
10 
11 token_resp: requests.Response = requests.post(
12     f"{BASE_URL}/oauth2/token",
13     data={
14         "client_id": CLIENT_ID,
15         "client_secret": CLIENT_SECRET,
16         "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
17         "subject_token": subject_token,
18         "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
19         "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
20         "scope": "thread",
21     },
22 )
23 
24 user_token: str = token_resp.json().get("access_token")

Example Response:

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

3. Using the Threads API

Use the access token to call Boosted APIs:

1 # Create a thread
2 threads_payload: dict = {"thread_name": "Test thread"}
3 threads_resp: dict = make_request("POST", "/v2/threads", user_token, threads_payload)
4 thread_id: str = threads_resp["thread_id"]
5 
6 # List threads
7 threads_list: dict = make_request("GET", "/v2/threads", user_token)
8 
9 # Send a message
10 message_resp: dict = make_request(
11     "POST",
12     f"/v2/threads/{thread_id}/messages",
13     user_token,
14     {"message": "Hello from Python client"},
15     use_form_urlencoded=True,
16 )

4. Complete Integration Example

1 import time
2 import uuid
3 
4 import jwt
5 import requests
6 from cryptography.hazmat.primitives import serialization
7 from cryptography.hazmat.primitives.asymmetric.types import PrivateKeyTypes
8 
9 # =============================================================================
10 # CONFIGURATION - Set these values before running
11 # =============================================================================
12 
13 KID: str = "SET - KID HERE"
14 CLIENT_ID: str = "SET - CLIENT ID HERE"
15 CLIENT_SECRET: str = "SET - CLIENT SECRET HERE"
16 PRIVATE_KEY: str = "SET - PRIVATE KEY HERE"
17 
18 BASE_URL: str = "https://sandbox.api.boosted.ai"
19 
20 
21 # =============================================================================
22 # API UTILITIES
23 # =============================================================================
24 
25 
26 def make_request(
27     method: str,
28     endpoint: str,
29     token: str,
30     data: dict | None = None,
31     use_form_urlencoded: bool = False,
32 ) -> dict:
33     """
34     Make an authenticated HTTP request to the Boosted API.
35 
36     Args:
37         method: HTTP method (GET, POST, PUT, etc.)
38         endpoint: API endpoint path
39         token: Bearer token for authorization
40         data: Request payload
41         use_form_urlencoded: If True, send as form data instead of JSON
42 
43     Returns:
44         Parsed JSON response as a dictionary
45 
46     Raises:
47         Exception: If the API request fails
48     """
49     url: str = f"{BASE_URL}{endpoint}"
50     headers: dict = {
51         "Content-Type": "application/json"
52         if not use_form_urlencoded
53         else "application/x-www-form-urlencoded",
54         "Authorization": f"Bearer {token}",
55     }
56 
57     print(f"\n[{method}] {endpoint}")
58     response: requests.Response
59     if use_form_urlencoded:
60         response = requests.request(method, url, data=data, headers=headers)
61     else:
62         response = requests.request(method, url, json=data, headers=headers)
63 
64     if response.status_code in [201, 200]:
65         print(f"Success ({response.status_code})")
66         return response.json()
67     else:
68         print(f"Failed ({response.status_code})")
69         print(response.text)
70         raise Exception(f"API Request failed: [{method}] {endpoint}")
71 
72 
73 def get_auth_token(user: str, scope: str, pkey: PrivateKeyTypes) -> str:
74     """
75     Exchange a JWT for an access token via OAuth2 token exchange.
76 
77     Args:
78         user: User ID or client ID to generate token for
79         scope: OAuth scope (e.g., "account-management", "thread")
80         pkey: RSA private key for signing the JWT
81 
82     Returns:
83         Access token string
84     """
85     payload: dict = {
86         "iss": CLIENT_ID,
87         "sub": user,
88         "aud": "https://sandbox.api.boosted.ai",
89         "exp": int(time.time()) + 300,
90         "jti": str(uuid.uuid4()),
91     }
92 
93     subject_token: str = jwt.encode(payload, pkey, algorithm="RS256", headers={"kid": KID})
94 
95     print("\nExchanging token...")
96     token_resp: requests.Response = requests.post(
97         f"{BASE_URL}/oauth2/token",
98         data={
99             "client_id": CLIENT_ID,
100             "client_secret": CLIENT_SECRET,
101             "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
102             "subject_token": subject_token,
103             "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
104             "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
105             "scope": scope,
106         },
107     )
108 
109     if token_resp.status_code != 200:
110         print(f"Token Exchange Failed: {token_resp.text}")
111         exit(1)
112 
113     token: str = token_resp.json().get("access_token")
114     print(f"Auth Token Acquired: {token}")
115     return token
116 
117 
118 # =============================================================================
119 # ACCOUNT MANAGEMENT
120 # =============================================================================
121 
122 
123 def create_user(token: str) -> str:
124     """
125     Create a new organization, workspace, and user.
126 
127     Args:
128         token: Service account token with account-management scope
129 
130     Returns:
131         The created user's ID
132     """
133     org_payload: dict = {"name": "Test Org"}
134     org_resp: dict = make_request("PUT", "/v2/account-management/org", token, org_payload)
135     print(org_resp)
136     org_id: str = org_resp["org"]["org_id"]
137     print(f"Created Org ID: {org_id}")
138 
139     workspace_payload: dict = {"org_id": org_id, "name": "Test Workspace"}
140     ws_resp: dict = make_request(
141         "PUT", "/v2/account-management/workspace", token, workspace_payload
142     )
143     print(ws_resp)
144     workspace_id: str = ws_resp["workspace"]["workspace_id"]
145     print(f"Created Workspace ID: {workspace_id}")
146 
147     user_payload: dict = {"workspace_id": workspace_id, "name": "Test User"}
148     user_resp: dict = make_request("PUT", "/v2/account-management/user", token, user_payload)
149     print(user_resp)
150     user_id: str = user_resp["user"]["user_id"]
151     print(f"Created User ID: {user_id}")
152 
153     return user_id
154 
155 
156 # =============================================================================
157 # THREAD OPERATIONS
158 # =============================================================================
159 
160 
161 def test_thread_operations(token: str) -> None:
162     """
163     Test thread API operations: create, list, and send message.
164 
165     Args:
166         token: User token with thread scope
167     """
168     threads_payload: dict = {"thread_name": "Test thread"}
169     threads_resp: dict = make_request("POST", "/v2/threads", token, threads_payload)
170     print(threads_resp)
171 
172     threads_list: dict = make_request("GET", "/v2/threads", token)
173     print(threads_list)
174 
175     thread_id: str = threads_resp["thread_id"]
176     message_resp: dict = make_request(
177         "POST",
178         f"/v2/threads/{thread_id}/messages",
179         token,
180         {"message": "string"},
181         True,
182     )
183     print(message_resp)
184 
185 
186 # =============================================================================
187 # MAIN
188 # =============================================================================
189 
190 if __name__ == "__main__":
191     key_bytes: bytes = PRIVATE_KEY.encode("utf-8")
192     private_key: PrivateKeyTypes = serialization.load_pem_private_key(key_bytes, password=None)
193 
194     print("Making Service Token")
195     service_token: str = get_auth_token(CLIENT_ID, "account-management", private_key)
196 
197     user_id: str = create_user(service_token)
198 
199     print("Making User Token")
200     user_token: str = get_auth_token(user_id, "thread", private_key)
201 
202     test_thread_operations(user_token)

5. Token Expiration Handling

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

1 def get_boosted_token_with_cache(user_id: str, scope: str, private_key) -> str:
2     """Get token with expiry check
3 
4     Args:
5         user_id: The User ID from create user API response
6         scope: OAuth scope to request
7         private_key: RSA private key for signing
8     """
9 
10     cache_key = f"boosted_token:{user_id}:{scope}"
11 
12     # Check cache
13     cached = cache.get(cache_key)
14     if cached and cached['expires_at'] > time.time():
15         return cached['token']
16 
17     # Token expired or not cached - perform exchange
18     access_token = get_auth_token(user_id, scope, private_key)
19 
20     # Cache with 55 min TTL (5 min buffer before actual expiry)
21     cache.set(cache_key, {
22         'token': access_token,
23         'expires_at': time.time() + 3300  # 55 minutes
24     })
25 
26     return access_token

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

Overview

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):

1 {
2   "keys": [
3     {
4       "kty": "RSA",
5       "alg": "RS256",
6       "use": "sig",
7       "kid": "your-key-id-1",
8       "n": "oO2Re_MzrCqR-1rLFcmuZqf2kYmqjWQax1gYo-cWIGEDX_UIvQ4FtDRx53fPfbl...",
9       "e": "AQAB"
10     }
11   ]
12 }

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:

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

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

Step 1: Get Service Account Token

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

1 import time
2 import uuid
3 
4 import jwt
5 import requests
6 from cryptography.hazmat.primitives import serialization
7 
8 KID: str = "your-key-id-1"
9 CLIENT_ID: str = "your_client_id_abc123"
10 CLIENT_SECRET: str = "your_client_secret_xyz789"
11 PRIVATE_KEY: str = "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"
12 
13 BASE_URL: str = "https://sandbox.api.boosted.ai"
14 
15 key_bytes: bytes = PRIVATE_KEY.encode("utf-8")
16 private_key = serialization.load_pem_private_key(key_bytes, password=None)
17 
18 payload: dict = {
19     "iss": CLIENT_ID,
20     "sub": CLIENT_ID,  # Use client_id for service account
21     "aud": "https://sandbox.api.boosted.ai",
22     "exp": int(time.time()) + 300,
23     "jti": str(uuid.uuid4()),
24 }
25 
26 subject_token: str = jwt.encode(payload, private_key, algorithm="RS256", headers={"kid": KID})
27 
28 token_resp: requests.Response = requests.post(
29     f"{BASE_URL}/oauth2/token",
30     data={
31         "client_id": CLIENT_ID,
32         "client_secret": CLIENT_SECRET,
33         "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
34         "subject_token": subject_token,
35         "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
36         "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
37         "scope": "account-management",
38     },
39 )
40 
41 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

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

Step 3: Create Workspace

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

Step 4: Create User

1 user_payload: dict = {"workspace_id": workspace_id, "name": "Test User"}
2 user_resp: dict = make_request("PUT", "/v2/account-management/user", service_token, user_payload)
3 user_id: str = user_resp["user"]["user_id"]
4 
5 # Store this user_id in your database
6 # 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:

1 payload: dict = {
2     "iss": CLIENT_ID,
3     "sub": user_id,  # User ID from create user response
4     "aud": "https://sandbox.api.boosted.ai",
5     "exp": int(time.time()) + 300,
6     "jti": str(uuid.uuid4()),
7 }
8 
9 subject_token: str = jwt.encode(payload, private_key, algorithm="RS256", headers={"kid": KID})
10 
11 token_resp: requests.Response = requests.post(
12     f"{BASE_URL}/oauth2/token",
13     data={
14         "client_id": CLIENT_ID,
15         "client_secret": CLIENT_SECRET,
16         "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
17         "subject_token": subject_token,
18         "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
19         "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
20         "scope": "thread",
21     },
22 )
23 
24 user_token: str = token_resp.json().get("access_token")

Example Response:

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

3. Using the Threads API

Use the access token to call Boosted APIs:

1 # Create a thread
2 threads_payload: dict = {"thread_name": "Test thread"}
3 threads_resp: dict = make_request("POST", "/v2/threads", user_token, threads_payload)
4 thread_id: str = threads_resp["thread_id"]
5 
6 # List threads
7 threads_list: dict = make_request("GET", "/v2/threads", user_token)
8 
9 # Send a message
10 message_resp: dict = make_request(
11     "POST",
12     f"/v2/threads/{thread_id}/messages",
13     user_token,
14     {"message": "Hello from Python client"},
15     use_form_urlencoded=True,
16 )

4. Complete Integration Example

1 import time
2 import uuid
3 
4 import jwt
5 import requests
6 from cryptography.hazmat.primitives import serialization
7 from cryptography.hazmat.primitives.asymmetric.types import PrivateKeyTypes
8 
9 # =============================================================================
10 # CONFIGURATION - Set these values before running
11 # =============================================================================
12 
13 KID: str = "SET - KID HERE"
14 CLIENT_ID: str = "SET - CLIENT ID HERE"
15 CLIENT_SECRET: str = "SET - CLIENT SECRET HERE"
16 PRIVATE_KEY: str = "SET - PRIVATE KEY HERE"
17 
18 BASE_URL: str = "https://sandbox.api.boosted.ai"
19 
20 
21 # =============================================================================
22 # API UTILITIES
23 # =============================================================================
24 
25 
26 def make_request(
27     method: str,
28     endpoint: str,
29     token: str,
30     data: dict | None = None,
31     use_form_urlencoded: bool = False,
32 ) -> dict:
33     """
34     Make an authenticated HTTP request to the Boosted API.
35 
36     Args:
37         method: HTTP method (GET, POST, PUT, etc.)
38         endpoint: API endpoint path
39         token: Bearer token for authorization
40         data: Request payload
41         use_form_urlencoded: If True, send as form data instead of JSON
42 
43     Returns:
44         Parsed JSON response as a dictionary
45 
46     Raises:
47         Exception: If the API request fails
48     """
49     url: str = f"{BASE_URL}{endpoint}"
50     headers: dict = {
51         "Content-Type": "application/json"
52         if not use_form_urlencoded
53         else "application/x-www-form-urlencoded",
54         "Authorization": f"Bearer {token}",
55     }
56 
57     print(f"\n[{method}] {endpoint}")
58     response: requests.Response
59     if use_form_urlencoded:
60         response = requests.request(method, url, data=data, headers=headers)
61     else:
62         response = requests.request(method, url, json=data, headers=headers)
63 
64     if response.status_code in [201, 200]:
65         print(f"Success ({response.status_code})")
66         return response.json()
67     else:
68         print(f"Failed ({response.status_code})")
69         print(response.text)
70         raise Exception(f"API Request failed: [{method}] {endpoint}")
71 
72 
73 def get_auth_token(user: str, scope: str, pkey: PrivateKeyTypes) -> str:
74     """
75     Exchange a JWT for an access token via OAuth2 token exchange.
76 
77     Args:
78         user: User ID or client ID to generate token for
79         scope: OAuth scope (e.g., "account-management", "thread")
80         pkey: RSA private key for signing the JWT
81 
82     Returns:
83         Access token string
84     """
85     payload: dict = {
86         "iss": CLIENT_ID,
87         "sub": user,
88         "aud": "https://sandbox.api.boosted.ai",
89         "exp": int(time.time()) + 300,
90         "jti": str(uuid.uuid4()),
91     }
92 
93     subject_token: str = jwt.encode(payload, pkey, algorithm="RS256", headers={"kid": KID})
94 
95     print("\nExchanging token...")
96     token_resp: requests.Response = requests.post(
97         f"{BASE_URL}/oauth2/token",
98         data={
99             "client_id": CLIENT_ID,
100             "client_secret": CLIENT_SECRET,
101             "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
102             "subject_token": subject_token,
103             "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
104             "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
105             "scope": scope,
106         },
107     )
108 
109     if token_resp.status_code != 200:
110         print(f"Token Exchange Failed: {token_resp.text}")
111         exit(1)
112 
113     token: str = token_resp.json().get("access_token")
114     print(f"Auth Token Acquired: {token}")
115     return token
116 
117 
118 # =============================================================================
119 # ACCOUNT MANAGEMENT
120 # =============================================================================
121 
122 
123 def create_user(token: str) -> str:
124     """
125     Create a new organization, workspace, and user.
126 
127     Args:
128         token: Service account token with account-management scope
129 
130     Returns:
131         The created user's ID
132     """
133     org_payload: dict = {"name": "Test Org"}
134     org_resp: dict = make_request("PUT", "/v2/account-management/org", token, org_payload)
135     print(org_resp)
136     org_id: str = org_resp["org"]["org_id"]
137     print(f"Created Org ID: {org_id}")
138 
139     workspace_payload: dict = {"org_id": org_id, "name": "Test Workspace"}
140     ws_resp: dict = make_request(
141         "PUT", "/v2/account-management/workspace", token, workspace_payload
142     )
143     print(ws_resp)
144     workspace_id: str = ws_resp["workspace"]["workspace_id"]
145     print(f"Created Workspace ID: {workspace_id}")
146 
147     user_payload: dict = {"workspace_id": workspace_id, "name": "Test User"}
148     user_resp: dict = make_request("PUT", "/v2/account-management/user", token, user_payload)
149     print(user_resp)
150     user_id: str = user_resp["user"]["user_id"]
151     print(f"Created User ID: {user_id}")
152 
153     return user_id
154 
155 
156 # =============================================================================
157 # THREAD OPERATIONS
158 # =============================================================================
159 
160 
161 def test_thread_operations(token: str) -> None:
162     """
163     Test thread API operations: create, list, and send message.
164 
165     Args:
166         token: User token with thread scope
167     """
168     threads_payload: dict = {"thread_name": "Test thread"}
169     threads_resp: dict = make_request("POST", "/v2/threads", token, threads_payload)
170     print(threads_resp)
171 
172     threads_list: dict = make_request("GET", "/v2/threads", token)
173     print(threads_list)
174 
175     thread_id: str = threads_resp["thread_id"]
176     message_resp: dict = make_request(
177         "POST",
178         f"/v2/threads/{thread_id}/messages",
179         token,
180         {"message": "string"},
181         True,
182     )
183     print(message_resp)
184 
185 
186 # =============================================================================
187 # MAIN
188 # =============================================================================
189 
190 if __name__ == "__main__":
191     key_bytes: bytes = PRIVATE_KEY.encode("utf-8")
192     private_key: PrivateKeyTypes = serialization.load_pem_private_key(key_bytes, password=None)
193 
194     print("Making Service Token")
195     service_token: str = get_auth_token(CLIENT_ID, "account-management", private_key)
196 
197     user_id: str = create_user(service_token)
198 
199     print("Making User Token")
200     user_token: str = get_auth_token(user_id, "thread", private_key)
201 
202     test_thread_operations(user_token)

5. Token Expiration Handling

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

1 def get_boosted_token_with_cache(user_id: str, scope: str, private_key) -> str:
2     """Get token with expiry check
3 
4     Args:
5         user_id: The User ID from create user API response
6         scope: OAuth scope to request
7         private_key: RSA private key for signing
8     """
9 
10     cache_key = f"boosted_token:{user_id}:{scope}"
11 
12     # Check cache
13     cached = cache.get(cache_key)
14     if cached and cached['expires_at'] > time.time():
15         return cached['token']
16 
17     # Token expired or not cached - perform exchange
18     access_token = get_auth_token(user_id, scope, private_key)
19 
20     # Cache with 55 min TTL (5 min buffer before actual expiry)
21     cache.set(cache_key, {
22         'token': access_token,
23         'expires_at': time.time() + 3300  # 55 minutes
24     })
25 
26     return access_token

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

$	# 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

1	{
2	"keys": [
3	{
4	"kty": "RSA",
5	"alg": "RS256",
6	"use": "sig",
7	"kid": "your-key-id-1",
8	"n": "oO2Re_MzrCqR-1rLFcmuZqf2kYmqjWQax1gYo-cWIGEDX_UIvQ4FtDRx53fPfbl...",
9	"e": "AQAB"
10	}
11	]
12	}

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

1	import time
2	import uuid
3
4	import jwt
5	import requests
6	from cryptography.hazmat.primitives import serialization
7
8	KID: str = "your-key-id-1"
9	CLIENT_ID: str = "your_client_id_abc123"
10	CLIENT_SECRET: str = "your_client_secret_xyz789"
11	PRIVATE_KEY: str = "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"
12
13	BASE_URL: str = "https://sandbox.api.boosted.ai"
14
15	key_bytes: bytes = PRIVATE_KEY.encode("utf-8")
16	private_key = serialization.load_pem_private_key(key_bytes, password=None)
17
18	payload: dict = {
19	"iss": CLIENT_ID,
20	"sub": CLIENT_ID, # Use client_id for service account
21	"aud": "https://sandbox.api.boosted.ai",
22	"exp": int(time.time()) + 300,
23	"jti": str(uuid.uuid4()),
24	}
25
26	subject_token: str = jwt.encode(payload, private_key, algorithm="RS256", headers={"kid": KID})
27
28	token_resp: requests.Response = requests.post(
29	f"{BASE_URL}/oauth2/token",
30	data={
31	"client_id": CLIENT_ID,
32	"client_secret": CLIENT_SECRET,
33	"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
34	"subject_token": subject_token,
35	"subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
36	"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
37	"scope": "account-management",
38	},
39	)
40
41	service_token: str = token_resp.json().get("access_token")

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

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

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

1	payload: dict = {
2	"iss": CLIENT_ID,
3	"sub": user_id, # User ID from create user response
4	"aud": "https://sandbox.api.boosted.ai",
5	"exp": int(time.time()) + 300,
6	"jti": str(uuid.uuid4()),
7	}
8
9	subject_token: str = jwt.encode(payload, private_key, algorithm="RS256", headers={"kid": KID})
10
11	token_resp: requests.Response = requests.post(
12	f"{BASE_URL}/oauth2/token",
13	data={
14	"client_id": CLIENT_ID,
15	"client_secret": CLIENT_SECRET,
16	"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
17	"subject_token": subject_token,
18	"subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
19	"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
20	"scope": "thread",
21	},
22	)
23
24	user_token: str = token_resp.json().get("access_token")

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

1	# Create a thread
2	threads_payload: dict = {"thread_name": "Test thread"}
3	threads_resp: dict = make_request("POST", "/v2/threads", user_token, threads_payload)
4	thread_id: str = threads_resp["thread_id"]
5
6	# List threads
7	threads_list: dict = make_request("GET", "/v2/threads", user_token)
8
9	# Send a message
10	message_resp: dict = make_request(
11	"POST",
12	f"/v2/threads/{thread_id}/messages",
13	user_token,
14	{"message": "Hello from Python client"},
15	use_form_urlencoded=True,
16	)

1	def get_boosted_token_with_cache(user_id: str, scope: str, private_key) -> str:
2	"""Get token with expiry check
3
4	Args:
5	user_id: The User ID from create user API response
6	scope: OAuth scope to request
7	private_key: RSA private key for signing
8	"""
9
10	cache_key = f"boosted_token:{user_id}:{scope}"
11
12	# Check cache
13	cached = cache.get(cache_key)
14	if cached and cached['expires_at'] > time.time():
15	return cached['token']
16
17	# Token expired or not cached - perform exchange
18	access_token = get_auth_token(user_id, scope, private_key)
19
20	# Cache with 55 min TTL (5 min buffer before actual expiry)
21	cache.set(cache_key, {
22	'token': access_token,
23	'expires_at': time.time() + 3300 # 55 minutes
24	})
25
26	return access_token

Table of Contents

Overview

What You Need to Provide Us

1. Public Signing Key URL (jwks_uri)

What We Will Provide You

1. Client ID

2. Client Secret

3. Token Endpoint

4. Account Management API Endpoints

5. Audience Value

6. JWT Requirements

7. Available Scopes

What You Need to Implement

1. Account Setup Flow

Step 1: Get Service Account Token

Step 2: Create Organization

Step 3: Create Workspace

Step 4: Create User

2. User Token Exchange

3. Using the Threads API

4. Complete Integration Example

5. Token Expiration Handling

Table of Contents

Overview

What You Need to Provide Us

1. Public Signing Key URL (jwks_uri)

What We Will Provide You

1. Client ID

2. Client Secret

3. Token Endpoint

4. Account Management API Endpoints

5. Audience Value

6. JWT Requirements

7. Available Scopes

What You Need to Implement

1. Account Setup Flow

Step 1: Get Service Account Token

Step 2: Create Organization

Step 3: Create Workspace

Step 4: Create User

2. User Token Exchange

3. Using the Threads API

4. Complete Integration Example

5. Token Expiration Handling

1. Public Signing Key URL (`jwks_uri`)

1. Public Signing Key URL (`jwks_uri`)