""" title: OpenRouter Credits Checker author: benasrau description: Adds a button to view OpenRouter Credits. Make sure to provide API key. The API doesn't update immediately so the value might be inaccurate. author_url: https://github.com/benasraudys funding_url: https://github.com/benasraudys version: 0.1.0 icon_url: data:image/svg+xml;base64,PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0iVVRGLTgiPz4KPHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIGlkPSJCb2xkIiB2aWV3Qm94PSIwIDAgMjQgMjQiIHdpZHRoPSI1MTIiIGhlaWdodD0iNTEyIj48cGF0aCBkPSJNMTguNSw5LjVBMS41LDEuNSwwLDAsMCwyMCw4VjcuMzEzQTUuMzIsNS4zMiwwLDAsMCwxNC42ODcsMkgxMy41VjEuNWExLjUsMS41LDAsMCwwLTMsMFYySDkuMzEzQTUuMzEzLDUuMzEzLDAsMCwwLDcuNzcyLDEyLjRsMi43MjguNzQ2VjE5SDkuMzEzQTIuMzE2LDIuMzE2LDAsMCwxLDcsMTYuNjg3VjE2YTEuNSwxLjUsMCwwLDAtMywwdi42ODdBNS4zMiw1LjMyLDAsMCwwLDkuMzEzLDIySDEwLjV2LjVhMS41LDEuNSwwLDAsMCwzLDBWMjJoMS4xODdhNS4zMTMsNS4zMTMsMCwwLDAsMS41NDEtMTAuNEwxMy41LDEwLjg1NlY1aDEuMTg3QTIuMzE2LDIuMzE2LDAsMCwxLDE3LDcuMzEzVjhBMS41LDEuNSwwLDAsMCwxOC41LDkuNVptLTMuMTE4LDQuOTc5YTIuMzE0LDIuMzE0LDAsMCwxLS43LDQuNTIxSDEzLjVWMTMuOTY1Wk0xMC41LDEwLjAzNSw4LjYxOCw5LjUyMUEyLjMxNCwyLjMxNCwwLDAsMSw5LjMxMyw1SDEwLjVaIi8+PC9zdmc+Cg== required_open_webui_version: 0.6.4 """ from pydantic import BaseModel, Field from typing import Optional import requests import asyncio class Action: class Valves(BaseModel): openrouter_api_key: str = Field( default="", description="OpenRouter API Key for checking credits" ) def __init__(self): self.valves = self.Valves() async def get_api_key_if_missing(self, __event_call__, __event_emitter__): if not self.valves.openrouter_api_key: api_key_response = await __event_call__( { "type": "input", "data": { "title": "OpenRouter API Key Needed", "message": "Please enter your OpenRouter API key:", "placeholder": "sk-...", }, } ) if api_key_response: self.valves.openrouter_api_key = api_key_response return True else: await __event_emitter__( { "type": "message", "data": { "content": "Action cancelled: No OpenRouter API key was provided." }, } ) return False return True async def action( self, body: dict, user=None, __event_emitter__=None, __event_call__=None, ) -> Optional[dict]: if not await self.get_api_key_if_missing(__event_call__, __event_emitter__): return None if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": "Fetching OpenRouter credits...", "done": False, }, } ) try: api_url = "https://openrouter.ai/api/v1/credits" headers = {"Authorization": f"Bearer {self.valves.openrouter_api_key}"} response = requests.get(api_url, headers=headers) response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx) credit_data = response.json().get("data", {}) total_credits = credit_data.get("total_credits", 0.0) total_usage = credit_data.get("total_usage", 0.0) remaining_credits = total_credits - total_usage formatted_remaining_credits = f"${remaining_credits:.8f}" if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": f"Credits remaining: {formatted_remaining_credits}", "done": True, }, } ) return { "total_credits": total_credits, "total_usage": total_usage, "remaining_credits": remaining_credits, "formatted_remaining_credits": formatted_remaining_credits, } except requests.exceptions.RequestException as e: error_message = f"API request failed: {e}" if hasattr(e, "response") and e.response is not None: error_message = ( f"Error retrieving credits. Status: {e.response.status_code}" ) if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": error_message, "done": True}, } ) return None except Exception as e: if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": f"An unexpected error occurred: {e}", "done": True, }, } ) return None