""" title: Accurate OpenRouter Cost Tracker - Uses API author: yayaduckd author_url: https://yayaduck.com git_url: https://github.com/yayaduckd description: Uses the OpenRouter API to calculate the cost of your conversations. licence: MIT """ from pydantic import BaseModel, Field from typing import Optional, Callable, Any, Awaitable import requests import time class Filter: class Valves(BaseModel): priority: int = Field( default=0, description="Priority level for the filter operations." ) openrouter_api_key: str = Field( default="", description="OpenRouter API key for cost tracking" ) pass class UserValves(BaseModel): openrouter_api_key: str = Field( default="", description="OpenRouter API key for cost tracking" ) pass def __init__(self): self.chat_id = None self.valves = self.Valves() self.costs = {} # Dictionary to track total cost per chat_id pass def get_generation_cost(self, generation_id, api_key): if not api_key: return "API key not configured", 0.0 url = "https://openrouter.ai/api/v1/generation" headers = {"Authorization": f"Bearer {api_key}"} params = {"id": generation_id} try: response = requests.get(url, headers=headers, params=params) response.raise_for_status() cost = float(response.json()["data"]["total_cost"]) # Format cost display if cost == 0: formatted_cost = "FREE!" elif cost < 0.001: formatted_cost = f"${cost:.6f} (<$0.01)" elif cost < 0.01: formatted_cost = f"${cost:.4f}" elif cost < 0.10: formatted_cost = f"${cost:.3f}" elif cost < 1.00: formatted_cost = f"${cost:.2f}" else: formatted_cost = f"${cost:.2f}" return formatted_cost, cost except requests.exceptions.RequestException as e: return f"Failed to fetch cost: {str(e)}", 0.0 except (KeyError, ValueError, TypeError): return "Invalid response format", 0.0 def format_total_cost(self, total_cost): """Format total cost consistently with generation cost formatting""" if total_cost == 0: return "FREE!" elif total_cost < 0.001: return f"${total_cost:.6f}" elif total_cost < 0.01: return f"${total_cost:.4f}" elif total_cost < 0.10: return f"${total_cost:.3f}" elif total_cost < 1.00: return f"${total_cost:.2f}" else: return f"${total_cost:.2f}" def stream(self, event: dict) -> dict: self.chat_id = f"{event['id']}" return event async def outlet( self, body: dict, __event_emitter__: Callable[[Any], Awaitable[None]], __model__: Optional[dict] = None, __user__: Optional[dict] = None, ) -> dict: await __event_emitter__( { "type": "status", "data": { "description": "Computing costs...", "done": False, }, } ) if not body.get("messages"): return body # Find the last assistant message last_assistant_idx = None for i in range(len(body["messages"]) - 1, -1, -1): if body["messages"][i]["role"] == "assistant": last_assistant_idx = i break if last_assistant_idx is None: return body time.sleep(1) # Get the chat_id from the body current_chat_id = body.get("chat_id") if not current_chat_id: current_chat_id = "default" # Initialize cost tracking for this chat if it doesn't exist if current_chat_id not in self.costs: self.costs[current_chat_id] = 0.0 # Get cost information if self.chat_id and self.valves.openrouter_api_key: cost_info, current_cost = self.get_generation_cost( self.chat_id, self.valves.openrouter_api_key ) # Add current cost to running total for this chat self.costs[current_chat_id] += current_cost total_cost_formatted = self.format_total_cost(self.costs[current_chat_id]) status_message = f"Cost: {cost_info} | Total Cost: {total_cost_formatted}" else: status_message = ( "Cost tracking unavailable (API key not set) 🔑" if not self.valves.openrouter_api_key else "Processing... ⏳" ) await __event_emitter__( { "type": "status", "data": { "description": status_message, "done": True, }, } ) return body