Skip to content

Governance API

The Governance API provides role-based access control (RBAC) for your AI Refinery platform. You can manage organizations, users, memberships, workspaces, projects, and API keys through the AIRefinery or AsyncAIRefinery client.

Note: All governance operations are automatically scoped to your organization. You do not need to specify an organization ID for most operations.

Prerequisites

To use the Governance API, you need:

  • An AI Refinery API key — see Creating API Keys.
  • A role with sufficient permissions for the operations you intend to perform. Most read operations require at least ORG_VIEWER; write operations require ORG_ADMIN. See the Permissions Matrix below.

Overview

The AIRefinery and AsyncAIRefinery clients expose the governance sub-clients through client.governance:

Sub-client Description
client.governance.organizations Retrieve organization details
client.governance.users Create, list, and look up users
client.governance.memberships Manage organization memberships and roles
client.governance.workspaces Create and list workspaces
client.governance.projects List projects
client.governance.api_keys Create, validate, inspect, and revoke API keys
  • On AsyncAIRefinery, all methods are async — use await.
  • On AIRefinery, all methods are synchronous blocking calls.

You can also import the governance client directly for standalone use:

from air import AsyncGovernanceClient  # or GovernanceClient

governance = AsyncGovernanceClient(api_key=api_key)

Resource Hierarchy

Governance resources are organized in the following hierarchy:

Organization
├── Users (via Memberships, each with an RBAC role)
├── Workspaces
│   └── Projects
└── API Keys
  • An Organization is the top-level tenancy boundary. All other resources belong to an organization.
  • Users are linked to organizations through Memberships, each carrying an RBAC role.
  • Workspaces are logical groupings within an organization. A default workspace is created automatically.
  • Projects live inside workspaces and represent individual AI Refinery configurations.
  • API Keys are scoped to an organization and inherit the creating user's role.

RBAC Roles

The platform supports three organizational roles, each with increasing levels of access:

Role Description
ORG_ADMIN Full control within the organization — manage users, workspaces, memberships, and API keys.
ORG_MEMBER Standard operational access — create projects, run inference, and manage own API keys.
ORG_VIEWER Read-only access — view organization resources without making changes.

Permissions Matrix

The table below shows which roles can perform each governance operation:

Operation ORG_ADMIN ORG_MEMBER ORG_VIEWER
Get organization Yes Yes Yes
Create user Yes
List users Yes Yes Yes
Get current user Yes Yes Yes
Create membership Yes
List memberships Yes Yes Yes
Update role Yes
Create workspace Yes
List workspaces Yes Yes Yes
List projects Yes Yes Yes
Create API key Yes Yes
Validate / inspect API key Yes Yes Yes
Revoke API key Yes Yes

Common Parameters

All governance methods accept the following optional keyword arguments:

  • timeout (float, Optional): Request timeout in seconds. Default: 60.
  • extra_headers (dict, Optional): Per-request header overrides merged on top of the client's default headers.

Exceptions

All governance methods may raise exceptions from air.governance.exceptions. See Governance Exceptions for the full hierarchy.


Getting Started

This end-to-end example demonstrates a common workflow: retrieving your organization, listing users, and creating a workspace.

import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def setup_workspace():
    client = AsyncAIRefinery(api_key=api_key)

    # 1. Retrieve the caller's organization
    org = await client.governance.organizations.me()
    print(f"Organization: {org.display_name or org.name}")

    # 2. List existing users in the organization
    users = await client.governance.users.list()
    print(f"Users in org: {len(users)}")

    # 3. Create a new workspace for the team
    workspace = await client.governance.workspaces.create(
        name="data-science-team",
        description="Workspace for DS experiments",
    )
    print(f"Created workspace: {workspace.name} (ID: {workspace.id})")

    # 4. Add a new team member
    user = await client.governance.users.create(
        email="bob@example.com",
        display_name="Bob Jones",
        role="ORG_MEMBER",
    )
    print(f"Added user: {user.display_name}")


if __name__ == "__main__":
    asyncio.run(setup_workspace())

Organizations

Retrieve details about your organization. Organizations are the top-level tenancy boundary — users, workspaces, and projects all belong to an organization.

Asynchronous Methods

client.governance.organizations.me()

Retrieve the caller's organization.

Returns:

Returns an Organization object (same structure as create()).

Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def get_my_organization():
    client = AsyncAIRefinery(api_key=api_key)

    # Retrieve the caller's organization
    org = await client.governance.organizations.me()

    print(f"Organization: {org.display_name or org.name}")
    print(f"ID: {org.id}")
    print(f"Status: {org.status}")


if __name__ == "__main__":
    asyncio.run(get_my_organization())

Synchronous Methods

AIRefinery.governance.organizations.me()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.organizations.me()) described above.

Example Usage:
import os

from air import AIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


def get_my_organization():
    client = AIRefinery(api_key=api_key)

    org = client.governance.organizations.me()
    print(f"Organization: {org.display_name or org.name} ({org.status})")


if __name__ == "__main__":
    get_my_organization()

Users

Create and manage users within your organization. Users are associated with organizations through memberships and assigned RBAC roles.

Asynchronous Methods

client.governance.users.create()

Create a new user and add them to your organization. Requires ORG_ADMIN role.

Parameters:
  • email (string, Required): User email address.
  • external_id (string, Optional): External identity provider identifier (e.g., Azure AD oid).
  • display_name (string, Optional): Human-friendly display name.
  • role (string, Optional): RBAC role to assign. Default: ORG_MEMBER.
Returns:

Returns a User object containing:

  • idstr: User UUID.
  • external_idOptional[str]: External IdP identifier.
  • emailstr: User email address.
  • display_nameOptional[str]: Human-friendly display name.
  • statusstr: User status (active, inactive, or deleted).
  • created_atdatetime: Creation timestamp.
  • updated_atdatetime: Last update timestamp.
  • last_login_atOptional[datetime]: Last login timestamp.
  • org_idOptional[str]: Organization UUID (included when returned via org-scoped endpoints).
  • roleOptional[str]: RBAC role in the organization (included when returned via org-scoped endpoints).
Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def create_user():
    client = AsyncAIRefinery(api_key=api_key)

    # Create a new user with ORG_MEMBER role (default)
    user = await client.governance.users.create(
        email="alice@example.com",
        display_name="Alice Smith",
    )

    print(f"Created user: {user.display_name} ({user.email})")
    print(f"User ID: {user.id}")
    print(f"Status: {user.status}")


if __name__ == "__main__":
    asyncio.run(create_user())

client.governance.users.list()

List all users in the caller's organization.

Parameters:
  • org_id (string, Optional): Override organization UUID (admin use only).
Returns:

Returns a list[User] — a list of User objects (same structure as create()).

Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def list_users():
    client = AsyncAIRefinery(api_key=api_key)

    # List all users in the caller's organization
    users = await client.governance.users.list()

    print(f"Found {len(users)} users:")
    for user in users:
        print(f"  {user.email}{user.role or 'N/A'} ({user.status})")


if __name__ == "__main__":
    asyncio.run(list_users())

client.governance.users.me()

Get the current authenticated user.

Returns:

Returns a User object (same structure as create()).

Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def get_current_user():
    client = AsyncAIRefinery(api_key=api_key)

    user = await client.governance.users.me()
    print(f"Logged in as: {user.display_name or user.email}")
    print(f"User ID: {user.id}")


if __name__ == "__main__":
    asyncio.run(get_current_user())

Synchronous Methods

AIRefinery.governance.users.create()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.users.create()) described above.

AIRefinery.governance.users.list()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.users.list()) described above.

AIRefinery.governance.users.me()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.users.me()) described above.

Example Usage:
import os

from air import AIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


def list_users():
    client = AIRefinery(api_key=api_key)

    users = client.governance.users.list()
    for user in users:
        print(f"{user.email}{user.role or 'N/A'}")


if __name__ == "__main__":
    list_users()

Memberships

Manage the association between users and organizations. Memberships control which users belong to which organizations and what role they hold.

Asynchronous Methods

client.governance.memberships.create()

Add a user to your organization. Requires ORG_ADMIN role.

Parameters:
  • user_id (string, Required): UUID of the user to add.
  • org_id (string, Optional): Target organization UUID. Defaults to the caller's organization.
  • role (string, Optional): RBAC role to assign. Default: ORG_MEMBER.
Returns:

Returns an OrgMembership object containing:

  • idstr: Membership UUID.
  • org_idstr: Organization UUID.
  • user_idstr: User UUID.
  • rolestr: RBAC role (ORG_ADMIN, ORG_MEMBER, or ORG_VIEWER).
  • created_atdatetime: Creation timestamp.
Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def add_user_to_org():
    client = AsyncAIRefinery(api_key=api_key)

    # Add an existing user to the caller's organization as a viewer
    membership = await client.governance.memberships.create(
        user_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        role="ORG_VIEWER",
    )

    print(f"Membership ID: {membership.id}")
    print(f"Role: {membership.role}")


if __name__ == "__main__":
    asyncio.run(add_user_to_org())

client.governance.memberships.list()

List all memberships in the caller's organization.

Parameters:
  • org_id (string, Optional): Override organization UUID (admin use only).
Returns:

Returns a list[OrgMembership] — a list of OrgMembership objects (same structure as create()).

Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def list_memberships():
    client = AsyncAIRefinery(api_key=api_key)

    memberships = await client.governance.memberships.list()

    print(f"Found {len(memberships)} memberships:")
    for m in memberships:
        print(f"  User {m.user_id}{m.role}")


if __name__ == "__main__":
    asyncio.run(list_memberships())

client.governance.memberships.update_role()

Update a user's role within the organization.

Parameters:
  • user_id (string, Required): UUID of the user whose role to update.
  • role (string, Required): New role (ORG_ADMIN, ORG_MEMBER, or ORG_VIEWER).
  • org_id (string, Optional): Override organization UUID (admin use only).
Returns:

This method does not return a value.

Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def promote_user():
    client = AsyncAIRefinery(api_key=api_key)

    # Promote a user to ORG_ADMIN
    await client.governance.memberships.update_role(
        user_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        role="ORG_ADMIN",
    )

    print("Role updated successfully.")


if __name__ == "__main__":
    asyncio.run(promote_user())

Synchronous Methods

AIRefinery.governance.memberships.create()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.memberships.create()) described above.

AIRefinery.governance.memberships.list()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.memberships.list()) described above.

AIRefinery.governance.memberships.update_role()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.memberships.update_role()) described above.

Example Usage:
import os

from air import AIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


def list_memberships():
    client = AIRefinery(api_key=api_key)

    memberships = client.governance.memberships.list()
    for m in memberships:
        print(f"User {m.user_id}{m.role}")


if __name__ == "__main__":
    list_memberships()

Workspaces

Workspaces are logical groupings of projects within an organization. Every organization is seeded with a default workspace on creation.

Asynchronous Methods

client.governance.workspaces.create()

Create a workspace in the caller's organization.

Parameters:
  • name (string, Required): Workspace name (must be unique within the organization).
  • description (string, Optional): Workspace description.
  • org_id (string, Optional): Override organization UUID (admin use only).
Returns:

Returns a Workspace object containing:

  • idstr: Workspace UUID.
  • org_idstr: Parent organization UUID.
  • namestr: Workspace name (unique within the organization).
  • descriptionOptional[str]: Workspace description.
  • statusstr: Workspace status (active, archived, or deleted).
  • created_atdatetime: Creation timestamp.
  • updated_atdatetime: Last update timestamp.
Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def create_workspace():
    client = AsyncAIRefinery(api_key=api_key)

    workspace = await client.governance.workspaces.create(
        name="ml-experiments",
        description="Machine learning experimentation workspace",
    )

    print(f"Created workspace: {workspace.name} (ID: {workspace.id})")
    print(f"Organization: {workspace.org_id}")


if __name__ == "__main__":
    asyncio.run(create_workspace())

client.governance.workspaces.list()

List all workspaces in the caller's organization.

Parameters:
  • org_id (string, Optional): Override organization UUID (admin use only).
Returns:

Returns a list[Workspace] — a list of Workspace objects (same structure as create()).

Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def list_workspaces():
    client = AsyncAIRefinery(api_key=api_key)

    workspaces = await client.governance.workspaces.list()

    print(f"Found {len(workspaces)} workspaces:")
    for ws in workspaces:
        print(f"  {ws.name}{ws.status}")


if __name__ == "__main__":
    asyncio.run(list_workspaces())

Synchronous Methods

AIRefinery.governance.workspaces.create()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.workspaces.create()) described above.

AIRefinery.governance.workspaces.list()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.workspaces.list()) described above.

Example Usage:
import os

from air import AIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


def list_workspaces():
    client = AIRefinery(api_key=api_key)

    workspaces = client.governance.workspaces.list()
    for ws in workspaces:
        print(f"{ws.name}{ws.description or 'No description'}")


if __name__ == "__main__":
    list_workspaces()

Projects

List projects registered within your organization. Projects belong to workspaces and represent individual AI Refinery configurations.

Note: The projects sub-client is read-only. To create projects, use the Distiller API.

Asynchronous Method

client.governance.projects.list()

List all projects in the caller's organization.

Parameters:
  • org_id (string, Optional): Override organization UUID (admin use only).
Returns:

Returns a list[Project] — a list of Project objects, each containing:

  • idstr: Project UUID.
  • workspace_idstr: Parent workspace UUID.
  • org_idstr: Parent organization UUID.
  • namestr: Project name (unique within the workspace).
  • descriptionOptional[str]: Project description.
  • created_byOptional[str]: UUID of the user who created the project.
  • statusstr: Project status (active, archived, or deleted).
  • created_atdatetime: Creation timestamp.
  • updated_atdatetime: Last update timestamp.
Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def list_projects():
    client = AsyncAIRefinery(api_key=api_key)

    projects = await client.governance.projects.list()

    print(f"Found {len(projects)} projects:")
    for proj in projects:
        print(f"  {proj.name} — workspace: {proj.workspace_id} ({proj.status})")


if __name__ == "__main__":
    asyncio.run(list_projects())

Synchronous Method

AIRefinery.governance.projects.list()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.projects.list()) described above.

Example Usage:
import os

from air import AIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


def list_projects():
    client = AIRefinery(api_key=api_key)

    projects = client.governance.projects.list()
    for proj in projects:
        print(f"{proj.name}{proj.status}")


if __name__ == "__main__":
    list_projects()

API Keys

Manage API key lifecycle — create, validate, inspect, and revoke keys programmatically.

Note: The create() method requires Studio JWT authentication, not a static API key. Use a TokenProvider when initializing the client if you need to create API keys programmatically.

Asynchronous Methods

client.governance.api_keys.create()

Create a new API key for an organization.

Parameters:
  • organization_id (string, Required): UUID of the target organization. The caller must be a member of this organization.
  • tenant_id (string, Optional): User UUID. Ignored when governance is enabled (the server uses the authenticated caller's identity). Default: "00000000-0000-0000-0000-000000000000".
  • label (string, Optional): Human-readable label for the key. Default: "default".
  • lifespans (integer, Optional): Validity period in days (30, 60, 90, 180, or 365). Default: 90.
Returns:

Returns an APIKeyCreated object containing:

  • api_keystr: The full API key. This is shown once — store it securely.
  • api_key_idstr: UUID of the API key.
  • organization_idstr: Organization UUID.
  • labelstr: Human-readable label.
  • permissionstr: Permission level (READ_ONLY or READ_WRITE).
  • rolestr: RBAC role associated with the key.
  • expires_datestr: Expiry date.
  • created_datestr: Creation date.
  • updated_datestr: Last update date.
Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def create_api_key():
    client = AsyncAIRefinery(api_key=api_key)

    # Create an API key valid for 90 days
    result = await client.governance.api_keys.create(
        organization_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        label="ci-pipeline",
        lifespans=90,
    )

    # Store this key securely — it is only shown once
    print(f"API Key: {result.api_key}")
    print(f"Key ID: {result.api_key_id}")
    print(f"Expires: {result.expires_date}")


if __name__ == "__main__":
    asyncio.run(create_api_key())

client.governance.api_keys.validate()

Validate the current API key and return the associated organization context.

Returns:

Returns a dict with:

  • messagestr: Validation status message.
  • organization_idstr: The organization UUID associated with the key.
Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def validate_key():
    client = AsyncAIRefinery(api_key=api_key)

    result = await client.governance.api_keys.validate()
    print(f"Message: {result['message']}")
    print(f"Organization: {result['organization_id']}")


if __name__ == "__main__":
    asyncio.run(validate_key())

client.governance.api_keys.inspect()

Inspect an API key's metadata without using it for authentication.

Parameters:
  • api_key (string, Required): The API key to inspect.
Returns:

Returns an APIKeyInfo object containing:

  • api_key_idstr: UUID of the API key.
  • user_idOptional[str]: UUID of the key owner.
  • organization_idstr: Organization UUID.
  • labelOptional[str]: Human-readable label.
  • permissionstr: Permission level (READ_ONLY or READ_WRITE).
  • rolestr: RBAC role.
  • expires_datestr: Expiry date.
  • created_datestr: Creation date.
  • updated_datestr: Last update date.
Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def inspect_key():
    client = AsyncAIRefinery(api_key=api_key)

    info = await client.governance.api_keys.inspect(
        api_key="air_...",
    )

    print(f"Key ID: {info.api_key_id}")
    print(f"Role: {info.role}")
    print(f"Expires: {info.expires_date}")


if __name__ == "__main__":
    asyncio.run(inspect_key())

client.governance.api_keys.revoke()

Revoke an API key. Exactly one of api_key or api_key_id must be provided.

Parameters:
  • api_key (string, Optional): The full API key to revoke.
  • api_key_id (string, Optional): UUID of the API key to revoke.
Returns:

This method does not return a value.

Example Usage:
import asyncio
import os

from air import AsyncAIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


async def revoke_key():
    client = AsyncAIRefinery(api_key=api_key)

    # Revoke by key ID
    await client.governance.api_keys.revoke(
        api_key_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    )

    print("API key revoked successfully.")


if __name__ == "__main__":
    asyncio.run(revoke_key())

Synchronous Methods

AIRefinery.governance.api_keys.create()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.api_keys.create()) described above.

AIRefinery.governance.api_keys.validate()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.api_keys.validate()) described above.

AIRefinery.governance.api_keys.inspect()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.api_keys.inspect()) described above.

AIRefinery.governance.api_keys.revoke()

This method supports the same parameters and return structure as the asynchronous method (AsyncAIRefinery.governance.api_keys.revoke()) described above.

Example Usage:
import os

from air import AIRefinery
from dotenv import load_dotenv

load_dotenv()
api_key = str(os.getenv("API_KEY"))


def manage_api_keys():
    client = AIRefinery(api_key=api_key)

    # Validate the current key
    result = client.governance.api_keys.validate()
    print(f"Valid — org: {result['organization_id']}")

    # Inspect a key
    info = client.governance.api_keys.inspect(api_key="air_...")
    print(f"Key role: {info.role}, expires: {info.expires_date}")


if __name__ == "__main__":
    manage_api_keys()