#!/usr/bin/env python3
"""
Looker Admin MCP - System administration tools.
Provides session management, user attributes, and system information.
"""
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.system')
@handle_sdk_errors
async def list_user_sessions(user_id: Optional[int] = None) -> Dict[str, Any]:
"""
List all active sessions, optionally filtered by user.
Args:
user_id: Filter by user ID (optional)
"""
sdk = get_sdk()
logger.info(f"Listing user sessions (user_id={user_id})")
if user_id:
sessions = sdk.all_user_sessions(user_id)
else:
# Get sessions for current user if no user_id specified
me = sdk.me()
sessions = sdk.all_user_sessions(me.id)
logger.info(f"Retrieved {len(sessions)} sessions")
return {
"sessions": [{
"id": s.id,
"ip_address": s.ip_address,
"browser": s.browser,
"operating_system": s.operating_system,
"city": s.city,
"state": s.state,
"country": s.country,
"credentials_type": s.credentials_type,
"created_at": s.created_at,
"expires_at": s.expires_at,
"extended_at": s.extended_at,
"extended_count": s.extended_count,
} for s in sessions],
"count": len(sessions),
"user_id": user_id,
"status": "success"
}
@handle_sdk_errors
async def delete_user_session(session_id: int, confirm: bool = False) -> Dict[str, Any]:
"""
Terminate a specific user session. Requires confirm=True to execute.
Args:
session_id: The ID of the session to terminate
confirm: Must be True to execute the deletion
"""
if not confirm:
return {
"error": "Destructive operation requires confirm=True",
"warning": f"This will terminate session {session_id}. Set confirm=True to proceed.",
"session_id": session_id
}
sdk = get_sdk()
logger.warning(f"Deleting user session {session_id} (confirmed)")
sdk.delete_user_session(session_id)
logger.warning(f"Deleted session {session_id}")
return {
"status": "success",
"deleted_session_id": session_id,
"message": f"Session {session_id} has been terminated"
}
@handle_sdk_errors
async def list_user_attributes() -> Dict[str, Any]:
"""List all user attributes defined in Looker."""
sdk = get_sdk()
logger.info("Listing all user attributes")
attributes = sdk.all_user_attributes()
logger.info(f"Retrieved {len(attributes)} user attributes")
return {
"user_attributes": [{
"id": a.id,
"name": a.name,
"label": a.label,
"type": a.type,
"default_value": a.default_value,
"is_system": a.is_system,
"is_permanent": a.is_permanent,
"value_is_hidden": a.value_is_hidden,
"user_can_view": a.user_can_view,
"user_can_edit": a.user_can_edit,
} for a in attributes],
"count": len(attributes),
"status": "success"
}
@handle_sdk_errors
async def get_user_attribute(user_attribute_id: int) -> Dict[str, Any]:
"""
Get details of a specific user attribute.
Args:
user_attribute_id: The ID of the user attribute
"""
sdk = get_sdk()
logger.info(f"Getting user attribute {user_attribute_id}")
attr = sdk.user_attribute(user_attribute_id)
logger.info(f"Retrieved user attribute: {attr.name}")
return {
"id": attr.id,
"name": attr.name,
"label": attr.label,
"type": attr.type,
"default_value": attr.default_value,
"is_system": attr.is_system,
"is_permanent": attr.is_permanent,
"value_is_hidden": attr.value_is_hidden,
"user_can_view": attr.user_can_view,
"user_can_edit": attr.user_can_edit,
"hidden_value_domain_whitelist": attr.hidden_value_domain_whitelist,
"status": "success"
}
@handle_sdk_errors
async def get_user_attribute_values(
user_id: int,
user_attribute_ids: Optional[str] = None
) -> Dict[str, Any]:
"""
Get user attribute values for a specific user.
Args:
user_id: The ID of the user
user_attribute_ids: Comma-separated list of attribute IDs to filter (optional)
"""
sdk = get_sdk()
logger.info(f"Getting attribute values for user {user_id}")
attr_id_list = None
if user_attribute_ids:
attr_id_list = [int(id.strip()) for id in user_attribute_ids.split(',')]
values = sdk.user_attribute_user_values(user_id, user_attribute_ids=attr_id_list)
logger.info(f"Retrieved {len(values)} attribute values for user {user_id}")
return {
"user_id": user_id,
"attribute_values": [{
"user_attribute_id": v.user_attribute_id,
"name": v.name,
"label": v.label,
"value": v.value,
"source": v.source,
"rank": v.rank,
} for v in values],
"count": len(values),
"status": "success"
}
@handle_sdk_errors
async def set_user_attribute_value(
user_id: int,
user_attribute_id: int,
value: str
) -> Dict[str, Any]:
"""
Set a user attribute value for a user.
Args:
user_id: The ID of the user
user_attribute_id: The ID of the user attribute
value: The value to set
"""
sdk = get_sdk()
logger.info(f"Setting attribute {user_attribute_id} for user {user_id}")
body = models.WriteUserAttributeWithValue(value=value)
result = sdk.set_user_attribute_user_value(user_id, user_attribute_id, body=body)
logger.info(f"Set attribute value for user {user_id}")
return {
"user_id": user_id,
"user_attribute_id": user_attribute_id,
"value": value,
"status": "success",
"message": f"Attribute {user_attribute_id} set to '{value}' for user {user_id}"
}
@handle_sdk_errors
async def get_api_versions() -> Dict[str, Any]:
"""Get available API versions."""
sdk = get_sdk()
logger.info("Getting API versions")
versions = sdk.versions()
logger.info("Retrieved API versions")
return {
"looker_release_version": versions.looker_release_version,
"current_version": {
"version": versions.current_version.version if versions.current_version else None,
"full_version": versions.current_version.full_version if versions.current_version else None,
} if versions.current_version else None,
"supported_versions": [{
"version": v.version,
"full_version": v.full_version,
"status": v.status,
} for v in (versions.supported_versions or [])],
"status": "success"
}
@handle_sdk_errors
async def get_looker_version() -> Dict[str, Any]:
"""Get Looker instance version information."""
sdk = get_sdk()
logger.info("Getting Looker version")
versions = sdk.versions()
logger.info(f"Looker version: {versions.looker_release_version}")
return {
"looker_release_version": versions.looker_release_version,
"api_server_url": versions.api_server_url if hasattr(versions, 'api_server_url') else None,
"web_server_url": versions.web_server_url if hasattr(versions, 'web_server_url') else None,
"status": "success"
}
