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

  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.

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

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