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


base_url = 'http://localhost:6789'

headers = {
    "Content-Type": "application/json",

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


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



There are 3 components to an API endpoint:

  1. Resource
  2. Policy
  3. Presenter

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.


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(

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.


Most API endpoints require authentication. The API request must include the following:

  1. API key
  2. 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:

terminal with API key

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:

  1. Supply the raw token in the header as:
{"Cookie": "oauth_token=[RAW-TOKEN-FROM-RESPONSE]"}
  1. Supply the decoded token in the header as:

To decode a token:

from mage_ai.authentication.auth2 import decode_token

decoded_token = decode_token(token_from_session_response) ['token']


Decoded token, API key in in URL:

curl -X GET "localhost:6789/api/pipelines?api_key=zkWlN0PkIKSN0C11CfUHUj84OT5XOJ6tDZ6bDRO2" \

Decoded token, API key in body:

curl -X POST "localhost:6789/api/pipelines" \
    -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": {}


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