Authorization API Reference

January 10, 2026 · View on GitHub

This document provides a comprehensive reference for all authorization and RBAC (Role-Based Access Control) APIs available in the Conductor Python SDK.

📚 Complete Working Example: See authorization_journey.py for a comprehensive example.

Applications
Application Roles
Application Tags
Access Keys
Users
Groups
Group Users
Permissions
Roles
Token & Authentication
API Gateway Authentication

Applications

Manage applications in your Conductor instance.

Method	Endpoint	Description	Example
`create_application()`	`POST /api/applications`	Create a new application	Example
`get_application()`	`GET /api/applications/{id}`	Get application by ID	Example
`list_applications()`	`GET /api/applications`	List all applications	Example
`update_application()`	`PUT /api/applications/{id}`	Update an existing application	Example
`delete_application()`	`DELETE /api/applications/{id}`	Delete an application	Example
`get_app_by_access_key_id()`	`GET /api/applications/key/{accessKeyId}`	Get application ID by access key	Example

Create Application

from conductor.client.configuration.configuration import Configuration
from conductor.client.orkes.orkes_authorization_client import OrkesAuthorizationClient
from conductor.client.http.models.create_or_update_application_request import CreateOrUpdateApplicationRequest

configuration = Configuration()
auth_client = OrkesAuthorizationClient(configuration)

# Create application
request = CreateOrUpdateApplicationRequest(name="my-application")
app = auth_client.create_application(request)

print(f"Created application with ID: {app.id}")

Get Application

# Get application by ID
app = auth_client.get_application("app-id-123")
print(f"Application name: {app.name}")

List Applications

# List all applications
apps = auth_client.list_applications()
for app in apps:
    print(f"App ID: {app.id}, Name: {app.name}")

Update Application

# Update application
request = CreateOrUpdateApplicationRequest(name="my-updated-application")
updated_app = auth_client.update_application(request, "app-id-123")

Delete Application

# Delete application
auth_client.delete_application("app-id-123")

Get App By Access Key ID

# Get application ID by access key
app_id = auth_client.get_app_by_access_key_id("access-key-123")
print(f"Application ID: {app_id}")

Application Roles

Manage roles for application users.

Method	Endpoint	Description	Example
`add_role_to_application_user()`	`POST /api/applications/{applicationId}/roles/{role}`	Add a role to application user	Example
`remove_role_from_application_user()`	`DELETE /api/applications/{applicationId}/roles/{role}`	Remove a role from application user	Example

Available Roles:

USER - Basic user access
ADMIN - Administrative access
METADATA_MANAGER - Manage workflow/task definitions
WORKFLOW_MANAGER - Manage workflow executions
WORKER - Worker task execution access

Add Role To Application User

# Add role to application user
auth_client.add_role_to_application_user("app-id-123", "ADMIN")

Remove Role From Application User

# Remove role from application user
auth_client.remove_role_from_application_user("app-id-123", "ADMIN")

Application Tags

Manage tags for applications.

Method	Endpoint	Description	Example
`set_application_tags()`	`PUT /api/applications/{id}/tags`	Set/add tags to application	Example
`get_application_tags()`	`GET /api/applications/{id}/tags`	Get all tags for application	Example
`delete_application_tags()`	`DELETE /api/applications/{id}/tags`	Delete tags from application	Example

Set Application Tags

from conductor.client.orkes.models.metadata_tag import MetadataTag

# Set application tags
tags = [
    MetadataTag("environment", "production"),
    MetadataTag("team", "platform")
]
auth_client.set_application_tags(tags, "app-id-123")

Get Application Tags

# Get application tags
tags = auth_client.get_application_tags("app-id-123")
for tag in tags:
    print(f"Tag: {tag.key} = {tag.value}")

Delete Application Tags

# Delete specific tags
tags = [
    MetadataTag("environment", "production")
]
auth_client.delete_application_tags(tags, "app-id-123")

Access Keys

Manage access keys for applications.

Method	Endpoint	Description	Example
`create_access_key()`	`POST /api/applications/{id}/accessKeys`	Create a new access key	Example
`get_access_keys()`	`GET /api/applications/{id}/accessKeys`	Get all access keys for application	Example
`toggle_access_key_status()`	`POST /api/applications/{applicationId}/accessKeys/{keyId}/status`	Toggle access key active/inactive	Example
`delete_access_key()`	`DELETE /api/applications/{applicationId}/accessKeys/{keyId}`	Delete an access key	Example

Create Access Key

# Create access key
access_key = auth_client.create_access_key("app-id-123")

# IMPORTANT: Save the secret immediately - it's only shown once!
print(f"Key ID: {access_key.id}")
print(f"Secret: {access_key.secret}")  # Only available at creation time

Get Access Keys

# Get all access keys for an application
keys = auth_client.get_access_keys("app-id-123")
for key in keys:
    print(f"Key ID: {key.id}, Status: {key.status}")

Toggle Access Key Status

# Toggle access key between ACTIVE and INACTIVE
key = auth_client.toggle_access_key_status("app-id-123", "key-id-456")
print(f"New status: {key.status}")

Delete Access Key

# Delete access key
auth_client.delete_access_key("app-id-123", "key-id-456")

Users

Manage users in your Conductor instance.

Method	Endpoint	Description	Example
`upsert_user()`	`PUT /api/users/{id}`	Create or update a user	Example
`get_user()`	`GET /api/users/{id}`	Get user by ID	Example
`list_users()`	`GET /api/users`	List all users	Example
`delete_user()`	`DELETE /api/users/{id}`	Delete a user	Example
`get_granted_permissions_for_user()`	`GET /api/users/{userId}/permissions`	Get all permissions granted to user	Example
`check_permissions()`	`GET /api/users/{userId}/checkPermissions`	Check if user has specific permissions	Example

Upsert User

from conductor.client.http.models.upsert_user_request import UpsertUserRequest

# Create or update user
user_id = "user@example.com"
request = UpsertUserRequest(
    name="John Doe",
    roles=["USER", "METADATA_MANAGER"]
)
user = auth_client.upsert_user(request, user_id)
print(f"User created: {user.id}")

Get User

# Get user by ID
user = auth_client.get_user("user@example.com")
print(f"User name: {user.name}")
print(f"Roles: {user.roles}")

List Users

# List all users
users = auth_client.list_users()
for user in users:
    print(f"User: {user.id}, Name: {user.name}")

# List users including applications
users_with_apps = auth_client.list_users(apps=True)

Delete User

# Delete user
auth_client.delete_user("user@example.com")

Get Granted Permissions For User

# Get all permissions granted to user
permissions = auth_client.get_granted_permissions_for_user("user@example.com")
for perm in permissions:
    print(f"Target: {perm.target.type}:{perm.target.id}")
    print(f"Access: {perm.access}")

Check Permissions

# Check if user has specific permissions on a target
result = auth_client.check_permissions(
    user_id="user@example.com",
    target_type="WORKFLOW_DEF",
    target_id="my-workflow"
)
print(f"Has access: {result}")

Groups

Manage user groups in your Conductor instance.

Method	Endpoint	Description	Example
`upsert_group()`	`PUT /api/groups/{id}`	Create or update a group	Example
`get_group()`	`GET /api/groups/{id}`	Get group by ID	Example
`list_groups()`	`GET /api/groups`	List all groups	Example
`delete_group()`	`DELETE /api/groups/{id}`	Delete a group	Example
`get_granted_permissions_for_group()`	`GET /api/groups/{groupId}/permissions`	Get all permissions granted to group	Example

Upsert Group

from conductor.client.http.models.upsert_group_request import UpsertGroupRequest

# Create or update group
group_id = "engineering-team"
request = UpsertGroupRequest(
    description="Engineering Team",
    roles=["USER", "WORKFLOW_MANAGER"]
)
group = auth_client.upsert_group(request, group_id)
print(f"Group created: {group.id}")

Get Group

# Get group by ID
group = auth_client.get_group("engineering-team")
print(f"Group description: {group.description}")
print(f"Roles: {group.roles}")

List Groups

# List all groups
groups = auth_client.list_groups()
for group in groups:
    print(f"Group: {group.id}, Description: {group.description}")

Delete Group

# Delete group
auth_client.delete_group("engineering-team")

Get Granted Permissions For Group

# Get all permissions granted to group
permissions = auth_client.get_granted_permissions_for_group("engineering-team")
for perm in permissions:
    print(f"Target: {perm.target.type}:{perm.target.id}")
    print(f"Access: {perm.access}")

Group Users

Manage users within groups.

Method	Endpoint	Description	Example
`add_user_to_group()`	`POST /api/groups/{groupId}/users/{userId}`	Add a single user to group	Example
`add_users_to_group()`	`POST /api/groups/{groupId}/users`	Add multiple users to group	Example
`get_users_in_group()`	`GET /api/groups/{id}/users`	Get all users in group	Example
`remove_user_from_group()`	`DELETE /api/groups/{groupId}/users/{userId}`	Remove a single user from group	Example
`remove_users_from_group()`	`DELETE /api/groups/{groupId}/users`	Remove multiple users from group	Example

Add User To Group

# Add single user to group
auth_client.add_user_to_group("engineering-team", "user@example.com")

Add Users To Group

# Add multiple users to group (bulk operation)
user_ids = [
    "user1@example.com",
    "user2@example.com",
    "user3@example.com"
]
auth_client.add_users_to_group("engineering-team", user_ids)

Get Users In Group

# Get all users in a group
users = auth_client.get_users_in_group("engineering-team")
for user in users:
    print(f"User: {user.id}, Name: {user.name}")

Remove User From Group

# Remove single user from group
auth_client.remove_user_from_group("engineering-team", "user@example.com")

Remove Users From Group

# Remove multiple users from group (bulk operation)
user_ids = [
    "user1@example.com",
    "user2@example.com"
]
auth_client.remove_users_from_group("engineering-team", user_ids)

Permissions

Manage permissions and access control.

Method	Endpoint	Description	Example
`grant_permissions()`	`POST /api/auth/authorization`	Grant permissions to subject on target	Example
`get_permissions()`	`GET /api/auth/authorization/{type}/{id}`	Get all permissions for a target	Example
`remove_permissions()`	`DELETE /api/auth/authorization`	Remove permissions from subject on target	Example

Target Types:

WORKFLOW_DEF - Workflow definition
TASK_DEF - Task definition
APPLICATION - Application
USER - User
DOMAIN - Domain

Subject Types:

USER - Individual user
GROUP - User group
ROLE - Role

Access Types:

READ - Read access
CREATE - Create access
UPDATE - Update access
EXECUTE - Execute access
DELETE - Delete access

Grant Permissions

from conductor.client.http.models.target_ref import TargetRef, TargetType
from conductor.client.http.models.subject_ref import SubjectRef, SubjectType
from conductor.client.orkes.models.access_type import AccessType

# Grant permissions to a group on a workflow
target = TargetRef(TargetType.WORKFLOW_DEF, "order-processing-workflow")
subject = SubjectRef(SubjectType.GROUP, "engineering-team")
access = [AccessType.READ, AccessType.EXECUTE]

auth_client.grant_permissions(subject, target, access)

# Grant permissions to a user on a task
target = TargetRef(TargetType.TASK_DEF, "send-email-task")
subject = SubjectRef(SubjectType.USER, "user@example.com")
access = [AccessType.READ, AccessType.UPDATE]

auth_client.grant_permissions(subject, target, access)

Get Permissions

from conductor.client.http.models.target_ref import TargetRef, TargetType

# Get all permissions for a workflow
target = TargetRef(TargetType.WORKFLOW_DEF, "order-processing-workflow")
permissions = auth_client.get_permissions(target)

# permissions is a Dict[str, List[SubjectRef]]
# Key is AccessType, value is list of subjects with that access
for access_type, subjects in permissions.items():
    print(f"Access Type: {access_type}")
    for subject in subjects:
        print(f"  Subject: {subject.type}:{subject.id}")

Remove Permissions

from conductor.client.http.models.target_ref import TargetRef, TargetType
from conductor.client.http.models.subject_ref import SubjectRef, SubjectType
from conductor.client.orkes.models.access_type import AccessType

# Remove permissions from a group
target = TargetRef(TargetType.WORKFLOW_DEF, "order-processing-workflow")
subject = SubjectRef(SubjectType.GROUP, "engineering-team")
access = [AccessType.EXECUTE]

auth_client.remove_permissions(subject, target, access)

Roles

Manage custom roles and role-based access control.

Method	Endpoint	Description	Example
`list_all_roles()`	`GET /api/roles`	List all roles (system + custom)	Example
`list_system_roles()`	`GET /api/roles/system`	List system-defined roles	Example
`list_custom_roles()`	`GET /api/roles/custom`	List custom roles only	Example
`list_available_permissions()`	`GET /api/roles/permissions`	List all available permissions	Example
`create_role()`	`POST /api/roles`	Create a new custom role	Example
`get_role()`	`GET /api/roles/{name}`	Get role by name	Example
`update_role()`	`PUT /api/roles/{name}`	Update an existing custom role	Example
`delete_role()`	`DELETE /api/roles/{name}`	Delete a custom role	Example

List All Roles

# List all roles (system + custom)
roles = auth_client.list_all_roles()
for role in roles:
    print(f"Role: {role['name']}")
    print(f"  Description: {role.get('description', 'N/A')}")
    print(f"  Type: {role.get('type', 'custom')}")

List System Roles

# List system-defined roles
system_roles = auth_client.list_system_roles()
for role_name, role_data in system_roles.items():
    print(f"System Role: {role_name}")
    print(f"  Permissions: {role_data.get('permissions', [])}")

List Custom Roles

# List custom roles only
custom_roles = auth_client.list_custom_roles()
for role in custom_roles:
    print(f"Custom Role: {role['name']}")

List Available Permissions

# List all available permissions that can be assigned to roles
permissions = auth_client.list_available_permissions()
for resource_type, perms in permissions.items():
    print(f"Resource: {resource_type}")
    print(f"  Permissions: {perms}")

Create Role

# Create a custom role
role_request = {
    "name": "workflow-operator",
    "description": "Can execute and monitor workflows",
    "permissions": [
        {
            "resource": "WORKFLOW_DEF",
            "actions": ["READ", "EXECUTE"]
        },
        {
            "resource": "WORKFLOW",
            "actions": ["READ", "EXECUTE"]
        }
    ]
}
role = auth_client.create_role(role_request)
print(f"Created role: {role['name']}")

Get Role

# Get role by name
role = auth_client.get_role("workflow-operator")
print(f"Role: {role['name']}")
print(f"Permissions: {role['permissions']}")

Update Role

# Update an existing custom role
role_update = {
    "description": "Updated description",
    "permissions": [
        {
            "resource": "WORKFLOW_DEF",
            "actions": ["READ", "EXECUTE", "UPDATE"]
        }
    ]
}
updated_role = auth_client.update_role("workflow-operator", role_update)

Delete Role

# Delete a custom role
auth_client.delete_role("workflow-operator")

Token & Authentication

Manage authentication tokens and retrieve user information.

Method	Endpoint	Description	Example
`get_user_info_from_token()`	`GET /api/token/userInfo`	Get user info from current auth token	Example
`generate_token()`	`POST /api/token`	Generate JWT with access key credentials	Example

Get User Info From Token

# Get user information from the current authentication token
user_info = auth_client.get_user_info_from_token()

print(f"User ID: {user_info.get('id')}")
print(f"User Name: {user_info.get('name')}")
print(f"Roles: {user_info.get('roles')}")
print(f"Application: {user_info.get('application')}")

Generate Token

# Generate JWT token using access key credentials
token_response = auth_client.generate_token(
    key_id="your-access-key-id",
    key_secret="your-access-key-secret"
)

jwt_token = token_response.get('token')
expires_in = token_response.get('expiresIn')

print(f"JWT Token: {jwt_token}")
print(f"Expires in: {expires_in} seconds")

# Use this token for API authentication
configuration = Configuration()
configuration.set_authentication_settings(jwt_token)

API Gateway Authentication

Manage authentication configurations for the API Gateway.

Method	Endpoint	Description	Example
`create_gateway_auth_config()`	`POST /api/gateway/config/auth`	Create gateway auth configuration	Example
`get_gateway_auth_config()`	`GET /api/gateway/config/auth/{id}`	Get gateway auth configuration by ID	Example
`list_gateway_auth_configs()`	`GET /api/gateway/config/auth`	List all gateway auth configurations	Example
`update_gateway_auth_config()`	`PUT /api/gateway/config/auth/{id}`	Update gateway auth configuration	Example
`delete_gateway_auth_config()`	`DELETE /api/gateway/config/auth/{id}`	Delete gateway auth configuration	Example

Create Gateway Auth Config

# Create API Gateway authentication configuration
auth_config = {
    "name": "my-api-gateway-auth",
    "type": "BEARER",
    "enabled": True,
    "config": {
        "headerName": "Authorization",
        "headerPrefix": "Bearer",
        "validateToken": True
    }
}

config = auth_client.create_gateway_auth_config(auth_config)
config_id = config.get('id')
print(f"Created gateway auth config with ID: {config_id}")

Get Gateway Auth Config

# Get gateway auth configuration by ID
config = auth_client.get_gateway_auth_config("config-id-123")
print(f"Config name: {config.get('name')}")
print(f"Config type: {config.get('type')}")
print(f"Enabled: {config.get('enabled')}")

List Gateway Auth Configs

# List all gateway auth configurations
configs = auth_client.list_gateway_auth_configs()
for config in configs:
    print(f"ID: {config.get('id')}")
    print(f"Name: {config.get('name')}")
    print(f"Type: {config.get('type')}")
    print(f"Enabled: {config.get('enabled')}")
    print("---")

Update Gateway Auth Config

# Update gateway auth configuration
updated_config = {
    "name": "my-api-gateway-auth-updated",
    "type": "BEARER",
    "enabled": False,  # Disable the config
    "config": {
        "headerName": "X-API-Key",
        "headerPrefix": "ApiKey",
        "validateToken": True
    }
}

config = auth_client.update_gateway_auth_config("config-id-123", updated_config)
print(f"Updated config: {config.get('name')}")

Delete Gateway Auth Config

# Delete gateway auth configuration
auth_client.delete_gateway_auth_config("config-id-123")
print("Gateway auth config deleted successfully")

Models Reference

This section provides detailed information about all the models (data classes) used in authorization APIs.

Core Models

SubjectRef

Represents a user, group, or role that is granted or removed access.

Module: conductor.client.http.models.subject_ref

Properties:

type (str, required): The subject type - one of USER, ROLE, or GROUP
id (str, required): The identifier of the subject (e.g., user email, group ID, role name)

Subject Types:

USER - An individual user identified by email or user ID
ROLE - A role name
GROUP - A group identified by group ID

Example:

from conductor.client.http.models.subject_ref import SubjectRef, SubjectType

# User subject
user_subject = SubjectRef(SubjectType.USER, "user@example.com")

# Group subject
group_subject = SubjectRef(SubjectType.GROUP, "engineering-team")

# Role subject
role_subject = SubjectRef(SubjectType.ROLE, "workflow-operator")

TargetRef

Represents the object over which access is being granted or removed.

Module: conductor.client.http.models.target_ref

Properties:

type (str, required): The target type (see Target Types below)
id (str, required): The identifier of the target resource

Target Types:

WORKFLOW_DEF - Workflow definition (template)
WORKFLOW - Workflow execution instance
WORKFLOW_SCHEDULE - Scheduled workflow
TASK_DEF - Task definition
TASK_REF_NAME - Task reference name
TASK_ID - Specific task instance
APPLICATION - Application
USER - User
SECRET_NAME - Secret
ENV_VARIABLE - Environment variable
TAG - Tag
DOMAIN - Domain
INTEGRATION_PROVIDER - Integration provider
INTEGRATION - Integration
PROMPT - AI prompt template
USER_FORM_TEMPLATE - User form template
SCHEMA - Schema definition
CLUSTER_CONFIG - Cluster configuration
WEBHOOK - Webhook
API_GATEWAY_SERVICE - API Gateway service
API_GATEWAY_SERVICE_ROUTE - API Gateway service route

Example:

from conductor.client.http.models.target_ref import TargetRef, TargetType

# Workflow definition target
workflow_target = TargetRef(TargetType.WORKFLOW_DEF, "order-processing")

# Task definition target
task_target = TargetRef(TargetType.TASK_DEF, "send-email")

# Application target
app_target = TargetRef(TargetType.APPLICATION, "payment-service")

# Secret target
secret_target = TargetRef(TargetType.SECRET_NAME, "db-password")

AccessType

Enum representing the types of access that can be granted.

Module: conductor.client.orkes.models.access_type

Values:

READ - Read access to view the resource
CREATE - Create new instances
UPDATE - Modify existing resources
EXECUTE - Execute workflows or tasks
DELETE - Delete resources

Example:

from conductor.client.orkes.models.access_type import AccessType

# Grant read and execute permissions
permissions = [AccessType.READ, AccessType.EXECUTE]

# Grant full access
full_access = [AccessType.READ, AccessType.CREATE, AccessType.UPDATE, AccessType.EXECUTE, AccessType.DELETE]

MetadataTag

Represents a metadata tag for categorizing and organizing resources.

Module: conductor.client.orkes.models.metadata_tag

Properties:

key (str, required): The tag key/name
value (str, required): The tag value
type (str, auto-set): Always set to "METADATA"

Use Cases:

Categorize applications by environment (dev, staging, prod)
Tag resources by team, project, or cost center
Add custom metadata for organizational purposes

Example:

from conductor.client.orkes.models.metadata_tag import MetadataTag

# Create tags
tags = [
    MetadataTag("environment", "production"),
    MetadataTag("team", "platform"),
    MetadataTag("cost-center", "engineering"),
    MetadataTag("version", "2.0")
]

# Apply to application
auth_client.set_application_tags(tags, "my-app-id")

Application Models

ConductorApplication

Represents an application in the Conductor system.

Module: conductor.client.http.models.conductor_application

Properties:

id (str): Unique application identifier
name (str): Application name
createTime (int): Creation timestamp (epoch millis)
createdBy (str): User who created the application
updateTime (int): Last update timestamp
updatedBy (str): User who last updated the application

Note: Application tags are managed through separate tagging APIs (get_application_tags(), set_application_tags(), delete_application_tags()) and are not included in the ConductorApplication object itself.

Example:

# Get application
app = auth_client.get_application("app-id-123")
print(f"Application: {app.name}")
print(f"Created by: {app.createdBy}")

# Get tags separately
tags = auth_client.get_application_tags("app-id-123")
print(f"Tags: {[f'{tag.key}={tag.value}' for tag in tags] if tags else 'No tags'}")

CreateOrUpdateApplicationRequest

Request model for creating or updating an application.

Module: conductor.client.http.models.create_or_update_application_request

Properties:

name (str, required): Application name (e.g., "Payment Processors")

Example:

from conductor.client.http.models.create_or_update_application_request import CreateOrUpdateApplicationRequest

# Create new application
request = CreateOrUpdateApplicationRequest(name="My Service Application")
app = auth_client.create_application(request)

Access Key Models

AccessKey

Represents an access key for application authentication.

Module: conductor.client.orkes.models.access_key

Properties:

id (str): Access key ID
status (str): Key status - ACTIVE or INACTIVE
createTime (int): Creation timestamp
createdBy (str): User who created the key

Example:

# List access keys
keys = auth_client.get_access_keys("app-id-123")
for key in keys:
    print(f"Key ID: {key.id}")
    print(f"Status: {key.status}")
    print(f"Created: {key.createTime}")

CreatedAccessKey

Represents a newly created access key (includes the secret).

Module: conductor.client.orkes.models.created_access_key

Properties:

id (str): Access key ID
secret (str): Access key secret (ONLY available at creation time!)

⚠️ Important: The secret field is only returned when the access key is first created. You must save it immediately as it cannot be retrieved later!

Example:

# Create access key
created_key = auth_client.create_access_key("app-id-123")

# SAVE THESE IMMEDIATELY - secret is only shown once!
key_id = created_key.id
key_secret = created_key.secret

print(f"Key ID: {key_id}")
print(f"Secret: {key_secret}")  # Save this securely!

User and Group Models

ConductorUser

Represents a user in the Conductor system.

Module: conductor.client.http.models.conductor_user

Properties:

id (str): User ID (usually email)
name (str): Full name
roles (List[str]): Assigned roles
groups (List[str]): Group memberships
applicationUser (bool): Whether this is an application user
namespace (str): User namespace
uuid (str): Unique user identifier
contactInformation (Dict[str, str]): User contact information (email, phone, etc.)

Example:

# Get user
user = auth_client.get_user("user@example.com")
print(f"Name: {user.name}")
print(f"Roles: {user.roles}")
print(f"Groups: {user.groups}")
print(f"Namespace: {user.namespace}")
print(f"Contact: {user.contactInformation if user.contactInformation else 'Not provided'}")

UpsertUserRequest

Request model for creating or updating a user.

Module: conductor.client.http.models.upsert_user_request

Properties:

name (str, required): User's full name
roles (List[str], optional): Roles to assign to the user
groups (List[str], optional): IDs of groups the user belongs to

Available Roles:

USER - Basic user access
ADMIN - Full administrative access
METADATA_MANAGER - Manage workflow/task definitions
WORKFLOW_MANAGER - Manage workflow executions
WORKER - Worker task execution access

Example:

from conductor.client.http.models.upsert_user_request import UpsertUserRequest

# Create user request
request = UpsertUserRequest(
    name="John Doe",
    roles=["USER", "WORKFLOW_MANAGER"],
    groups=["engineering-team", "ops-team"]
)

user = auth_client.upsert_user(request, "john.doe@example.com")

Group

Represents a user group in the Conductor system.

Module: conductor.client.http.models.group

Properties:

id (str): Group ID
description (str): Group description
roles (List[str]): Roles assigned to the group
defaultAccess (Dict): Default access permissions for the group
contactInformation (Dict): Group contact information

Example:

# Get group
group = auth_client.get_group("engineering-team")
print(f"Description: {group.description}")
print(f"Roles: {group.roles}")

UpsertGroupRequest

Request model for creating or updating a group.

Module: conductor.client.http.models.upsert_group_request

Properties:

description (str, required): Description of the group
roles (List[str], optional): Roles to assign to the group
defaultAccess (Dict, optional): Default Map<TargetType, Set
> to share permissions
- Allowed target types: WORKFLOW_DEF, TASK_DEF, WORKFLOW_SCHEDULE

Example:

from conductor.client.http.models.upsert_group_request import UpsertGroupRequest

# Create group with default access
request = UpsertGroupRequest(
    description="Engineering Team",
    roles=["USER", "WORKFLOW_MANAGER"],
    defaultAccess={
        "WORKFLOW_DEF": ["READ", "EXECUTE"],
        "TASK_DEF": ["READ"]
    }
)

group = auth_client.upsert_group(request, "engineering-team")

Permission Models

GrantedPermission

Represents a granted permission showing the target and access levels.

Module: conductor.client.orkes.models.granted_permission

Properties:

target (TargetRef): The resource the permission applies to
access (List[AccessType]): The types of access granted

Example:

# Get user permissions
permissions = auth_client.get_granted_permissions_for_user("user@example.com")

for perm in permissions:
    print(f"Target: {perm.target.type}:{perm.target.id}")
    print(f"Access: {[access.name for access in perm.access]}")

AuthorizationRequest

Request model for granting or removing permissions.

Module: Internal model used by API

Properties:

subject (SubjectRef, required): The subject being granted/removed access
target (TargetRef, required): The target resource
access (List[AccessType], required): The access types to grant/remove

Example:

# This is handled internally by grant_permissions() and remove_permissions()
from conductor.client.http.models.target_ref import TargetRef, TargetType
from conductor.client.http.models.subject_ref import SubjectRef, SubjectType
from conductor.client.orkes.models.access_type import AccessType

target = TargetRef(TargetType.WORKFLOW_DEF, "my-workflow")
subject = SubjectRef(SubjectType.USER, "user@example.com")
access = [AccessType.READ, AccessType.EXECUTE]

auth_client.grant_permissions(subject, target, access)

Role Models

Role

Represents a role with associated permissions.

Properties:

name (str): Role name
permissions (List[Dict]): List of permissions
- Each permission has:
  - resource (str): Resource type (e.g., "WORKFLOW_DEF")
  - actions (List[str]): Allowed actions (e.g., ["READ", "EXECUTE"])

Example:

# Get role
role = auth_client.get_role("workflow-operator")
print(f"Role: {role['name']}")
print(f"Permissions: {role['permissions']}")

CreateOrUpdateRoleRequest

Request model for creating or updating a custom role.

Properties:

name (str, required): Role name
permissions (List[Dict], required): List of permission definitions

Example:

# Create custom role
role_request = {
    "name": "data-analyst",
    "description": "Can read and execute data workflows",
    "permissions": [
        {
            "resource": "WORKFLOW_DEF",
            "actions": ["READ", "EXECUTE"]
        },
        {
            "resource": "TASK_DEF",
            "actions": ["READ"]
        }
    ]
}

role = auth_client.create_role(role_request)

Token Models

GenerateTokenRequest

Request model for generating a JWT token.

Properties:

keyId (str, required): Access key ID
keySecret (str, required): Access key secret
expiration (int, optional): Token expiration time in seconds

Example:

# Generate JWT token
token_response = auth_client.generate_token(
    key_id="your-key-id",
    key_secret="your-key-secret"
)

jwt_token = token_response.get('token')

Gateway Models

AuthenticationConfig

Configuration for API Gateway authentication.

Module: conductor.client.http.models.authentication_config

Properties:

id (str, required): Configuration ID
applicationId (str, required): Associated application ID
authenticationType (str, required): Type of authentication - one of: NONE, API_KEY, OIDC
apiKeys (List[str]): List of API keys (when using API_KEY authentication)
audience (str): OAuth audience
conductorToken (str): Conductor token for authentication
createdBy (str): User who created the configuration
fallbackToDefaultAuth (bool): Use default auth as fallback
issuerUri (str): OAuth issuer URI (for OIDC authentication)
passthrough (bool): Whether to pass auth through
tokenInWorkflowInput (bool): Include token in workflow input
updatedBy (str): User who last updated the configuration

Example:

# Create gateway auth config with API_KEY authentication
auth_config = {
    "id": "my-gateway-auth",
    "authenticationType": "API_KEY",
    "applicationId": "app-id-123",
    "apiKeys": ["key1", "key2"],
    "fallbackToDefaultAuth": False,
    "tokenInWorkflowInput": True
}

config = auth_client.create_gateway_auth_config(auth_config)

# Create gateway auth config with OIDC authentication
oidc_config = {
    "id": "my-oidc-auth",
    "authenticationType": "OIDC",
    "applicationId": "app-id-123",
    "issuerUri": "https://auth.example.com",
    "audience": "https://api.example.com",
    "passthrough": True
}

config = auth_client.create_gateway_auth_config(oidc_config)

Model Import Reference

Quick reference for importing all models:

# Core authorization models
from conductor.client.http.models.subject_ref import SubjectRef, SubjectType
from conductor.client.http.models.target_ref import TargetRef, TargetType
from conductor.client.orkes.models.access_type import AccessType
from conductor.client.orkes.models.metadata_tag import MetadataTag
from conductor.client.orkes.models.granted_permission import GrantedPermission

# Access key models
from conductor.client.orkes.models.access_key import AccessKey
from conductor.client.orkes.models.created_access_key import CreatedAccessKey

# User and group models
from conductor.client.http.models.conductor_user import ConductorUser
from conductor.client.http.models.upsert_user_request import UpsertUserRequest
from conductor.client.http.models.group import Group
from conductor.client.http.models.upsert_group_request import UpsertGroupRequest

# Application models
from conductor.client.http.models.conductor_application import ConductorApplication
from conductor.client.http.models.create_or_update_application_request import CreateOrUpdateApplicationRequest

Complete Example: Setting Up RBAC

Here's a complete example showing how to set up RBAC for a workflow:

from conductor.client.configuration.configuration import Configuration
from conductor.client.orkes.orkes_authorization_client import OrkesAuthorizationClient
from conductor.client.http.models.upsert_user_request import UpsertUserRequest
from conductor.client.http.models.upsert_group_request import UpsertGroupRequest
from conductor.client.http.models.target_ref import TargetRef, TargetType
from conductor.client.http.models.subject_ref import SubjectRef, SubjectType
from conductor.client.orkes.models.access_type import AccessType

# Initialize
configuration = Configuration()
auth_client = OrkesAuthorizationClient(configuration)

# 1. Create users
developer = auth_client.upsert_user(
    UpsertUserRequest(name="Developer User", roles=["USER"]),
    "developer@example.com"
)

operator = auth_client.upsert_user(
    UpsertUserRequest(name="Operator User", roles=["USER"]),
    "operator@example.com"
)

# 2. Create group
engineering = auth_client.upsert_group(
    UpsertGroupRequest(description="Engineering Team", roles=["USER"]),
    "engineering-team"
)

# 3. Add users to group
auth_client.add_users_to_group("engineering-team", [
    "developer@example.com",
    "operator@example.com"
])

# 4. Grant permissions to group on workflow
workflow_target = TargetRef(TargetType.WORKFLOW_DEF, "order-processing")
group_subject = SubjectRef(SubjectType.GROUP, "engineering-team")

auth_client.grant_permissions(
    group_subject,
    workflow_target,
    [AccessType.READ, AccessType.EXECUTE]
)

# 5. Grant additional permissions to developer
developer_subject = SubjectRef(SubjectType.USER, "developer@example.com")
auth_client.grant_permissions(
    developer_subject,
    workflow_target,
    [AccessType.UPDATE]  # Developers can also modify
)

# 6. Verify permissions
permissions = auth_client.get_permissions(workflow_target)
print("Workflow permissions:")
for access_type, subjects in permissions.items():
    print(f"  {access_type}:")
    for subject in subjects:
        print(f"    - {subject.type}: {subject.id}")

# 7. Check specific user permissions
can_update = auth_client.check_permissions(
    user_id="developer@example.com",
    target_type="WORKFLOW_DEF",
    target_id="order-processing"
)
print(f"Developer can update: {can_update}")

Best Practices

Principle of Least Privilege: Grant only the minimum permissions required for users/groups to perform their tasks.
Use Groups: Assign permissions to groups rather than individual users for easier management.
Secure Access Keys:
- Store access key secrets securely (they're only shown once at creation)
- Rotate access keys regularly
- Use inactive status instead of deletion when temporarily revoking access
Audit Regularly: Use get_granted_permissions_for_user() and get_granted_permissions_for_group() to audit access.
Role-Based Organization:
- Use system roles for standard permissions
- Create custom roles for specific use cases
- Document custom role purposes
Testing: Always verify permissions with check_permissions() before granting production access.
Cleanup: Remove unused users, groups, and applications to maintain security.

Error Handling

All authorization methods may raise exceptions. Always use proper error handling:

from conductor.client.http.rest import RestException

try:
    user = auth_client.get_user("user@example.com")
except RestException as e:
    if e.status == 404:
        print("User not found")
    elif e.status == 403:
        print("Access denied")
    else:
        print(f"Error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Complete Working Example

Authorization Journey - All 49 APIs in Action

For a comprehensive implementation that demonstrates all authorization APIs in a single, cohesive narrative, see:

📚 examples/authorization_journey.py

This complete example includes:

✅ All 49 Authorization APIs - 100% coverage with proper model classes ✅ Real-World Scenario - E-commerce platform RBAC setup ✅ Progressive Learning - 12 chapters building on each other ✅ Update Operations - Demonstrates CREATE, READ, UPDATE, DELETE for all entities ✅ Custom Roles - Creating and managing custom roles with actual permissions ✅ Error Handling - Graceful fallbacks and clear error messages ✅ Cleanup - Automatic resource cleanup (can be disabled with --no-cleanup)

Running the Example

# Standard execution with automatic cleanup
python3 examples/authorization_journey.py

# Keep resources for inspection
python3 examples/authorization_journey.py --no-cleanup

# Run as pytest
python3 -m pytest examples/authorization_journey.py -v

Coverage Verification

See examples/authorization_coverage.md for detailed verification that all APIs are covered.