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 requireORG_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 — useawait. - 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:
- id →
str: User UUID. - external_id →
Optional[str]: External IdP identifier. - email →
str: User email address. - display_name →
Optional[str]: Human-friendly display name. - status →
str: User status (active,inactive, ordeleted). - created_at →
datetime: Creation timestamp. - updated_at →
datetime: Last update timestamp. - last_login_at →
Optional[datetime]: Last login timestamp. - org_id →
Optional[str]: Organization UUID (included when returned via org-scoped endpoints). - role →
Optional[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:
- id →
str: Membership UUID. - org_id →
str: Organization UUID. - user_id →
str: User UUID. - role →
str: RBAC role (ORG_ADMIN,ORG_MEMBER, orORG_VIEWER). - created_at →
datetime: 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, orORG_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:
- id →
str: Workspace UUID. - org_id →
str: Parent organization UUID. - name →
str: Workspace name (unique within the organization). - description →
Optional[str]: Workspace description. - status →
str: Workspace status (active,archived, ordeleted). - created_at →
datetime: Creation timestamp. - updated_at →
datetime: 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:
- id →
str: Project UUID. - workspace_id →
str: Parent workspace UUID. - org_id →
str: Parent organization UUID. - name →
str: Project name (unique within the workspace). - description →
Optional[str]: Project description. - created_by →
Optional[str]: UUID of the user who created the project. - status →
str: Project status (active,archived, ordeleted). - created_at →
datetime: Creation timestamp. - updated_at →
datetime: 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 aTokenProviderwhen 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, or365). Default:90.
Returns:¶
Returns an APIKeyCreated object containing:
- api_key →
str: The full API key. This is shown once — store it securely. - api_key_id →
str: UUID of the API key. - organization_id →
str: Organization UUID. - label →
str: Human-readable label. - permission →
str: Permission level (READ_ONLYorREAD_WRITE). - role →
str: RBAC role associated with the key. - expires_date →
str: Expiry date. - created_date →
str: Creation date. - updated_date →
str: 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:
- message →
str: Validation status message. - organization_id →
str: 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_id →
str: UUID of the API key. - user_id →
Optional[str]: UUID of the key owner. - organization_id →
str: Organization UUID. - label →
Optional[str]: Human-readable label. - permission →
str: Permission level (READ_ONLYorREAD_WRITE). - role →
str: RBAC role. - expires_date →
str: Expiry date. - created_date →
str: Creation date. - updated_date →
str: 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()