2. REST API¶
Agora exposes a REST API at /api/v2/. All responses are JSON. Authentication uses a personal API key (see User Profile and Settings).
2.1. Authentication¶
Pass your API key in the Authorization header on every request:
Authorization: Token <your-api-key>
import requests
BASE = "https://agora.example.com/api/v2"
HEADERS = {"Authorization": "Token <your-api-key>"}
r = requests.get(f"{BASE}/project/", headers=HEADERS)
r.raise_for_status()
All examples below assume BASE and HEADERS are defined as above.
2.2. Pagination¶
List endpoints return paginated results:
{
"count": 120,
"next": "https://agora.example.com/api/v2/project/?page=2",
"previous": null,
"results": [ ... ]
}
Use the page query parameter to navigate pages, or no_page=1 to disable pagination and receive all results in a single response.
2.3. Projects¶
Method |
Endpoint |
Description |
|---|---|---|
GET |
|
List all projects the current user is a member of |
GET |
|
Get a single project |
POST |
|
Create a project (body: |
GET |
|
Get the current user’s personal MyAgora project |
# List projects
projects = requests.get(f"{BASE}/project/", headers=HEADERS).json()["results"]
# Get MyAgora project
myagora = requests.get(f"{BASE}/project/myagora/", headers=HEADERS).json()
2.4. Folders¶
Method |
Endpoint |
Description |
|---|---|---|
GET |
|
List folders (filter by project: |
GET |
|
Get a single folder |
GET |
|
List the contents of a folder (subfolders, studies, datasets) |
POST |
|
Create a folder (body: |
GET |
|
Get the root folder of a project |
# Get root folder of a project
root = requests.get(f"{BASE}/project/42/root_folder/", headers=HEADERS).json()
root_id = root["id"]
# List folder contents
items = requests.get(f"{BASE}/folder/{root_id}/items/", headers=HEADERS).json()
2.5. Studies (Exams)¶
In the API, studies are called exams.
Method |
Endpoint |
Description |
|---|---|---|
GET |
|
List exams (filter by project: |
GET |
|
Get a single exam |
GET |
|
List the series belonging to an exam |
POST |
|
Copy an exam to a folder (body: |
POST |
|
Move an exam to a folder (body: |
2.6. Series¶
Method |
Endpoint |
Description |
|---|---|---|
GET |
|
List series (filter by exam: |
GET |
|
Get a single series |
GET |
|
List datasets belonging to a series |
2.7. Datasets¶
Method |
Endpoint |
Description |
|---|---|---|
GET |
|
List datasets (filter by series: |
GET |
|
Get a single dataset |
GET |
|
List the files belonging to a dataset |
GET |
|
List acquisition parameters attached to a dataset |
# Walk from project to files
exam = requests.get(f"{BASE}/exam/7/", headers=HEADERS).json()
series_list = requests.get(f"{BASE}/exam/7/series/", headers=HEADERS).json()
datasets = requests.get(f"{BASE}/series/3/datasets/", headers=HEADERS).json()
files = requests.get(f"{BASE}/dataset/12/datafiles/", headers=HEADERS).json()
2.8. Uploading Data¶
Uploads use a three-step session protocol:
Create an import session
Upload files (supports chunked/resumable upload)
Complete the session to trigger import
Method |
Endpoint |
Description |
|---|---|---|
POST |
|
Create an import session (body: |
GET |
|
Get import session status |
POST |
|
Upload a file (multipart |
POST |
|
Mark all uploads done and start the import |
# Create session
session = requests.post(f"{BASE}/importsession/",
json={"folder": root_id},
headers=HEADERS).json()
sid = session["id"]
# Upload a file
with open("scan.zip", "rb") as f:
requests.post(f"{BASE}/importsession/{sid}/upload/",
files={"file": f},
headers=HEADERS)
# Trigger import
requests.post(f"{BASE}/importsession/{sid}/complete/", headers=HEADERS)
2.9. Downloading Data¶
Downloads are initiated asynchronously. Agora packages the requested objects and provides a stream URL once ready.
Method |
Endpoint |
Description |
|---|---|---|
POST |
|
Initiate a download (body: list of |
GET |
|
Poll status ( |
GET |
|
Download the ZIP once state is FINISHED |
import time
dl = requests.post(f"{BASE}/download/",
json=[{"object_type": "exam", "object_id": 7}],
headers=HEADERS).json()
dl_id = dl["id"]
# Poll until ready
while True:
status = requests.get(f"{BASE}/download/{dl_id}/", headers=HEADERS).json()
if status["state"] == "FINISHED":
break
time.sleep(2)
# Stream the ZIP
with requests.get(f"{BASE}/download/{dl_id}/stream/",
headers=HEADERS, stream=True) as r:
with open("download.zip", "wb") as f:
for chunk in r.iter_content(chunk_size=8192):
f.write(chunk)
2.10. Tasks¶
Method |
Endpoint |
Description |
|---|---|---|
GET |
|
List task definitions available to the current user |
GET |
|
Get a task definition (includes expected inputs) |
POST |
|
Run a task (see below) |
GET |
|
List timeline items (task runs, imports, downloads) |
GET |
|
Get a single timeline item (includes state, output, log) |
POST |
|
Cancel a running task |
To run a task, POST to /runtask/ with the task ID and a list of input objects:
payload = {
"task": <task_id>,
"inputs": [
{"object_type": "dataset", "object_id": 12}
]
}
result = requests.post(f"{BASE}/runtask/", json=payload, headers=HEADERS).json()
timeline_id = result["timeline_item_id"]
Poll the timeline item to track progress:
while True:
item = requests.get(f"{BASE}/timelineitem/{timeline_id}/", headers=HEADERS).json()
state = item["state"] # QUEUED / STARTED / FINISHED / ERROR / CANCELED
if state in ("FINISHED", "ERROR", "CANCELED"):
break
time.sleep(3)
print(item.get("output")) # task stdout/stderr
2.11. Search¶
Method |
Endpoint |
Description |
|---|---|---|
GET |
|
Quick full-text search across all object types |
POST |
|
Run an AQL query (body: |
GET |
|
List saved queries |
POST |
|
Save a query (body: |
POST |
|
Run a saved query |
# AQL query
results = requests.post(f"{BASE}/search/query/",
json={"query": "exam.name ~ \"brain\" AND exam.start_time.year = 2024"},
headers=HEADERS).json()
See Database Search for the full AQL syntax reference.
2.12. Current User¶
Method |
Endpoint |
Description |
|---|---|---|
GET |
|
Get the current user’s profile and settings |
PATCH |
|
Update profile fields (e.g. |
me = requests.get(f"{BASE}/user/me/", headers=HEADERS).json()
print(me["username"], me["email"])