Authentication
# Set your API key
export ATAI_API_KEY="your-api-key-here"
# Basic API call
curl https://api.u1.archetypeai.app/v0.5/lens/info \
-H "Authorization: Bearer $ATAI_API_KEY"
Common Patterns
Create and Use a Session
import requests
# 1. Create session
response = requests.post(
"https://api.u1.archetypeai.app/v0.5/lens/sessions/create",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={"lens_id": "lns-fd669361822b07e2-237ab3ffd79199b0"}
)
session = response.json()
# 2. Use WebSocket endpoint for data streaming
# session['session_endpoint'] contains WebSocket URL
# 3. Destroy session when done
requests.post(
"https://api.u1.archetypeai.app/v0.5/lens/sessions/destroy",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={"session_id": session['session_id']}
)
# 1. Create session
SESSION_ID=$(curl -s https://api.u1.archetypeai.app/v0.5/lens/sessions/create \
-H "Authorization: Bearer $ATAI_API_KEY" \
-H "Content-Type: application/json" \
-d '{"lens_id": "lns-fd669361822b07e2-237ab3ffd79199b0"}' \
| jq -r '.session_id')
# 2. Destroy session
curl https://api.u1.archetypeai.app/v0.5/lens/sessions/destroy \
-H "Authorization: Bearer $ATAI_API_KEY" \
-H "Content-Type: application/json" \
-d "{\"session_id\": \"$SESSION_ID\"}"
Upload and Process Files
import requests
# Upload file
with open('image.png', 'rb') as f:
response = requests.post(
'https://api.u1.archetypeai.app/v0.5/files',
headers={'Authorization': 'Bearer YOUR_API_KEY'},
files={'file': f}
)
file_data = response.json()
print(f"Uploaded: {file_data['file_id']}")
# List files
response = requests.get(
"https://api.u1.archetypeai.app/v0.5/files/metadata",
headers={"Authorization": "Bearer YOUR_API_KEY"}
)
files = response.json()
for file in files:
print(f"- {file['file_id']} ({file['num_bytes']} bytes)")
# Upload file
curl -X POST https://api.u1.archetypeai.app/v0.5/files \
-H "Authorization: Bearer $ATAI_API_KEY" \
-F "[email protected]"
# List files
curl https://api.u1.archetypeai.app/v0.5/files/metadata \
-H "Authorization: Bearer $ATAI_API_KEY"
API Endpoints Summary
Lens API
| Method | Endpoint | Description |
|---|---|---|
POST | /lens/sessions/create | Create new lens session |
POST | /lens/sessions/destroy | Destroy lens session |
GET | /lens/sessions/info | Get sessions summary |
GET | /lens/sessions/metadata | Get active sessions metadata |
POST | /lens/register | Register lens |
POST | /lens/clone | Clone lens |
POST | /lens/modify | Modify lens |
POST | /lens/delete | Delete lens |
GET | /lens/info | Get lens summary info |
GET | /lens/metadata | Get detailed lens metadata |
Files API
| Method | Endpoint | Description |
|---|---|---|
POST | /files | Upload file |
POST | /files/base64 | Upload base64 file |
DELETE | /files/delete/{file_id} | Delete file |
GET | /files/download/{file_id} | Download file |
GET | /files/info | Get files info |
GET | /files/metadata | List all files and their metadata |
GET | /files/metadata/{file_id} | Get a file’s metadata |
POST | /files/uploads/initiate | Initiate a direct-to-cloud upload |
POST | /files/uploads/{upload_id}/parts/urls | Generate presigned URLs for upload parts |
POST | /files/uploads/{upload_id}/parts/checkpoint | Checkpoint completed upload parts |
POST | /files/uploads/{upload_id}/complete | Complete a direct-to-cloud upload |
POST | /files/uploads/{upload_id}/abort | Abort a direct-to-cloud upload |
Batch Processing API
| Method | Endpoint | Description |
|---|---|---|
| Jobs | ||
POST | /batch/jobs | Create job |
GET | /batch/jobs | List jobs |
GET | /batch/jobs/{id} | Get job info |
POST | /batch/jobs/{id}/retry | Retry job |
DELETE | /batch/jobs/{id} | Delete job |
POST | /batch/jobs/{id}/cancel | Cancel job |
GET | /batch/jobs/queue | Get global queue depths grouped by pipeline type |
| Events | ||
GET | /batch/jobs/{id}/events | List job events |
| I/O | ||
GET | /batch/jobs/{id}/inputs | List job inputs |
GET | /batch/jobs/{id}/outputs | List job outputs |
| Progress | ||
GET | /batch/jobs/{id}/progress | List job progress |
| Registry: Pipelines | ||
GET | /batch/registry/pipelines | List pipelines |
GET | /batch/registry/pipelines/{id} | Get pipeline info |
GET | /batch/registry/pipelines/{id}/schema | Get pipeline schema |
Fine-Tuning Service
| Method | Endpoint | Description |
|---|---|---|
POST | /fine-tuning/create-job | Create a fine-tuning job |
GET | /fine-tuning/get-job | Get information about a fine-tuning job |
GET | /fine-tuning/list-jobs | List fine-tuning jobs |
POST | /fine-tuning/pause-job | Pause a fine-tuning job |
POST | /fine-tuning/resume-job | Resume a fine-tuning job |
POST | /fine-tuning/cancel-job | Cancel a fine-tuning job |
GET | /fine-tuning/list-job-checkpoints | List the checkpoints saved for a given fine-tuning job |
GET | /fine-tuning/list-job-events | List the job events for a given fine-tuning job |
GET | /fine-tuning/list-job-logs | List the events for a given fine-tuning job |
GET | /fine-tuning/list-job-metrics | List a fine-tuning job’s metrics |
GET | /fine-tuning/list-checkpoint-options | List the checkpoints saved for jobs that match the specified filters |
Error Handling
import requests
import time
import random
def api_call_with_retry(func, *args, **kwargs):
max_retries = 3
for attempt in range(max_retries):
try:
response = func(*args, **kwargs)
if response.status_code == 429:
# Rate limited
retry_after = response.json().get('error', {}).get('details', {}).get('retry_after', 60)
time.sleep(retry_after)
continue
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if e.response.status_code < 500:
# Client error - don't retry
raise e
# Server error - retry with backoff
wait_time = (2 ** attempt) + random.uniform(0, 1)
time.sleep(wait_time)
raise Exception("Max retries exceeded")
Common Error Codes
| Code | HTTP Status | Description | Solution |
|---|---|---|---|
UNAUTHORIZED | 401 | Invalid API key | Check API key format: Bearer YOUR_KEY |
INVALID_LENS_ID | 400 | Lens doesn’t exist | Get valid IDs from /lens/metadata |
SESSION_NOT_FOUND | 404 | Session destroyed/expired | Create new session |
RATE_LIMIT_EXCEEDED | 429 | Too many requests | Implement exponential backoff |
FILE_TOO_LARGE | 413 | File > 1GB | Compress or split file |
WebSocket Usage
import websocket
import json
import base64
def on_message(ws, message):
response = json.loads(message)
print(f"Newton: {response}")
def on_open(ws):
# Send image data
with open('image.jpg', 'rb') as f:
image_data = base64.b64encode(f.read()).decode()
ws.send(json.dumps({
"type": "image",
"data": image_data,
"focus": "safety equipment" # Optional
}))
# Connect to session WebSocket
ws = websocket.WebSocketApp(
session_endpoint, # From session creation response
on_message=on_message,
on_open=on_open
)
ws.run_forever()
Rate Limits
- Default: 100 requests per minute
- Burst: Up to 20 requests per second
- Sessions: 10 concurrent sessions per organization
- Files: 1GB max size, 100 files per hour
Quick Troubleshooting
1
Check API Key
curl https://api.u1.archetypeai.app/v0.5/lens/info \
-H "Authorization: Bearer $ATAI_API_KEY"
2
List Available Lenses
curl https://api.u1.archetypeai.app/v0.5/lens/metadata \
-H "Authorization: Bearer $ATAI_API_KEY" | jq '.[].lens_id'
3
Check Active Sessions
curl https://api.u1.archetypeai.app/v0.5/lens/sessions/info \
-H "Authorization: Bearer $ATAI_API_KEY"
4
Clean Up Sessions
# Get all sessions and destroy them
curl https://api.u1.archetypeai.app/v0.5/lens/sessions/metadata \
-H "Authorization: Bearer $ATAI_API_KEY" \
| jq -r '.[].session_id' \
| xargs -I {} curl -X POST https://api.u1.archetypeai.app/v0.5/lens/sessions/destroy \
-H "Authorization: Bearer $ATAI_API_KEY" \
-H "Content-Type: application/json" \
-d "{\"session_id\": \"{}\"}"
Resources
Live Monitor
View real-time API usage and active sessions
Newton Console
Manage API keys and organization settings
System Status
Check API health and service status
Support
Get help from our technical team