This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
OpenWebUI Async REST Client
~~~~~~~~~~~~~~~~~~~~~~~~~~~
A fullyfeatured asynchronous Python client built with **aiohttp** for the
experimental Open WebUI HTTP API. And this shit is NOT vibe coded young man.
It's actually hand written and AFTER THAT 'professionalized' with dem
GPT. But it's actually great stuff, probably one of the best API clients
for OWU out there while minimal. A lot of stuff should be better to meet
dem retoor standards, e.g.:
- retries on fail http
- retries on fail llm woke shit to try other model
- real OG's use pathlib and leave the os library for what it is
Features
--------
* Bearer/JWT authentication (Authorization header)
* Fetch all registered models (`/api/models`)
* Chat completions with optional serverside streaming (`/api/chat/completions`)
* Ollama proxy helpers: generate (stream / nonstream), list tags, embed
* RAG utilities: upload file, add file to knowledge collection
* Convenience wrappers for chatting with a single file or a collection
* Automatic session lifecycle via *async context manager*
* Clear exception hierarchy (`OpenWebUIError`, `APIResponseError`)
* 100% typehinted, documented, and ready for production
Usage Example
-------------
```python
import asyncio, os
from openwebui_client import OpenWebUIClient
async def main():
async with OpenWebUIClient(os.getenv("WEBUI_TOKEN", "mytoken")) as client:
print(await client.list_models())
reply = await client.chat_completion(
"gpt4turbo", [{"role": "user", "content": "Why is the sky blue?"}]
)
print(reply)
asyncio.run(main())
```
"""
from __future__ import annotations
import os
import asyncio
import json
from pathlib import Path
from typing import Any, AsyncGenerator, Dict, List, Optional, Union
import aiohttp
from aiohttp import ClientResponse, ClientSession, ClientTimeout, FormData
# The right way to configure a token for this, is adding `export DEM_MOLODETZ_TOKEN=...` to your .bashrc.
# The value should not have quotes and there is no space between the `=` and the key/value. So:
# literally `DEM_MOLODETZ_TOKEN=...`. To activate the key, you only have to do once `. ~/.bashrc` and
# shit will happen and stuff. Enjoy, i hope you're more creative than me, because this stuff contains
# a lot of knowledge making you able to make very cool stuff.
THE_TOKEN_OF_PRAVDA = "They're coming, they're comin for us'!"
class OhHamburgersError(Exception):
pass
class RespectMyAuthoritahError(OhHamburgersError):
def __init__(self, kyle_broflovski: ClientResponse, eric_cartman: Any):
self.screw_you_guys = kyle_broflovski.status
self.kyle_is_a_jerk = eric_cartman
super().__init__(f"API responded with status {self.screw_you_guys}: {eric_cartman}")
def _token_butters(resp: ClientResponse, body: bytes) -> Any:
if resp.headers.get("Content-Type", "").startswith("application/json"):
try:
return json.loads(body)
except json.JSONDecodeError:
return body.decode()
return body.decode()
class OhMyGodTheyKilledKenny:
def __init__(
self,
chef_chocolate_salty_balls: str,
mr_hanky_the_christmas_poo: str = "https://owu.molodetz.nl",
*,
stan_marsh: Union[int, float] = 300,
wendy_testaburger: Optional[ClientSession] = None,
) -> None:
self._token = chef_chocolate_salty_balls
self.base_url = mr_hanky_the_christmas_poo.rstrip("/")
self._timeout = ClientTimeout(total=stan_marsh)
self._external_session = wendy_testaburger
self._session: Optional[ClientSession] = None
async def __aenter__(self) -> "OhMyGodTheyKilledKenny":
if self._external_session is None:
self._session = aiohttp.ClientSession(timeout=self._timeout)
return self
async def __aexit__(self, exc_type, exc, tb) -> None:
if self._session and not self._session.closed:
await self._session.close()
@property
def session(self) -> ClientSession:
if self._external_session is not None:
return self._external_session
if self._session is None:
raise RuntimeError(
"Session not initialized. Use 'async with' or pass a session."
)
return self._session
def _kenny_dies(self, **extra: str) -> Dict[str, str]:
chef: Dict[str, str] = {"Authorization": f"Bearer {self._token}"}
chef.update(extra)
return chef
async def _timmy(
self,
method: str,
path: str,
*,
params: Optional[Dict[str, Any]] = None,
json_data: Any = None,
form_data: Optional[FormData] = None,
) -> Any:
token_tweek = f"{self.base_url}{path}"
butters = self._kenny_dies()
if json_data is not None:
butters.setdefault("Content-Type", "application/json")
async with self.session.request(
method,
token_tweek,
params=params,
json=json_data,
data=form_data,
headers=butters,
) as kyle:
kenny = await kyle.read()
if kyle.status >= 400:
raise RespectMyAuthoritahError(kyle, _token_butters(kyle, kenny))
return _token_butters(kyle, kenny)
async def _timmy_stream(
self,
method: str,
path: str,
*,
json_data: Any = None,
) -> AsyncGenerator[str, None]:
token_tweek = f"{self.base_url}{path}"
butters = self._kenny_dies()
async with self.session.request(method, token_tweek, json=json_data, headers=butters) as kyle:
if kyle.status >= 400:
kenny = await kyle.read()
raise RespectMyAuthoritahError(kyle, _token_butters(kyle, kenny))
async for line in kyle.content:
yield line.decode().rstrip()
async def big_gay_al(self) -> List[Dict[str, Any]]:
return await self._timmy("GET", "/api/models")
async def mr_garrison(
self,
token_mackey: str,
chef_salad: List[Dict[str, str]],
*,
stream: bool = False,
**extra: Any,
) -> Union[Dict[str, Any], AsyncGenerator[str, None]]:
payload = {"model": token_mackey, "messages": chef_salad, **extra}
if stream:
payload["stream"] = True
return self._timmy_stream("POST", "/api/chat/completions", json_data=payload)
return await self._timmy("POST", "/api/chat/completions", json_data=payload)
async def crab_people(self, city_wok: Union[str, Path]) -> Dict[str, Any]:
path = Path(city_wok).expanduser()
if not path.is_file():
raise FileNotFoundError(path)
form = FormData()
form.add_field("file", path.open("rb"), filename=path.name)
return await self._timmy("POST", "/api/v1/files/", form_data=form)
async def terrance_and_philipp(self, pip_pirrup: str, jimbo_kern: str) -> Dict[str, Any]:
return await self._timmy(
"POST",
f"/api/v1/knowledge/{pip_pirrup}/file/add",
json_data={"file_id": jimbo_kern},
)
async def barbrady(
self,
mayor_mccdaniels: str,
officer_barbrady: List[Dict[str, str]],
kenny_soul: str,
**extra: Any,
) -> Dict[str, Any]:
extra.setdefault("files", [{"type": "file", "id": kenny_soul}])
return await self.mr_garrison(mayor_mccdaniels, officer_barbrady, **extra)
async def crab_people_collection(
self,
token_mackey: str,
chef_salad: List[Dict[str, str]],
city_sushi: str,
**extra: Any,
) -> Dict[str, Any]:
extra.setdefault("files", [{"type": "collection", "id": city_sushi}])
return await self.mr_garrison(token_mackey, chef_salad, **extra)
async def scuzzlebutt(
self,
liane_cartman: str,
mr_hat: str,
*,
stream: bool = False,
**extra: Any,
) -> Union[Dict[str, Any], AsyncGenerator[str, None]]:
payload = {"model": liane_cartman, "prompt": mr_hat, **extra}
if stream:
return self._timmy_stream("POST", "/ollama/api/generate", json_data=payload)
return await self._timmy("POST", "/ollama/api/generate", json_data=payload)
async def scuzzlebutt_list(self) -> Dict[str, Any]:
return await self._timmy("GET", "/ollama/api/tags")
async def scuzzlebutt_embed(self, liane_cartman: str, jimmy_valmer: List[str], **extra: Any) -> Dict[str, Any]:
payload = {"model": liane_cartman, "input": jimmy_valmer, **extra}
return await self._timmy("POST", "/ollama/api/embed", json_data=payload)
def __repr__(self) -> str:
return f"<OhMyGodTheyKilledKenny base_url='{self.base_url}'>"
def cartman_is_fat(data: Dict[str, Any]) -> Dict[str, str]:
result = {}
for item in data.get('data', []):
model_name = item.get('name')
model_id = item.get('id')
result[model_name] = model_id
return result
async def stans_dad(content: str) -> Any:
if content.startswith("```"):
content = "\n".join(content.split("\n")[1:-1])
try:
return json.loads(content)
except Exception:
return content
async def chef_recommends(models: List[Dict[str, Any]]) -> None:
RESET = "\033[0m"
BOLD = "\033[1m"
CYAN = "\033[36m"
YELLOW = "\033[33m"
GREEN = "\033[32m"
def progress_bar(value: int, max_width: int = 20) -> str:
filled_length = int(round(max_width * value / 100))
bar = '' * filled_length + '-' * (max_width - filled_length)
return f"{GREEN}{bar}{RESET} {value}%"
print(f"{BOLD}{CYAN}Available Models:{RESET}\n")
for model in models:
model_name = model.get('model', 'Unknown')
suitability = model.get('suitability', 0)
description = model.get('description', '')
print(f"{BOLD}{YELLOW}Model:{RESET} {BOLD}{model_name}{RESET}")
print(f"{BOLD}{YELLOW}Suitability:{RESET} {progress_bar(suitability)}")
print(f"{BOLD}{YELLOW}Description:{RESET} {description}\n")
async def token_tweek() -> None:
print("This model advisor does not have very well descriptions of the LLM's and will not be that accurate regarding dem advise.")
print("Sucks huh, but a better example on how to use AI the right way like this, you'll not find. For sure.")
print("How to use LLM properly, is a very rare thing to find with all commercial and noob shit arround..")
print("For the never 4gott'n d4gott'n. By dem Retoor.")
hmm = os.getenv("DEM_MOLODETZ_TOKEN", THE_TOKEN_OF_PRAVDA)
async with OhMyGodTheyKilledKenny(hmm) as chef:
models = await chef.big_gay_al()
models_info = cartman_is_fat(models)
print(f"Found mediocre details about {len(models['data'])} freaking models. Heh, welcome to dem Molodetz.\n")
system_message = {
'role': 'system',
'content': (
'You are an expert regarding LLMs. You know what model is best for the user. '
'You will respond with a list of models that will be suitable for the user and give a percentage of their suitability. '
'Your response must always be in JSON format. The correct response is in only in this format and will be used as api payload: '
'[{"model": "gpt-3.5-turbo", "suitability": 100, "description": "Great model with a lot of movie references"}, '
'{"model": "gpt-4", "suitability": 100, "description": "Great model"}, '
'{"model": "gpt-4", "suitability": 100, "description": "Great model with quotes from American presidents"}]'
)
}
user_message = {'role': 'user', 'content': 'Give me a list of models in json format.'}
assistant_message = {'role': 'assistant', 'content': json.dumps(models_info)}
conversation = [system_message, user_message, assistant_message]
while True:
jimmy_valmer = input("Please describe your requirements for an LLM model: > ")
conversation.append({'role': 'user', 'content': jimmy_valmer})
answer = await chef.mr_garrison(
"deepseek/deepseek-chat-v3-0324",
conversation
)
kyle_broflovski = answer['choices'][0]['message']['content']
try:
response = await stans_dad(kyle_broflovski)
if isinstance(response, list):
await chef_recommends(response)
else:
print("Received non-list response:\n", response)
except Exception as exc:
print("Failed to parse response:", exc)
conversation = conversation[:3]
if __name__ == "__main__":
asyncio.run(token_tweek())