"""Users Handler - manages WordPress user operations""" import json from typing import Any from plugins.wordpress.client import WordPressClient def get_tool_specifications() -> list[dict[str, Any]]: """Return tool specifications for ToolGenerator""" return [ # === USERS === { "name": "list_users", "method_name": "list_users", "description": "List WordPress users. Returns paginated list of users with name, username, email, and roles.", "schema": { "type": "object", "properties": { "per_page": { "type": "integer", "description": "Number of users per page (1-100)", "default": 10, "minimum": 1, "maximum": 100, }, "page": { "type": "integer", "description": "Page number", "default": 1, "minimum": 1, }, "roles": { "anyOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}], "description": "Filter by user roles (e.g., ['administrator', 'editor', 'author'])", }, }, }, "scope": "read", }, { "name": "get_current_user", "method_name": "get_current_user", "description": "Get information about the currently authenticated user including ID, name, username, email, and roles.", "schema": {"type": "object", "properties": {}}, "scope": "read", }, ] class UsersHandler: """Handle user-related operations for WordPress""" def __init__(self, client: WordPressClient): """ Initialize users handler. Args: client: WordPress API client instance """ self.client = client # === USERS === async def list_users( self, per_page: int = 10, page: int = 1, roles: list[str] | None = None ) -> str: """ List WordPress users. Args: per_page: Number of users per page (1-100) page: Page number roles: Filter by user roles (e.g., ['administrator', 'editor']) Returns: JSON string with users list """ try: params = {"per_page": per_page, "page": page} if roles: params["roles"] = ",".join(roles) users = await self.client.get("users", params=params) # Format response result = { "total": len(users), "page": page, "per_page": per_page, "users": [ { "id": user["id"], "name": user["name"], "username": user["slug"], "email": user.get("email", "N/A"), "roles": user.get("roles", []), "link": user.get("link", ""), } for user in users ], } return json.dumps(result, indent=2) except Exception as e: return json.dumps( {"error": str(e), "message": f"Failed to list users: {str(e)}"}, indent=2 ) async def get_current_user(self) -> str: """ Get current authenticated user. Returns: JSON string with current user data """ try: user = await self.client.get("users/me") result = { "id": user["id"], "name": user["name"], "username": user["slug"], "email": user.get("email", "N/A"), "roles": user.get("roles", []), "capabilities": user.get("capabilities", {}), "description": user.get("description", ""), "link": user.get("link", ""), "registered_date": user.get("registered_date", ""), } return json.dumps(result, indent=2) except Exception as e: return json.dumps( {"error": str(e), "message": f"Failed to get current user: {str(e)}"}, indent=2 )