Google OAuth

[ineiti]

This looks really nice - do you know if it would be difficult to add google oauth to the project? We're currently running 'librechat', but it misses a lot of functionalities you added here.

[mythz]

GitHub OAuth Required a fairly large surface area, I'm assuming Google would require similar effort at least.

This markdown render doesn't seem to seem to support line number links, may need to copy/paste link:

https://github.com/ServiceStack/llms/blob/ec52d5e45b4dc5a72e6c318e41bcdeecf279874d/llms/main.py#L4194-L4421

[anilgulecha]

More generally - implementing OIDC protocol as a provider would give automatic coverage of any others (including Google and others).

[ineiti]

Why didn't you use something like https://docs.authlib.org/en/latest/client/oauth2.html, which would be able to handle both cases?

If I were to create a PR and use authlib for both github and google, would you consider accepting it?

@anilgulecha - it seems that github does not provide authentication via OIDC...

[mythz]

Why didn't you use something like https://docs.authlib.org/en/latest/client/oauth2.html

Because I want to avoid additional deps in the main.py default install and I only wanted to use GitHub which could be easily implemented without a library.

If I were to create a PR and use authlib for both github and google, would you consider accepting it?

But yeah happy to switch over to a OIDC library & impl, but prefer it implemented as an optional extension so the default install could be used without needing to sign in or install additional deps. Happy to add any extension points / generic session handling in main.py to support an external extension implementation.

Let me know if you're interested in taking a shot at it and I can create the repo in llmspy and add you as a contributor. Also let me know what repo name you prefer to use, it should just be different to any libraries used to avoid conflicts.

Development

I'd recommend adding the impl into a llmspy repo so it can be easily installed with just llms --add oauth. For easy of development I just add a symlink to the local repo to the ~/.llms/extensions/oauth folder that way you can just add/remove the symlink to test without it.

To run a local dev version I use ./llms.sh script in the main folder which can be substituted for llms. I run the Server + UI with both debug (ctx.dbg()) and verbose (ctx.log()) logging on with:

DEBUG=1 ./llms.sh --serve 8000 --verbose

I've documented the server extensions API but as always consult the source code for the most uptodate APIs, it's just the AppExtensions and ExtensionContext classes, the server extension hooks receive an instance of ExtensionContext in its ctx arg. ctx.app lets you access the singleton AppExtensions instance shared by main.py and all extensions.

llms/llms/main.py

Lines 2777 to 3394 in 05cc325

	class AppExtensions:
	"""
	APIs extensions can use to extend the app
	"""
	def __init__(self, cli_args: argparse.Namespace, extra_args: Dict[str, Any]):
	self.cli_args = cli_args
	self.extra_args = extra_args
	self.config = None
	self.error_auth_required = create_error_response("Authentication required", "Unauthorized")
	self.ui_extensions = []
	self.chat_request_filters = []
	self.extensions = []
	self.loaded = False
	self.chat_tool_filters = []
	self.chat_response_filters = []
	self.chat_error_filters = []
	self.server_add_get = []
	self.server_add_post = []
	self.server_add_put = []
	self.server_add_delete = []
	self.server_add_patch = []
	self.cache_saved_filters = []
	self.shutdown_handlers = []
	self.tools = {}
	self.tool_definitions = []
	self.tool_groups = {}
	self.index_headers = []
	self.index_footers = []
	self.allowed_directories = []
	self.request_args = {
	"image_config": dict, # e.g. { "aspect_ratio": "1:1" }
	"temperature": float, # e.g: 0.7
	"max_completion_tokens": int, # e.g: 2048
	"seed": int, # e.g: 42
	"top_p": float, # e.g: 0.9
	"frequency_penalty": float, # e.g: 0.5
	"presence_penalty": float, # e.g: 0.5
	"stop": list, # e.g: ["Stop"]
	"reasoning_effort": str, # e.g: minimal, low, medium, high
	"verbosity": str, # e.g: low, medium, high
	"service_tier": str, # e.g: auto, default
	"top_logprobs": int,
	"safety_identifier": str,
	"store": bool,
	"enable_thinking": bool,
	}
	self.all_providers = [
	OpenAiCompatible,
	MistralProvider,
	GroqProvider,
	XaiProvider,
	CodestralProvider,
	OllamaProvider,
	LMStudioProvider,
	]
	self.aspect_ratios = {
	"1:1": "1024×1024",
	"2:3": "832×1248",
	"3:2": "1248×832",
	"3:4": "864×1184",
	"4:3": "1184×864",
	"4:5": "896×1152",
	"5:4": "1152×896",
	"9:16": "768×1344",
	"16:9": "1344×768",
	"21:9": "1536×672",
	}
	self.import_maps = {
	"vue-prod": "/ui/lib/vue.min.mjs",
	"vue": "/ui/lib/vue.mjs",
	"vue-router": "/ui/lib/vue-router.min.mjs",
	"@servicestack/client": "/ui/lib/servicestack-client.mjs",
	"@servicestack/vue": "/ui/lib/servicestack-vue.mjs",
	"idb": "/ui/lib/idb.min.mjs",
	"marked": "/ui/lib/marked.min.mjs",
	"highlight.js": "/ui/lib/highlight.min.mjs",
	"chart.js": "/ui/lib/chart.js",
	"color.js": "/ui/lib/color.js",
	"ctx.mjs": "/ui/ctx.mjs",
	}
	def set_config(self, config: Dict[str, Any]):
	self.config = config
	self.auth_enabled = self.config.get("auth", {}).get("enabled", False)
	def set_allowed_directories(
	self, directories: List[Annotated[str, "List of absolute paths that are allowed to be accessed."]]
	) -> None:
	"""Set the list of allowed directories."""
	self.allowed_directories = [os.path.abspath(d) for d in directories]
	def add_allowed_directory(self, path: str) -> None:
	"""Add an allowed directory."""
	abs_path = os.path.abspath(path)
	if abs_path not in self.allowed_directories:
	self.allowed_directories.append(abs_path)
	def get_allowed_directories(self) -> List[str]:
	"""Get the list of allowed directories."""
	return self.allowed_directories
	# Authentication middleware helper
	def check_auth(self, request: web.Request) -> Tuple[bool, Optional[Dict[str, Any]]]:
	"""Check if request is authenticated. Returns (is_authenticated, user_data)"""
	if not self.auth_enabled:
	return True, None
	# Check for OAuth session token
	session_token = get_session_token(request)
	if session_token and session_token in g_sessions:
	return True, g_sessions[session_token]
	# Check for API key
	auth_header = request.headers.get("Authorization", "")
	if auth_header.startswith("Bearer "):
	api_key = auth_header[7:]
	if api_key:
	return True, {"authProvider": "apikey"}
	return False, None
	def get_session(self, request: web.Request) -> Optional[Dict[str, Any]]:
	session_token = get_session_token(request)
	if not session_token or session_token not in g_sessions:
	return None
	session_data = g_sessions[session_token]
	return session_data
	def get_username(self, request: web.Request) -> Optional[str]:
	session = self.get_session(request)
	if session:
	return session.get("userName")
	return None
	def get_user_path(self, username: Optional[str] = None) -> str:
	if username:
	return home_llms_path(os.path.join("user", username))
	return home_llms_path(os.path.join("user", "default"))
	def get_providers(self) -> Dict[str, Any]:
	return g_handlers
	def chat_request(
	self,
	template: Optional[str] = None,
	text: Optional[str] = None,
	model: Optional[str] = None,
	system_prompt: Optional[str] = None,
	) -> Dict[str, Any]:
	return g_chat_request(template=template, text=text, model=model, system_prompt=system_prompt)
	async def chat_completion(self, chat: Dict[str, Any], context: Optional[Dict[str, Any]] = None) -> Any:
	response = await g_chat_completion(chat, context)
	return response
	def on_cache_saved_filters(self, context: Dict[str, Any]):
	# _log(f"on_cache_saved_filters {len(self.cache_saved_filters)}: {context['url']}")
	for filter_func in self.cache_saved_filters:
	filter_func(context)
	async def on_chat_error(self, e: Exception, context: Dict[str, Any]):
	# Apply chat error filters
	if "stackTrace" not in context:
	context["stackTrace"] = traceback.format_exc()
	for filter_func in self.chat_error_filters:
	try:
	task = filter_func(e, context)
	if asyncio.isfuture(task):
	await task
	except Exception as e:
	_err("chat error filter failed", e)
	async def on_chat_tool(self, chat: Dict[str, Any], context: Dict[str, Any]):
	m_len = len(chat.get("messages", []))
	t_len = len(self.chat_tool_filters)
	_dbg(
	f"on_tool_call for thread {context.get('threadId')} with {m_len} {pluralize('message', m_len)}, invoking {t_len} {pluralize('filter', t_len)}:"
	)
	for filter_func in self.chat_tool_filters:
	await filter_func(chat, context)
	def shutdown(self):
	if len(self.shutdown_handlers) > 0:
	_dbg(f"running {len(self.shutdown_handlers)} shutdown handlers...")
	for handler in self.shutdown_handlers:
	handler()
	def exit(self, exit_code: int = 0):
	self.shutdown()
	_dbg(f"exit({exit_code})")
	sys.exit(exit_code)
	def create_chat_with_tools(self, chat: Dict[str, Any], use_tools: str = "all") -> Dict[str, Any]:
	# Inject global tools if present
	current_chat = chat.copy()
	tools = current_chat.get("tools")
	if tools is None:
	tools = current_chat["tools"] = []
	if self.tool_definitions and len(tools) == 0:
	include_all_tools = use_tools == "all"
	only_tools_list = use_tools.split(",")
	if include_all_tools or len(only_tools_list) > 0:
	if "tools" not in current_chat:
	current_chat["tools"] = []
	_dbg(
	f"create_chat_with_tools: all_tools:{include_all_tools}, only_tools:{only_tools_list}, chat tools: "
	+ str(len(current_chat["tools"]))
	)
	existing_tools = {t["function"]["name"] for t in current_chat["tools"]}
	for tool_def in self.tool_definitions:
	name = tool_def["function"]["name"]
	if name not in existing_tools and (include_all_tools or name in only_tools_list):
	current_chat["tools"].append(tool_def)
	return current_chat
	def get_tool_definition(self, name: str) -> Optional[Dict[str, Any]]:
	for tool_def in self.tool_definitions:
	if tool_def["function"]["name"] == name:
	return tool_def
	return None
	def handler_name(handler):
	if hasattr(handler, "__name__"):
	return handler.__name__
	return "unknown"
	class ExtensionContext:
	def __init__(self, app: AppExtensions, path: str):
	self.app = app
	self.cli_args = app.cli_args
	self.extra_args = app.extra_args
	self.error_auth_required = app.error_auth_required
	self.path = path
	self.name = os.path.basename(path)
	if self.name.endswith(".py"):
	self.name = self.name[:-3]
	self.ext_prefix = f"/ext/{self.name}"
	self.MOCK = MOCK
	self.MOCK_DIR = MOCK_DIR
	self.debug = DEBUG
	self.verbose = g_verbose
	self.aspect_ratios = app.aspect_ratios
	self.request_args = app.request_args
	self.disabled = False
	def set_allowed_directories(
	self, directories: List[Annotated[str, "List of absolute paths that are allowed to be accessed."]]
	) -> None:
	"""Set the list of allowed directories."""
	self.app.set_allowed_directories(directories)
	def add_allowed_directory(self, path: str) -> None:
	"""Add an allowed directory."""
	self.app.add_allowed_directory(path)
	def get_allowed_directories(self) -> List[str]:
	"""Get the list of allowed directories."""
	return self.app.get_allowed_directories()
	def chat_to_prompt(self, chat: Dict[str, Any]) -> str:
	return chat_to_prompt(chat)
	def chat_to_system_prompt(self, chat: Dict[str, Any]) -> str:
	return chat_to_system_prompt(chat)
	def chat_response_to_message(self, response: Dict[str, Any]) -> Dict[str, Any]:
	return chat_response_to_message(response)
	def last_user_prompt(self, chat: Dict[str, Any]) -> str:
	return last_user_prompt(chat)
	def to_file_info(
	self, chat: Dict[str, Any], info: Optional[Dict[str, Any]] = None, response: Optional[Dict[str, Any]] = None
	) -> Dict[str, Any]:
	return to_file_info(chat, info=info, response=response)
	def save_image_to_cache(
	self, base64_data: Union[str, bytes], filename: str, image_info: Dict[str, Any], ignore_info: bool = False
	) -> Tuple[str, Optional[Dict[str, Any]]]:
	return save_image_to_cache(base64_data, filename, image_info, ignore_info=ignore_info)
	def save_bytes_to_cache(
	self, bytes_data: Union[str, bytes], filename: str, file_info: Optional[Dict[str, Any]]
	) -> Tuple[str, Optional[Dict[str, Any]]]:
	return save_bytes_to_cache(bytes_data, filename, file_info)
	def text_from_file(self, path: str) -> str:
	return text_from_file(path)
	def json_from_file(self, path: str) -> Any:
	return json_from_file(path)
	def download_file(self, url: str) -> Tuple[bytes, Dict[str, Any]]:
	return download_file(url)
	def session_download_file(self, session: aiohttp.ClientSession, url: str) -> Tuple[bytes, Dict[str, Any]]:
	return session_download_file(session, url)
	def read_binary_file(self, url: str) -> Tuple[bytes, Dict[str, Any]]:
	return read_binary_file(url)
	def log(self, message: Any):
	if self.verbose:
	print(f"[{self.name}] {message}", flush=True)
	return message
	def log_json(self, obj: Any):
	if self.verbose:
	print(f"[{self.name}] {json.dumps(obj, indent=2)}", flush=True)
	return obj
	def dbg(self, message: Any):
	if self.debug:
	print(f"DEBUG [{self.name}]: {message}", flush=True)
	def err(self, message: str, e: Exception):
	print(f"ERROR [{self.name}]: {message}", e)
	if self.verbose:
	print(traceback.format_exc(), flush=True)
	def error_message(self, e: Exception) -> str:
	return to_error_message(e)
	def error_response(self, e: Exception, stacktrace: bool = False) -> Dict[str, Any]:
	return to_error_response(e, stacktrace=stacktrace)
	def add_provider(self, provider: Any):
	self.log(f"Registered provider: {provider.__name__}")
	self.app.all_providers.append(provider)
	def register_ui_extension(self, index: str):
	path = os.path.join(self.ext_prefix, index)
	self.log(f"Registered UI extension: {path}")
	self.app.ui_extensions.append({"id": self.name, "path": path})
	def register_chat_request_filter(self, handler: Callable):
	self.log(f"Registered chat request filter: {handler_name(handler)}")
	self.app.chat_request_filters.append(handler)
	def register_chat_tool_filter(self, handler: Callable):
	self.log(f"Registered chat tool filter: {handler_name(handler)}")
	self.app.chat_tool_filters.append(handler)
	def register_chat_response_filter(self, handler: Callable):
	self.log(f"Registered chat response filter: {handler_name(handler)}")
	self.app.chat_response_filters.append(handler)
	def register_chat_error_filter(self, handler: Callable):
	self.log(f"Registered chat error filter: {handler_name(handler)}")
	self.app.chat_error_filters.append(handler)
	def register_cache_saved_filter(self, handler: Callable):
	self.log(f"Registered cache saved filter: {handler_name(handler)}")
	self.app.cache_saved_filters.append(handler)
	def register_shutdown_handler(self, handler: Callable):
	self.log(f"Registered shutdown handler: {handler_name(handler)}")
	self.app.shutdown_handlers.append(handler)
	def add_static_files(self, ext_dir: str):
	self.log(f"Registered static files: {ext_dir}")
	async def serve_static(request):
	path = request.match_info["path"]
	file_path = os.path.join(ext_dir, path)
	if os.path.exists(file_path):
	return web.FileResponse(file_path)
	return web.Response(status=404)
	self.app.server_add_get.append((os.path.join(self.ext_prefix, "{path:.*}"), serve_static, {}))
	def web_path(self, method: str, path: str) -> str:
	full_path = os.path.join(self.ext_prefix, path) if path else self.ext_prefix
	self.dbg(f"Registered {method:<6} {full_path}")
	return full_path
	def add_get(self, path: str, handler: Callable, **kwargs: Any):
	self.app.server_add_get.append((self.web_path("GET", path), handler, kwargs))
	def add_post(self, path: str, handler: Callable, **kwargs: Any):
	self.app.server_add_post.append((self.web_path("POST", path), handler, kwargs))
	def add_put(self, path: str, handler: Callable, **kwargs: Any):
	self.app.server_add_put.append((self.web_path("PUT", path), handler, kwargs))
	def add_delete(self, path: str, handler: Callable, **kwargs: Any):
	self.app.server_add_delete.append((self.web_path("DELETE", path), handler, kwargs))
	def add_patch(self, path: str, handler: Callable, **kwargs: Any):
	self.app.server_add_patch.append((self.web_path("PATCH", path), handler, kwargs))
	def add_importmaps(self, dict: Dict[str, str]):
	self.app.import_maps.update(dict)
	def add_index_header(self, html: str):
	self.app.index_headers.append(html)
	def add_index_footer(self, html: str):
	self.app.index_footers.append(html)
	def get_home_path(self, name: str = "") -> str:
	return home_llms_path(name)
	def get_config(self) -> Optional[Dict[str, Any]]:
	return g_config
	def get_cache_path(self, path: str = "") -> str:
	return get_cache_path(path)
	def get_file_mime_type(self, filename: str) -> str:
	return get_file_mime_type(filename)
	def chat_request(
	self,
	template: Optional[str] = None,
	text: Optional[str] = None,
	model: Optional[str] = None,
	system_prompt: Optional[str] = None,
	) -> Dict[str, Any]:
	return self.app.chat_request(template=template, text=text, model=model, system_prompt=system_prompt)
	async def chat_completion(self, chat: Dict[str, Any], context: Optional[Dict[str, Any]] = None) -> Any:
	return await self.app.chat_completion(chat, context=context)
	def get_providers(self) -> Dict[str, Any]:
	return g_handlers
	def get_provider(self, name: str) -> Optional[Any]:
	return g_handlers.get(name)
	def sanitize_tool_def(self, tool_def: Dict[str, Any]) -> Dict[str, Any]:
	"""
	Merge $defs parameter into tool_def property to reduce client/server complexity
	"""
	# parameters = {
	# "$defs": {
	# "AspectRatio": {
	# "description": "Supported aspect ratios for image generation.",
	# "enum": [
	# "1:1",
	# "2:3",
	# "16:9"
	# ],
	# "type": "string"
	# }
	# },
	# "properties": {
	# "prompt": {
	# "type": "string"
	# },
	# "model": {
	# "default": "gemini-2.5-flash-image",
	# "type": "string"
	# },
	# "aspect_ratio": {
	# "$ref": "#/$defs/AspectRatio",
	# "default": "1:1"
	# }
	# },
	# "required": [
	# "prompt"
	# ],
	# "type": "object"
	# }
	type = tool_def.get("type")
	if type == "function":
	func_def = tool_def.get("function", {})
	parameters = func_def.get("parameters", {})
	defs = parameters.get("$defs", {})
	properties = parameters.get("properties", {})
	for _, prop_def in properties.items():
	if "$ref" in prop_def:
	ref = prop_def["$ref"]
	if ref.startswith("#/$defs/"):
	def_name = ref.replace("#/$defs/", "")
	if def_name in defs:
	prop_def.update(defs[def_name])
	del prop_def["$ref"]
	if "$defs" in parameters:
	del parameters["$defs"]
	return tool_def
	def register_tool(self, func: Callable, tool_def: Optional[Dict[str, Any]] = None, group: Optional[str] = None):
	if tool_def is None:
	tool_def = function_to_tool_definition(func)
	name = tool_def["function"]["name"]
	if name in self.app.tools:
	self.log(f"Overriding existing tool: {name}")
	self.app.tool_definitions = [t for t in self.app.tool_definitions if t["function"]["name"] != name]
	for g_tools in self.app.tool_groups.values():
	if name in g_tools:
	g_tools.remove(name)
	else:
	self.log(f"Registered tool: {name}")
	self.app.tools[name] = func
	self.app.tool_definitions.append(self.sanitize_tool_def(tool_def))
	if not group:
	group = "custom"
	if group not in self.app.tool_groups:
	self.app.tool_groups[group] = []
	self.app.tool_groups[group].append(name)
	def get_tool_definition(self, name: str) -> Optional[Dict[str, Any]]:
	return self.app.get_tool_definition(name)
	def group_resources(self, resources: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]:
	return group_resources(resources)
	def check_auth(self, request: web.Request) -> Tuple[bool, Optional[Dict[str, Any]]]:
	return self.app.check_auth(request)
	def get_session(self, request: web.Request) -> Optional[Dict[str, Any]]:
	return self.app.get_session(request)
	def get_username(self, request: web.Request) -> Optional[str]:
	return self.app.get_username(request)
	def get_user_path(self, username: Optional[str] = None) -> str:
	return self.app.get_user_path(username)
	def context_to_username(self, context: Optional[Dict[str, Any]]) -> Optional[str]:
	if context and "request" in context:
	return self.get_username(context["request"])
	return None
	def should_cancel_thread(self, context: Dict[str, Any]) -> bool:
	return should_cancel_thread(context)
	def cache_message_inline_data(self, message: Dict[str, Any]):
	return cache_message_inline_data(message)
	async def exec_tool(self, name: str, args: Dict[str, Any]) -> Tuple[Optional[str], List[Dict[str, Any]]]:
	return await g_exec_tool(name, args)
	def tool_result(
	self, result: Any, function_name: Optional[str] = None, function_args: Optional[Dict[str, Any]] = None
	) -> Dict[str, Any]:
	return g_tool_result(result, function_name, function_args)
	def tool_result_part(
	self,
	result: Dict[str, Any],
	function_name: Optional[str] = None,
	function_args: Optional[Dict[str, Any]] = None,
	) -> Dict[str, Any]:
	return tool_result_part(result, function_name, function_args)
	def to_content(self, result: Any) -> str:
	return to_content(result)
	def create_chat_with_tools(self, chat: Dict[str, Any], use_tools: str = "all") -> Dict[str, Any]:
	return self.app.create_chat_with_tools(chat, use_tools)
	def chat_to_aspect_ratio(self, chat: Dict[str, Any]) -> str:
	return chat_to_aspect_ratio(chat)
	def get_extensions_path():
	return os.getenv("LLMS_EXTENSIONS_DIR", home_llms_path("extensions"))
	def get_disabled_extensions():
	ret = DISABLE_EXTENSIONS.copy()
	if g_config:
	for ext in g_config.get("disable_extensions", []):
	if ext not in ret:
	ret.append(ext)
	return ret
	def get_extensions_dirs():
	"""
	Returns a list of extension directories.
	"""
	extensions_path = get_extensions_path()
	os.makedirs(extensions_path, exist_ok=True)
	# allow overriding builtin extensions
	override_extensions = []
	if os.path.exists(extensions_path):
	override_extensions = os.listdir(extensions_path)
	ret = []
	disabled_extensions = get_disabled_extensions()
	builtin_extensions_dir = _ROOT / "extensions"
	if not os.path.exists(builtin_extensions_dir):
	# look for local ./extensions dir from script
	builtin_extensions_dir = os.path.join(os.path.dirname(__file__), "extensions")
	_dbg(f"Loading extensions from {builtin_extensions_dir}")
	if os.path.exists(builtin_extensions_dir):
	for item in os.listdir(builtin_extensions_dir):
	if os.path.isdir(os.path.join(builtin_extensions_dir, item)):
	if item in override_extensions:
	continue
	if item in disabled_extensions:
	continue
	ret.append(os.path.join(builtin_extensions_dir, item))
	if os.path.exists(extensions_path):
	for item in os.listdir(extensions_path):
	if os.path.isdir(os.path.join(extensions_path, item)):
	if item in disabled_extensions:
	continue
	ret.append(os.path.join(extensions_path, item))
	return ret

The built-in extensions and llmspy repositories will show how each extension implements its functionality.

Also note that the backing SQLite tables only use a single user column to specify the user which stores the GitHub username when auth is enabled or null for anon users when it isn't.

An Extension configuration and data is recommended to be stored at ~/.llms/user/default/<ext>

I'll look at pulling the github auth into an external extension as well to avoid multiple implementations and to ensure the extensions APIs are able to support implementing an external auth provider.

[ineiti]

Cool - thanks a lot for the pointers! Now I have week-end project #32 on my list :)

[anilgulecha]

@mythz BTW, kudos on the choice of holding to the simplicity in the main app, and moving these to extensions. It's a strong design vision, and will make the project very strong in the long run.