"""Auth Handler - manages Supabase authentication via GoTrue Admin API"""
import json
from typing import Any
from plugins.supabase.client import SupabaseClient
def get_tool_specifications() -> list[dict[str, Any]]:
"""Return tool specifications for ToolGenerator (14 tools)"""
return [
{
"name": "list_users",
"method_name": "list_users",
"description": "List all users with pagination. Returns user details including email, phone, metadata, and confirmation status.",
"schema": {
"type": "object",
"properties": {
"page": {
"type": "integer",
"description": "Page number (1-indexed)",
"default": 1,
},
"per_page": {
"type": "integer",
"description": "Users per page (max 1000)",
"default": 50,
},
},
},
"scope": "admin",
},
{
"name": "get_user",
"method_name": "get_user",
"description": "Get detailed information about a specific user by their ID.",
"schema": {
"type": "object",
"properties": {"user_id": {"type": "string", "description": "User UUID"}},
"required": ["user_id"],
},
"scope": "read",
},
{
"name": "create_user",
"method_name": "create_user",
"description": "Create a new user with email and password. Can optionally auto-confirm email and set metadata.",
"schema": {
"type": "object",
"properties": {
"email": {
"type": "string",
"format": "email",
"description": "User email address",
},
"password": {
"type": "string",
"minLength": 6,
"description": "Password (min 6 characters)",
},
"phone": {
"anyOf": [{"type": "string"}, {"type": "null"}],
"description": "Phone number in E.164 format",
},
"email_confirm": {
"type": "boolean",
"description": "Auto-confirm email without verification",
"default": False,
},
"user_metadata": {
"anyOf": [{"type": "object"}, {"type": "null"}],
"description": "Custom user metadata (name, avatar, etc.)",
},
"app_metadata": {
"anyOf": [{"type": "object"}, {"type": "null"}],
"description": "App-specific metadata (roles, permissions)",
},
},
"required": ["email", "password"],
},
"scope": "admin",
},
{
"name": "update_user",
"method_name": "update_user",
"description": "Update user details including email, password, phone, metadata, or ban status.",
"schema": {
"type": "object",
"properties": {
"user_id": {"type": "string", "description": "User UUID"},
"email": {
"anyOf": [{"type": "string"}, {"type": "null"}],
"description": "New email address",
},
"password": {
"anyOf": [{"type": "string"}, {"type": "null"}],
"description": "New password",
},
"phone": {
"anyOf": [{"type": "string"}, {"type": "null"}],
"description": "New phone number",
},
"email_confirm": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Set email confirmation status",
},
"phone_confirm": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Set phone confirmation status",
},
"user_metadata": {
"anyOf": [{"type": "object"}, {"type": "null"}],
"description": "Update user metadata",
},
"app_metadata": {
"anyOf": [{"type": "object"}, {"type": "null"}],
"description": "Update app metadata",
},
},
"required": ["user_id"],
},
"scope": "admin",
},
{
"name": "delete_user",
"method_name": "delete_user",
"description": "Permanently delete a user and all their data.",
"schema": {
"type": "object",
"properties": {"user_id": {"type": "string", "description": "User UUID to delete"}},
"required": ["user_id"],
},
"scope": "admin",
},
{
"name": "invite_user",
"method_name": "invite_user",
"description": "Send an email invitation to a new user. Creates user in 'invited' state.",
"schema": {
"type": "object",
"properties": {
"email": {
"type": "string",
"format": "email",
"description": "Email to invite",
},
"redirect_to": {
"anyOf": [{"type": "string"}, {"type": "null"}],
"description": "URL to redirect after accepting invitation",
},
"user_metadata": {
"anyOf": [{"type": "object"}, {"type": "null"}],
"description": "Initial user metadata",
},
},
"required": ["email"],
},
"scope": "admin",
},
{
"name": "generate_link",
"method_name": "generate_link",
"description": "Generate a magic link, recovery link, or invite link for a user.",
"schema": {
"type": "object",
"properties": {
"email": {"type": "string", "format": "email", "description": "User email"},
"link_type": {
"type": "string",
"enum": [
"magiclink",
"recovery",
"invite",
"signup",
"email_change_new",
"email_change_current",
],
"description": "Type of link to generate",
"default": "magiclink",
},
"redirect_to": {
"anyOf": [{"type": "string"}, {"type": "null"}],
"description": "URL to redirect after link is used",
},
},
"required": ["email"],
},
"scope": "admin",
},
{
"name": "ban_user",
"method_name": "ban_user",
"description": "Ban a user for a specified duration. Banned users cannot sign in.",
"schema": {
"type": "object",
"properties": {
"user_id": {"type": "string", "description": "User UUID"},
"duration": {
"type": "string",
"description": "Ban duration (e.g., '24h', '7d', '1y', 'none' for permanent)",
"default": "none",
},
},
"required": ["user_id"],
},
"scope": "admin",
},
{
"name": "unban_user",
"method_name": "unban_user",
"description": "Remove ban from a user, allowing them to sign in again.",
"schema": {
"type": "object",
"properties": {"user_id": {"type": "string", "description": "User UUID"}},
"required": ["user_id"],
},
"scope": "admin",
},
{
"name": "list_user_factors",
"method_name": "list_user_factors",
"description": "List all MFA factors (TOTP, phone) for a user.",
"schema": {
"type": "object",
"properties": {"user_id": {"type": "string", "description": "User UUID"}},
"required": ["user_id"],
},
"scope": "read",
},
{
"name": "delete_user_factor",
"method_name": "delete_user_factor",
"description": "Delete an MFA factor from a user.",
"schema": {
"type": "object",
"properties": {
"user_id": {"type": "string", "description": "User UUID"},
"factor_id": {"type": "string", "description": "Factor UUID to delete"},
},
"required": ["user_id", "factor_id"],
},
"scope": "admin",
},
{
"name": "get_auth_config",
"method_name": "get_auth_config",
"description": "Get current GoTrue authentication configuration.",
"schema": {"type": "object", "properties": {}},
"scope": "read",
},
{
"name": "search_users",
"method_name": "search_users",
"description": "Search users by email or phone number.",
"schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query (email or phone)"},
"page": {"type": "integer", "default": 1},
"per_page": {"type": "integer", "default": 50},
},
"required": ["query"],
},
"scope": "read",
},
{
"name": "get_user_by_email",
"method_name": "get_user_by_email",
"description": "Find a user by their email address.",
"schema": {
"type": "object",
"properties": {
"email": {
"type": "string",
"format": "email",
"description": "Email to search for",
}
},
"required": ["email"],
},
"scope": "read",
},
]
# =====================
# Auth Operations (14 tools)
# =====================
async def list_users(client: SupabaseClient, page: int = 1, per_page: int = 50) -> str:
"""List all users with pagination"""
try:
result = await client.list_users(page=page, per_page=per_page)
users = result.get("users", []) if isinstance(result, dict) else result
result.get("aud", len(users)) if isinstance(result, dict) else len(users)
return json.dumps(
{
"success": True,
"page": page,
"per_page": per_page,
"count": len(users) if isinstance(users, list) else 0,
"users": users,
},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def get_user(client: SupabaseClient, user_id: str) -> str:
"""Get user by ID"""
try:
result = await client.get_user(user_id)
return json.dumps({"success": True, "user": result}, indent=2, ensure_ascii=False)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def create_user(
client: SupabaseClient,
email: str,
password: str,
phone: str | None = None,
email_confirm: bool = False,
user_metadata: dict | None = None,
app_metadata: dict | None = None,
) -> str:
"""Create a new user"""
try:
result = await client.create_user(
email=email,
password=password,
phone=phone,
email_confirm=email_confirm,
user_metadata=user_metadata,
app_metadata=app_metadata,
)
return json.dumps(
{"success": True, "message": "User created successfully", "user": result},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def update_user(
client: SupabaseClient,
user_id: str,
email: str | None = None,
password: str | None = None,
phone: str | None = None,
email_confirm: bool | None = None,
phone_confirm: bool | None = None,
user_metadata: dict | None = None,
app_metadata: dict | None = None,
) -> str:
"""Update user details"""
try:
result = await client.update_user(
user_id=user_id,
email=email,
password=password,
phone=phone,
email_confirm=email_confirm,
phone_confirm=phone_confirm,
user_metadata=user_metadata,
app_metadata=app_metadata,
)
return json.dumps(
{"success": True, "message": "User updated successfully", "user": result},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def delete_user(client: SupabaseClient, user_id: str) -> str:
"""Delete a user"""
try:
await client.delete_user(user_id)
return json.dumps(
{"success": True, "message": f"User {user_id} deleted successfully"},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def invite_user(
client: SupabaseClient,
email: str,
redirect_to: str | None = None,
user_metadata: dict | None = None,
) -> str:
"""Invite a user by email"""
try:
result = await client.generate_link(
email=email, link_type="invite", redirect_to=redirect_to
)
return json.dumps(
{"success": True, "message": f"Invitation sent to {email}", "result": result},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def generate_link(
client: SupabaseClient, email: str, link_type: str = "magiclink", redirect_to: str | None = None
) -> str:
"""Generate auth link (magic link, recovery, invite)"""
try:
result = await client.generate_link(
email=email, link_type=link_type, redirect_to=redirect_to
)
return json.dumps(
{"success": True, "link_type": link_type, "email": email, "result": result},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def ban_user(client: SupabaseClient, user_id: str, duration: str = "none") -> str:
"""Ban a user"""
try:
result = await client.update_user(user_id=user_id, ban_duration=duration)
return json.dumps(
{"success": True, "message": f"User {user_id} banned for {duration}", "user": result},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def unban_user(client: SupabaseClient, user_id: str) -> str:
"""Unban a user"""
try:
result = await client.update_user(user_id=user_id, ban_duration="0")
return json.dumps(
{"success": True, "message": f"User {user_id} unbanned", "user": result},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def list_user_factors(client: SupabaseClient, user_id: str) -> str:
"""List user MFA factors"""
try:
result = await client.list_user_factors(user_id)
return json.dumps(
{"success": True, "user_id": user_id, "factors": result}, indent=2, ensure_ascii=False
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def delete_user_factor(client: SupabaseClient, user_id: str, factor_id: str) -> str:
"""Delete an MFA factor"""
try:
await client.delete_user_factor(user_id, factor_id)
return json.dumps(
{"success": True, "message": f"Factor {factor_id} deleted from user {user_id}"},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def get_auth_config(client: SupabaseClient) -> str:
"""Get auth configuration"""
try:
# Get health which includes some config info
result = await client.request("GET", "/auth/v1/health", use_service_role=False)
return json.dumps({"success": True, "config": result}, indent=2, ensure_ascii=False)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def search_users(
client: SupabaseClient, query: str, page: int = 1, per_page: int = 50
) -> str:
"""Search users by email or phone"""
try:
# Get all users and filter
result = await client.list_users(page=page, per_page=per_page)
users = result.get("users", []) if isinstance(result, dict) else result
# Filter by query
query_lower = query.lower()
filtered = [
u
for u in users
if (
u.get("email", "").lower().find(query_lower) >= 0
or u.get("phone", "").find(query) >= 0
)
]
return json.dumps(
{"success": True, "query": query, "count": len(filtered), "users": filtered},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
async def get_user_by_email(client: SupabaseClient, email: str) -> str:
"""Find user by email"""
try:
# Get all users and find by email
result = await client.list_users(page=1, per_page=1000)
users = result.get("users", []) if isinstance(result, dict) else result
# Find exact match
user = next((u for u in users if u.get("email", "").lower() == email.lower()), None)
if user:
return json.dumps(
{"success": True, "found": True, "user": user}, indent=2, ensure_ascii=False
)
else:
return json.dumps(
{"success": True, "found": False, "message": f"No user found with email: {email}"},
indent=2,
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"success": False, "error": str(e)}, indent=2, ensure_ascii=False)
