#!/usr/bin/env python3
"""
Looker Admin MCP - Group management tools.
Provides group CRUD operations and membership management.
"""
import logging
from typing import Dict, Any, Optional
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.groups')
@handle_sdk_errors
async def list_groups(
page: Optional[int] = None,
per_page: Optional[int] = 50,
sorts: Optional[str] = None,
ids: Optional[str] = None
) -> Dict[str, Any]:
"""
List all groups with optional pagination.
Args:
page: Page number to return (1-indexed)
per_page: Number of results per page (default 50)
sorts: Field to sort by (e.g., 'name', '-id')
ids: Comma-separated list of group IDs to filter
"""
sdk = get_sdk()
logger.info(f"Listing groups (page={page}, per_page={per_page})")
id_list = None
if ids:
id_list = [int(id.strip()) for id in ids.split(',')]
groups = sdk.all_groups(
page=page,
per_page=per_page,
sorts=sorts,
ids=id_list
)
logger.info(f"Retrieved {len(groups)} groups")
return {
"groups": [{
"id": g.id,
"name": g.name,
"user_count": g.user_count,
"externally_managed": g.externally_managed,
"external_group_id": g.external_group_id,
} for g in groups],
"count": len(groups),
"status": "success"
}
@handle_sdk_errors
async def search_groups(
name: Optional[str] = None,
external_group_id: Optional[str] = None,
externally_managed: Optional[bool] = None,
page: Optional[int] = None,
per_page: Optional[int] = 50
) -> Dict[str, Any]:
"""
Search groups by name or external ID.
Args:
name: Filter by group name (partial match)
external_group_id: Filter by external group ID
externally_managed: Filter by externally managed status
page: Page number
per_page: Results per page
"""
sdk = get_sdk()
logger.info(f"Searching groups (name={name})")
groups = sdk.search_groups(
name=name,
external_group_id=external_group_id,
externally_managed=externally_managed,
page=page,
per_page=per_page
)
logger.info(f"Found {len(groups)} matching groups")
return {
"groups": [{
"id": g.id,
"name": g.name,
"user_count": g.user_count,
"externally_managed": g.externally_managed,
} for g in groups],
"count": len(groups),
"status": "success"
}
@handle_sdk_errors
async def get_group(group_id: int) -> Dict[str, Any]:
"""
Get details of a specific group.
Args:
group_id: The ID of the group to retrieve
"""
sdk = get_sdk()
logger.info(f"Getting group {group_id}")
group = sdk.group(group_id)
logger.info(f"Retrieved group: {group.name}")
return {
"id": group.id,
"name": group.name,
"user_count": group.user_count,
"externally_managed": group.externally_managed,
"external_group_id": group.external_group_id,
"include_by_default": group.include_by_default,
"contains_current_user": group.contains_current_user,
"status": "success"
}
@handle_sdk_errors
async def create_group(name: str) -> Dict[str, Any]:
"""
Create a new group.
Args:
name: Name of the group to create
"""
sdk = get_sdk()
logger.info(f"Creating group: {name}")
group_body = models.WriteGroup(name=name)
created_group = sdk.create_group(body=group_body)
logger.info(f"Created group {created_group.id}: {name}")
return {
"id": created_group.id,
"name": created_group.name,
"status": "success",
"message": f"Group '{name}' created successfully"
}
@handle_sdk_errors
async def update_group(group_id: int, name: str) -> Dict[str, Any]:
"""
Update a group's name.
Args:
group_id: The ID of the group to update
name: New name for the group
"""
sdk = get_sdk()
logger.info(f"Updating group {group_id} to name: {name}")
group_body = models.WriteGroup(name=name)
updated_group = sdk.update_group(group_id, body=group_body)
logger.info(f"Updated group {group_id}")
return {
"id": updated_group.id,
"name": updated_group.name,
"status": "success",
"message": f"Group {group_id} renamed to '{name}'"
}
@handle_sdk_errors
async def delete_group(group_id: int, confirm: bool = False) -> Dict[str, Any]:
"""
Delete a group. Requires confirm=True to execute.
WARNING: This is a destructive operation that cannot be undone.
Args:
group_id: The ID of the group 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 group {group_id}. Set confirm=True to proceed.",
"group_id": group_id
}
sdk = get_sdk()
logger.warning(f"Deleting group {group_id} (confirmed)")
# Get group info before deletion
try:
group = sdk.group(group_id)
group_name = group.name
except Exception:
group_name = f"group_{group_id}"
sdk.delete_group(group_id)
logger.warning(f"Deleted group {group_id}")
return {
"status": "success",
"deleted_group_id": group_id,
"deleted_group_name": group_name,
"message": f"Group {group_id} has been permanently deleted"
}
@handle_sdk_errors
async def list_group_users(
group_id: int,
page: Optional[int] = None,
per_page: Optional[int] = 50,
sorts: Optional[str] = None
) -> Dict[str, Any]:
"""
List all users in a group.
Args:
group_id: The ID of the group
page: Page number
per_page: Results per page
sorts: Field to sort by
"""
sdk = get_sdk()
logger.info(f"Listing users in group {group_id}")
users = sdk.all_group_users(
group_id,
page=page,
per_page=per_page,
sorts=sorts
)
logger.info(f"Group {group_id} has {len(users)} users")
return {
"group_id": group_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"
}
@handle_sdk_errors
async def add_user_to_group(group_id: int, user_id: int) -> Dict[str, Any]:
"""
Add a user to a group.
Args:
group_id: The ID of the group
user_id: The ID of the user to add
"""
sdk = get_sdk()
logger.info(f"Adding user {user_id} to group {group_id}")
body = models.GroupIdForGroupUserInclusion(user_id=user_id)
result = sdk.add_group_user(group_id, body=body)
logger.info(f"Added user {user_id} to group {group_id}")
return {
"group_id": group_id,
"user_id": user_id,
"status": "success",
"message": f"User {user_id} added to group {group_id}"
}
@handle_sdk_errors
async def remove_user_from_group(group_id: int, user_id: int) -> Dict[str, Any]:
"""
Remove a user from a group.
Args:
group_id: The ID of the group
user_id: The ID of the user to remove
"""
sdk = get_sdk()
logger.info(f"Removing user {user_id} from group {group_id}")
sdk.delete_group_user(group_id, user_id)
logger.info(f"Removed user {user_id} from group {group_id}")
return {
"group_id": group_id,
"user_id": user_id,
"status": "success",
"message": f"User {user_id} removed from group {group_id}"
}
