"""
title: User Info Retrieval Tool
description: Get the User, Email and ID from the user object. This tool also serves as an example for setting up an Open WebUI tool with Valves & Event Emitters.
author: @iamg30
author_url: https://openwebui.com/u/iamg30/
funding_url: https://github.com/open-webui
version: 0.2.3
license: MIT
"""
import asyncio
from typing import Callable, Awaitable, Any
from pydantic import BaseModel, Field
class Tools:
class Valves(BaseModel):
ALLOW_ID_RETRIEVAL: bool = Field(
default=True, description="Whether to allow retrieval of user IDs."
)
ALLOW_NAME_RETRIEVAL: bool = Field(
default=True, description="Whether to allow retrieval of user names."
)
ALLOW_EMAIL_RETRIEVAL: bool = Field(
default=True, description="Whether to allow retrieval of user emails."
)
ALLOW_ROLE_RETRIEVAL: bool = Field(
default=True, description="Whether to allow retrieval of user roles."
)
DISPLAY_EVENT_EMITTERS: bool = Field(
default=True,
description="Whether to display event emitters during user information retrieval.",
)
class UserValves(BaseModel):
EXCLUDE_USER_ID: bool = Field(
default=False,
description="Whether to exclude the user's ID from the retrieved information.",
)
EXCLUDE_USER_NAME: bool = Field(
default=False,
description="Whether to exclude the user's name from the retrieved information.",
)
EXCLUDE_USER_EMAIL: bool = Field(
default=False,
description="Whether to exclude the user's email from the retrieved information.",
)
EXCLUDE_USER_ROLE: bool = Field(
default=False,
description="Whether to exclude the user's role from the retrieved information.",
)
DISPLAY_EVENT_EMITTERS: bool = Field(
default=True,
description="Whether to display event emitters during user information retrieval (user-specific setting).",
)
def __init__(self):
self.valves = self.Valves()
pass
async def get_user_info(
self, __event_emitter__: Callable[[dict], Awaitable[None]], __user__: dict = {}
) -> str:
"""
Get the user ID, name, email, and role from the user object, respecting both Valves and UserValves settings.
"""
result = []
retrieved_items = []
# Get user_valves directly from __user__ dict
user_valves = __user__.get("valves")
if not user_valves:
user_valves = self.UserValves() # Use default UserValves if not provided
# Determine whether to display event emitters based on both global and user-specific settings
display_event_emitters = (
self.valves.DISPLAY_EVENT_EMITTERS and user_valves.DISPLAY_EVENT_EMITTERS
)
if display_event_emitters:
await __event_emitter__(
{
"type": "status",
"data": {
"status": "in_progress",
"description": "Starting user information retrieval",
"done": False,
},
}
)
if (
self.valves.ALLOW_ID_RETRIEVAL
and not user_valves.EXCLUDE_USER_ID
and "id" in __user__
):
result.append(f"ID: {__user__['id']}")
retrieved_items.append("ID")
if (
self.valves.ALLOW_NAME_RETRIEVAL
and not user_valves.EXCLUDE_USER_NAME
and "name" in __user__
):
result.append(f"User: {__user__['name']}")
retrieved_items.append("name")
if (
self.valves.ALLOW_EMAIL_RETRIEVAL
and not user_valves.EXCLUDE_USER_EMAIL
and "email" in __user__
):
result.append(f"Email: {__user__['email']}")
retrieved_items.append("email")
if (
self.valves.ALLOW_ROLE_RETRIEVAL
and not user_valves.EXCLUDE_USER_ROLE
and "role" in __user__
):
result.append(f"Role: {__user__['role']}")
retrieved_items.append("role")
if not result:
result = ["User: Unknown"]
retrieved_items = ["no data"]
final_result = " | ".join(result)
if display_event_emitters:
await __event_emitter__(
{
"type": "status",
"data": {
"status": "complete",
"description": f"User information retrieval completed. Retrieved: {', '.join(retrieved_items)}",
"done": True,
},
}
)
return final_result
