API overview
Quickstart
Want to jump-in and interact with Mage via API? Replace the variables at the top of this snippet and use it to get started with authetication!
Initialize Session
import requests
import json
API_KEY = "YOUR_API_KEY"
EMAIL = "EMAIL"
PASSWORD = "USER_PASSWORD"
base_url = 'http://localhost:6789'
headers = {
"Content-Type": "application/json",
"X-API-KEY": API_KEY
}
json_data = {
"session": {
"email": EMAIL,
"password": PASSWORD
}
}
s = requests.Session()
s.headers = headers
sessions_url = base_url + "/api/sessions"
r = s.post(url=sessions_url, json=json_data)
mage_session = r.json().get('session')
Authenticate
token = mage_session.get('token')
s.headers['Cookie'] = f'oauth_token={token}'
Get Pipelines
pipelines_url = base_url + '/api/pipelines'
r = s.get(pipelines_url)
list_pipelines = r.json()
print(list_pipelines)
Components
There are 3 components to an API endpoint:
You’ll need to create each of those 3 components for every new API endpoint.
Refer to each documentation on how to create and configure them.
Testing
Create a test file in mage_ai/tests/api/operations/test_[resource_name].py
;
replace resource_name
with the name of your resource.
Paste in the following sample code into your file:
from mage_ai.orchestration.db.models.oauth import User
from mage_ai.tests.api.operations.base import BaseApiTestCase
from mage_ai.tests.factory import create_user
class SomeResourceOperationTests(BaseApiTestCase):
model_class = User
async def test_execute_delete(self):
user = create_user(email=self.faker.email())
await self.base_test_execute_delete(user.id)
async def test_execute_detail(self):
user = create_user()
await self.base_test_execute_detail(user.id, dict(
email=user.email,
username=user.username,
))
This is only a sample test file. It’s running tests on the UserResource
.
You’ll need to update the test logic to unit test your custom API endpoint.
Authentication
Most API endpoints require authentication. The API request must include the following:
- API key
- OAuth token
API key
The API request must include the API key in the query parameter as api_key
or in the request payload body as api_key
. You can find your API key quickly by navigating to the GUI and checking the request log from the container:

OAuth token
The OAuth token is generated by creating a new session.
The token provided in the session payload is the raw value. There are two ways to authenticate with this token:
- Supply the raw token in the header as:
{"Cookie": "oauth_token=[RAW-TOKEN-FROM-RESPONSE]"}
- Supply the decoded token in the header as:
{"OAUTH-TOKEN": "[DECODED-TOKEN-FROM-RESPONSE]"}
To decode a token:
from mage_ai.authentication.auth2 import decode_token
decoded_token = decode_token(token_from_session_response) ['token']
Examples
Decoded token, API key in in URL:
curl -X GET "localhost:6789/api/pipelines?api_key=zkWlN0PkIKSN0C11CfUHUj84OT5XOJ6tDZ6bDRO2" \
-H "OAUTH-TOKEN: [DECODED-TOKEN-FROM-RESPONSE]"
Decoded token, API key in body:
curl -X POST "localhost:6789/api/pipelines" \
-H "OAUTH-TOKEN: [DECODED-TOKEN-FROM-RESPONSE]" \
-d '{
"api_key": "zkWlN0PkIKSN0C11CfUHUj84OT5XOJ6tDZ6bDRO2",
"pipeline": {}
}'
Raw token, API key in body:
curl -X POST "localhost:6789/api/pipelines" \
-H "Cookie: oauth_token=[RAW-TOKEN-FROM-RESPONSE]" \
-d '{
"api_key": "zkWlN0PkIKSN0C11CfUHUj84OT5XOJ6tDZ6bDRO2",
"pipeline": {}
}'
Responses
All our responses return a 200
status in the main payload. The any error code will be stored in the error
key. For example, the following is a response from a failed session auth:
{
"error": {
"code": 404,
"message": "Email/username and/or password invalid.",
"type": "record_not_found"
},
"status": 200
}
To check for an error, one might try a solution like the following:
err_code = r.json().get('error').get('code'):