#!/usr/bin/env python3
"""
Looker Admin MCP - Role and permission management tools.
Provides role CRUD, permission sets, and model sets management.
"""
import logging
from typing import Dict, Any, Optional, List
from looker_sdk import models40 as models
from looker_admin_mcp.server.looker import handle_sdk_errors, get_sdk
logger = logging.getLogger('looker-admin-mcp.tools.roles')
@handle_sdk_errors
async def list_roles() -> Dict[str, Any]:
"""List all roles."""
sdk = get_sdk()
logger.info("Listing all roles")
roles = sdk.all_roles()
logger.info(f"Retrieved {len(roles)} roles")
return {
"roles": [{
"id": r.id,
"name": r.name,
"permission_set_id": r.permission_set_id,
"model_set_id": r.model_set_id,
"user_count": r.user_count,
} for r in roles],
"count": len(roles),
"status": "success"
}
@handle_sdk_errors
async def get_role(role_id: int) -> Dict[str, Any]:
"""
Get details of a specific role.
Args:
role_id: The ID of the role to retrieve
"""
sdk = get_sdk()
logger.info(f"Getting role {role_id}")
role = sdk.role(role_id)
logger.info(f"Retrieved role: {role.name}")
return {
"id": role.id,
"name": role.name,
"permission_set_id": role.permission_set_id,
"permission_set": {
"id": role.permission_set.id,
"name": role.permission_set.name,
"permissions": role.permission_set.permissions,
} if role.permission_set else None,
"model_set_id": role.model_set_id,
"model_set": {
"id": role.model_set.id,
"name": role.model_set.name,
"models": role.model_set.models,
} if role.model_set else None,
"user_count": role.user_count,
"status": "success"
}
@handle_sdk_errors
async def create_role(
name: str,
permission_set_id: int,
model_set_id: int
) -> Dict[str, Any]:
"""
Create a new role.
Args:
name: Name of the role
permission_set_id: ID of the permission set to use
model_set_id: ID of the model set to use
"""
sdk = get_sdk()
logger.info(f"Creating role: {name}")
role_body = models.WriteRole(
name=name,
permission_set_id=permission_set_id,
model_set_id=model_set_id
)
created_role = sdk.create_role(body=role_body)
logger.info(f"Created role {created_role.id}: {name}")
return {
"id": created_role.id,
"name": created_role.name,
"permission_set_id": created_role.permission_set_id,
"model_set_id": created_role.model_set_id,
"status": "success",
"message": f"Role '{name}' created successfully"
}
@handle_sdk_errors
async def update_role(
role_id: int,
name: Optional[str] = None,
permission_set_id: Optional[int] = None,
model_set_id: Optional[int] = None
) -> Dict[str, Any]:
"""
Update an existing role.
Args:
role_id: The ID of the role to update
name: New name for the role (optional)
permission_set_id: New permission set ID (optional)
model_set_id: New model set ID (optional)
"""
sdk = get_sdk()
logger.info(f"Updating role {role_id}")
update_fields = {}
if name is not None:
update_fields['name'] = name
if permission_set_id is not None:
update_fields['permission_set_id'] = permission_set_id
if model_set_id is not None:
update_fields['model_set_id'] = model_set_id
role_body = models.WriteRole(**update_fields)
updated_role = sdk.update_role(role_id, body=role_body)
logger.info(f"Updated role {role_id}")
return {
"id": updated_role.id,
"name": updated_role.name,
"permission_set_id": updated_role.permission_set_id,
"model_set_id": updated_role.model_set_id,
"status": "success",
"message": f"Role {role_id} updated successfully"
}
@handle_sdk_errors
async def delete_role(role_id: int, confirm: bool = False) -> Dict[str, Any]:
"""
Delete a role. Requires confirm=True to execute.
WARNING: This is a destructive operation that cannot be undone.
Args:
role_id: The ID of the role to delete
confirm: Must be True to execute the deletion
"""
if not confirm:
return {
"error": "Destructive operation requires confirm=True",
"warning": f"This will permanently delete role {role_id}. Set confirm=True to proceed.",
"role_id": role_id
}
sdk = get_sdk()
logger.warning(f"Deleting role {role_id} (confirmed)")
try:
role = sdk.role(role_id)
role_name = role.name
except Exception:
role_name = f"role_{role_id}"
sdk.delete_role(role_id)
logger.warning(f"Deleted role {role_id}")
return {
"status": "success",
"deleted_role_id": role_id,
"deleted_role_name": role_name,
"message": f"Role {role_id} has been permanently deleted"
}
@handle_sdk_errors
async def list_permission_sets() -> Dict[str, Any]:
"""List all permission sets."""
sdk = get_sdk()
logger.info("Listing all permission sets")
permission_sets = sdk.all_permission_sets()
logger.info(f"Retrieved {len(permission_sets)} permission sets")
return {
"permission_sets": [{
"id": ps.id,
"name": ps.name,
"built_in": ps.built_in,
"all_access": ps.all_access,
"permissions": ps.permissions,
} for ps in permission_sets],
"count": len(permission_sets),
"status": "success"
}
@handle_sdk_errors
async def get_permission_set(permission_set_id: int) -> Dict[str, Any]:
"""
Get details of a permission set.
Args:
permission_set_id: The ID of the permission set
"""
sdk = get_sdk()
logger.info(f"Getting permission set {permission_set_id}")
ps = sdk.permission_set(permission_set_id)
logger.info(f"Retrieved permission set: {ps.name}")
return {
"id": ps.id,
"name": ps.name,
"built_in": ps.built_in,
"all_access": ps.all_access,
"permissions": ps.permissions,
"status": "success"
}
@handle_sdk_errors
async def list_model_sets() -> Dict[str, Any]:
"""List all model sets."""
sdk = get_sdk()
logger.info("Listing all model sets")
model_sets = sdk.all_model_sets()
logger.info(f"Retrieved {len(model_sets)} model sets")
return {
"model_sets": [{
"id": ms.id,
"name": ms.name,
"built_in": ms.built_in,
"all_access": ms.all_access,
"models": ms.models,
} for ms in model_sets],
"count": len(model_sets),
"status": "success"
}
@handle_sdk_errors
async def get_model_set(model_set_id: int) -> Dict[str, Any]:
"""
Get details of a model set.
Args:
model_set_id: The ID of the model set
"""
sdk = get_sdk()
logger.info(f"Getting model set {model_set_id}")
ms = sdk.model_set(model_set_id)
logger.info(f"Retrieved model set: {ms.name}")
return {
"id": ms.id,
"name": ms.name,
"built_in": ms.built_in,
"all_access": ms.all_access,
"models": ms.models,
"status": "success"
}
@handle_sdk_errors
async def create_model_set(name: str, models_list: str) -> Dict[str, Any]:
"""
Create a new model set with specified LookML models.
Args:
name: Name of the model set
models_list: Comma-separated list of model names
"""
sdk = get_sdk()
logger.info(f"Creating model set: {name}")
# Parse models list
model_names = [m.strip() for m in models_list.split(',')]
model_set_body = models.WriteModelSet(
name=name,
models=model_names
)
created_ms = sdk.create_model_set(body=model_set_body)
logger.info(f"Created model set {created_ms.id}: {name}")
return {
"id": created_ms.id,
"name": created_ms.name,
"models": created_ms.models,
"status": "success",
"message": f"Model set '{name}' created successfully with {len(model_names)} models"
}
@handle_sdk_errors
async def list_role_users(role_id: int) -> Dict[str, Any]:
"""
List all users assigned to a role.
Args:
role_id: The ID of the role
"""
sdk = get_sdk()
logger.info(f"Listing users with role {role_id}")
users = sdk.role_users(role_id)
logger.info(f"Role {role_id} has {len(users)} users")
return {
"role_id": role_id,
"users": [{
"id": u.id,
"first_name": u.first_name,
"last_name": u.last_name,
"email": u.email,
"is_disabled": u.is_disabled,
} for u in users],
"count": len(users),
"status": "success"
}
