laantungir/routstr-core
1
0
Fork 0
mirror of https://github.com/Routstr/routstr-core.git synced 2026-08-09 02:54:37 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
4bde69e15ea27432f8bd44cb778a89cd97a4e045
routstr-core/routstr/upstream/base.py
T

3211 lines
119 KiB
Python
Raw Blame History

from __future__ import annotations
import asyncio
import json
import re
import traceback
from collections.abc import AsyncGenerator
from typing import Mapping
import httpx
from fastapi import BackgroundTasks, HTTPException, Request
from fastapi.responses import Response, StreamingResponse
from pydantic import BaseModel
from sqlmodel import select
from ..auth import adjust_payment_for_tokens, revert_pay_for_request
from ..core import get_logger
from ..core.db import ApiKey, AsyncSession, UpstreamProviderRow, create_session
from ..core.exceptions import UpstreamError
from ..payment.cost_calculation import (
CostData,
CostDataError,
MaxCostData,
calculate_cost,
)
from ..payment.helpers import create_error_response
from ..payment.models import (
Model,
Pricing,
_calculate_usd_max_costs,
_update_model_sats_pricing,
list_models,
)
from ..payment.price import sats_usd_price
from ..wallet import recieve_token, send_token
logger = get_logger(__name__)
class TopupData(BaseModel):
"""Universal top-up data schema for Lightning Network invoices."""
invoice_id: str
payment_request: str
amount: int
currency: str
expires_at: int | None = None
checkout_url: str | None = None
class BaseUpstreamProvider:
"""Provider for forwarding requests to an upstream AI service API."""
provider_type: str = "base"
default_base_url: str | None = None
platform_url: str | None = None
base_url: str
api_key: str
provider_fee: float = 1.05
_models_cache: list[Model] = []
_models_by_id: dict[str, Model] = {}
def __init__(self, base_url: str, api_key: str, provider_fee: float = 1.01):
"""Initialize the upstream provider.
Args:
base_url: Base URL of the upstream API endpoint
api_key: API key for authenticating with the upstream service
provider_fee: Provider fee multiplier (default 1.01 for 1% fee)
"""
self.base_url = base_url
self.api_key = api_key
self.provider_fee = provider_fee
self._models_cache = []
self._models_by_id = {}
@classmethod
def from_db_row(
cls, provider_row: "UpstreamProviderRow"
) -> "BaseUpstreamProvider | None":
"""Factory method to instantiate provider from database row.
Args:
provider_row: Database row containing provider configuration
Returns:
Instantiated provider or None if instantiation fails
"""
return cls(
base_url=provider_row.base_url,
api_key=provider_row.api_key,
provider_fee=provider_row.provider_fee,
)
@classmethod
def get_provider_metadata(cls) -> dict[str, object]:
"""Get metadata about this provider type for API responses.
Returns:
Dict with provider type metadata including id, name, default_base_url, fixed_base_url, platform_url, can_create_account, can_topup, can_show_balance
"""
return {
"id": cls.provider_type,
"name": cls.provider_type.title(),
"default_base_url": cls.default_base_url or "",
"fixed_base_url": bool(cls.default_base_url),
"platform_url": cls.platform_url,
"can_create_account": False,
"can_topup": False,
"can_show_balance": False,
}
def prepare_headers(self, request_headers: dict) -> dict:
"""Prepare headers for upstream request by removing proxy-specific headers and adding authentication.
Args:
request_headers: Original request headers from the client
Returns:
Headers dict ready for upstream forwarding with authentication added
"""
logger.debug(
"Preparing upstream headers",
extra={
"original_headers_count": len(request_headers),
"has_upstream_api_key": bool(self.api_key),
},
)
headers = dict(request_headers)
removed_headers = []
for header in [
"host",
"content-length",
"refund-lnurl",
"key-expiry-time",
"x-cashu",
]:
if headers.pop(header, None) is not None:
removed_headers.append(header)
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
if headers.pop("authorization", None) is not None:
removed_headers.append("authorization (replaced with upstream key)")
else:
for auth_header in ["Authorization", "authorization"]:
if headers.pop(auth_header, None) is not None:
removed_headers.append(auth_header)
for header in ["authorization", "accept-encoding"]:
if headers.pop(header, None) is not None:
removed_headers.append(f"{header} (replaced with routstr-safe version)")
# Explicitly define the list of supported compression encodings
headers["accept-encoding"] = "gzip, deflate, br, identity"
logger.debug(
"Headers prepared for upstream",
extra={
"final_headers_count": len(headers),
"removed_headers": removed_headers,
"added_upstream_auth": bool(self.api_key),
},
)
return headers
def prepare_params(
self, path: str, query_params: Mapping[str, str] | None
) -> Mapping[str, str]:
"""Prepare query parameters for upstream request.
Base implementation passes through query params unchanged. Override in subclasses for provider-specific params.
Args:
path: Request path
query_params: Original query parameters from the client
Returns:
Query parameters dict ready for upstream forwarding
"""
return query_params or {}
def transform_model_name(self, model_id: str) -> str:
"""Transform model ID for this provider's API format.
Base implementation returns model_id unchanged. Override in subclasses for provider-specific transformations.
Args:
model_id: Model identifier (may include provider prefix)
Returns:
Transformed model ID for this provider
"""
return model_id
def prepare_responses_request_body(
self, body: bytes | None, model_obj: Model
) -> bytes | None:
"""Transform request body for Responses API specific requirements.
Handles Responses API specific transformations while maintaining model name transforms.
Args:
body: Original request body bytes
model_obj: Model object containing the original model information
Returns:
Transformed request body bytes
"""
if not body:
return body
try:
data = json.loads(body)
if isinstance(data, dict):
# Handle model transformation in various locations
if "model" in data:
original_model = model_obj.id
transformed_model = self.transform_model_name(original_model)
data["model"] = transformed_model
logger.debug(
"Transformed model name in Responses API request",
extra={
"original": original_model,
"transformed": transformed_model,
"provider": self.provider_type or self.base_url,
},
)
# Handle model in input field (alternative format)
if (
"input" in data
and isinstance(data["input"], dict)
and "model" in data["input"]
):
original_model = model_obj.id
transformed_model = self.transform_model_name(original_model)
data["input"]["model"] = transformed_model
# Ensure proper Responses API structure
# Add any Responses-specific transformations here
return json.dumps(data).encode()
except Exception as e:
logger.debug(
"Could not transform Responses API request body",
extra={
"error": str(e),
"provider": self.provider_type or self.base_url,
},
)
return body
def prepare_request_body(
self, body: bytes | None, model_obj: Model
) -> bytes | None:
"""Transform request body for provider-specific requirements.
Automatically transforms model names in the request body.
Args:
body: Original request body bytes
Returns:
Transformed request body bytes
"""
if not body:
return body
try:
data = json.loads(body)
if isinstance(data, dict) and "model" in data:
original_model = model_obj.id
transformed_model = self.transform_model_name(original_model)
data["model"] = transformed_model
logger.debug(
"Transformed model name in request",
extra={
"original": original_model,
"transformed": transformed_model,
"provider": self.provider_type or self.base_url,
},
)
return json.dumps(data).encode()
except Exception as e:
logger.debug(
"Could not transform request body",
extra={
"error": str(e),
"provider": self.provider_type or self.base_url,
},
)
return body
def _extract_upstream_error_message(
self, body_bytes: bytes
) -> tuple[str, str | None]:
"""Extract error message and code from upstream error response body.
Args:
body_bytes: Raw response body bytes from upstream
Returns:
Tuple of (error_message, error_code), where error_code may be None
"""
message: str = "Upstream request failed"
upstream_code: str | None = None
if not body_bytes:
return message, upstream_code
try:
data = json.loads(body_bytes)
if isinstance(data, dict):
err = data.get("error")
if isinstance(err, dict):
raw_msg = (
err.get("message") or err.get("detail") or err.get("error")
)
if isinstance(raw_msg, (str, int, float)):
message = str(raw_msg)
upstream_code_raw = err.get("code") or err.get("type")
if isinstance(upstream_code_raw, (str, int, float)):
upstream_code = str(upstream_code_raw)
elif "message" in data and isinstance(
data["message"], (str, int, float)
):
message = str(data["message"]) # type: ignore[arg-type]
elif "detail" in data and isinstance(data["detail"], (str, int, float)):
message = str(data["detail"]) # type: ignore[arg-type]
except Exception:
preview = body_bytes.decode("utf-8", errors="ignore").strip()
if preview:
message = preview[:500]
return message, upstream_code
async def on_upstream_error_redirect(
self, status_code: int, error_message: str
) -> None:
"""Hook called when the proxy redirects to another provider due to an error.
Subclasses can implement this to perform actions like disabling the provider
if it's out of balance.
Args:
status_code: The HTTP status code returned by the upstream
error_message: The error message extracted from the upstream response
"""
pass
async def map_upstream_error_response(
self, request: Request, path: str, upstream_response: httpx.Response
) -> Response:
"""Map upstream error responses to appropriate proxy error responses.
Args:
request: Original FastAPI request
path: Request path
upstream_response: Response from upstream service
Returns:
Mapped error response with appropriate status code and error type
"""
status_code = upstream_response.status_code
headers = dict(upstream_response.headers)
content_type = headers.get("content-type", "")
try:
body_bytes = await upstream_response.aread()
except Exception:
body_bytes = b""
message, upstream_code = self._extract_upstream_error_message(body_bytes)
lowered_message = message.lower()
lowered_code = (upstream_code or "").lower()
error_type = "upstream_error"
mapped_status = 502
if status_code in (400, 422):
error_type = "invalid_request_error"
mapped_status = 400
elif status_code in (401, 403):
error_type = "upstream_auth_error"
mapped_status = 502
elif status_code == 404:
if path.endswith("chat/completions"):
error_type = "invalid_model"
mapped_status = 400
if not message or message == "Upstream request failed":
message = "Requested model is not available upstream"
elif "model" in lowered_message or "model" in lowered_code:
error_type = "invalid_model"
mapped_status = 400
if not message or message == "Upstream request failed":
message = "Requested model is not available upstream"
else:
error_type = "upstream_error"
mapped_status = 502
elif status_code == 429:
error_type = "rate_limit_exceeded"
mapped_status = 429
elif status_code >= 500:
error_type = "upstream_error"
mapped_status = 502
logger.debug(
"Mapped upstream error",
extra={
"path": path,
"upstream_status": status_code,
"mapped_status": mapped_status,
"error_type": error_type,
"upstream_content_type": content_type,
"message_preview": message[:200],
},
)
return create_error_response(
error_type, message, mapped_status, request=request
)
async def handle_streaming_chat_completion(
self,
response: httpx.Response,
key: ApiKey,
max_cost_for_model: int,
background_tasks: BackgroundTasks,
) -> StreamingResponse:
"""Handle streaming chat completion responses with token usage tracking and cost adjustment.
Args:
response: Streaming response from upstream
key: API key for the authenticated user
max_cost_for_model: Maximum cost deducted upfront for the model
Returns:
StreamingResponse with cost data injected at the end
"""
logger.info(
"Processing streaming chat completion",
extra={
"key_hash": key.hashed_key[:8] + "...",
"key_balance": key.balance,
"response_status": response.status_code,
},
)
async def stream_with_cost(
max_cost_for_model: int,
) -> AsyncGenerator[bytes, None]:
usage_finalized: bool = False
last_model_seen: str | None = None
usage_chunk_data: dict | None = None
done_seen: bool = False
async def finalize_db_only() -> None:
nonlocal usage_finalized
if usage_finalized:
return
async with create_session() as new_session:
fresh_key = await new_session.get(key.__class__, key.hashed_key)
if not fresh_key:
return
try:
await adjust_payment_for_tokens(
fresh_key,
{"model": last_model_seen or "unknown", "usage": None},
new_session,
max_cost_for_model,
)
usage_finalized = True
except Exception:
pass
try:
async for chunk in response.aiter_bytes():
# Split chunk into SSE events
parts = re.split(b"data: ", chunk)
for i, part in enumerate(parts):
if not part:
continue
stripped_part = part.strip()
if not stripped_part:
continue
if stripped_part == b"[DONE]":
done_seen = True
continue
try:
obj = json.loads(part)
if isinstance(obj, dict):
if obj.get("model"):
last_model_seen = str(obj.get("model"))
if isinstance(obj.get("usage"), dict):
# Hold this chunk back to merge cost later
usage_chunk_data = obj
continue
except json.JSONDecodeError:
pass
prefix = (
b"data: " if (i > 0 or chunk.startswith(b"data: ")) else b""
)
yield prefix + part
# Stream finished, process usage if found
if usage_chunk_data:
async with create_session() as session:
fresh_key = await session.get(key.__class__, key.hashed_key)
if fresh_key:
try:
cost_data = await adjust_payment_for_tokens(
fresh_key,
usage_chunk_data,
session,
max_cost_for_model,
)
remaining_balance_msats = fresh_key.balance
# Merge cost into usage
usage_chunk_data["usage"]["cost"] = cost_data.get(
"total_usd", 0.0
)
usage_chunk_data["usage"]["cost_sats"] = (
cost_data.get("total_msats", 0) // 1000
)
usage_chunk_data["usage"]["remaining_balance_msats"] = (
remaining_balance_msats
)
# Keep detailed cost in metadata
usage_chunk_data["metadata"] = usage_chunk_data.get(
"metadata", {}
)
usage_chunk_data["metadata"]["routstr"] = {
"cost": cost_data
}
usage_chunk_data["metadata"]["routstr"]["cost"][
"sats_cost"
] = cost_data.get("total_msats", 0) // 1000
usage_chunk_data["metadata"]["routstr"]["cost"][
"remaining_balance_msats"
] = remaining_balance_msats
yield f"data: {json.dumps(usage_chunk_data)}\n\n".encode()
usage_finalized = True
except Exception as e:
logger.exception(
"Error during usage finalization",
extra={
"key_hash": key.hashed_key[:8] + "...",
"error": str(e),
},
)
# Fallback: yield original usage chunk if adjustment fails
yield f"data: {json.dumps(usage_chunk_data)}\n\n".encode()
if not usage_finalized:
await finalize_db_only()
if done_seen:
yield b"data: [DONE]\n\n"
except Exception as stream_error:
logger.warning(
"Streaming interrupted; finalizing in background",
extra={
"error": str(stream_error),
"key_hash": key.hashed_key[:8] + "...",
},
)
raise
finally:
if not usage_finalized:
# Create a background task to ensure finalization happens
# even if the generator is closed early
background_tasks.add_task(finalize_db_only)
# Remove inaccurate encoding headers from upstream response
response_headers = dict(response.headers)
response_headers.pop("content-encoding", None)
response_headers.pop("content-length", None)
return StreamingResponse(
stream_with_cost(max_cost_for_model),
status_code=response.status_code,
headers=response_headers,
)
async def handle_non_streaming_chat_completion(
self,
response: httpx.Response,
key: ApiKey,
session: AsyncSession,
deducted_max_cost: int,
) -> Response:
"""Handle non-streaming chat completion responses with token usage tracking and cost adjustment.
Args:
response: Response from upstream
key: API key for the authenticated user
session: Database session for updating balance
deducted_max_cost: Maximum cost deducted upfront
Returns:
Response with cost data added to JSON body
"""
logger.info(
"Processing non-streaming chat completion",
extra={
"key_hash": key.hashed_key[:8] + "...",
"key_balance": key.balance,
"response_status": response.status_code,
},
)
content: bytes | None = None
try:
content = await response.aread()
response_json = json.loads(content)
logger.debug(
"Parsed response JSON",
extra={
"key_hash": key.hashed_key[:8] + "...",
"model": response_json.get("model", "unknown"),
"has_usage": "usage" in response_json,
},
)
cost_data = await adjust_payment_for_tokens(
key, response_json, session, deducted_max_cost
)
await session.refresh(key)
remaining_balance_msats = key.balance
# Merge cost into usage for OpenCode
if "usage" in response_json:
response_json["usage"]["cost"] = cost_data.get("total_usd", 0.0)
response_json["usage"]["cost_sats"] = (
cost_data.get("total_msats", 0) // 1000
)
response_json["usage"]["remaining_balance_msats"] = (
remaining_balance_msats
)
# Keep detailed cost
response_json["metadata"] = response_json.get("metadata", {})
response_json["metadata"]["routstr"] = {"cost": cost_data}
response_json["metadata"]["routstr"]["cost"]["sats_cost"] = (
cost_data.get("total_msats", 0) // 1000
)
response_json["metadata"]["routstr"]["cost"]["remaining_balance_msats"] = (
remaining_balance_msats
)
response_json["cost"] = cost_data
response_json["cost"]["sats_cost"] = cost_data.get("total_msats", 0) // 1000
response_json["cost"]["remaining_balance_msats"] = remaining_balance_msats
logger.info(
"Payment adjustment completed for non-streaming",
extra={
"key_hash": key.hashed_key[:8] + "...",
"cost_data": cost_data,
"model": response_json.get("model", "unknown"),
"balance_after_adjustment": key.balance,
},
)
allowed_headers = {
"content-type",
"cache-control",
"date",
"vary",
"access-control-allow-origin",
"access-control-allow-methods",
"access-control-allow-headers",
"access-control-allow-credentials",
"access-control-expose-headers",
"access-control-max-age",
}
response_headers = {
k: v
for k, v in response.headers.items()
if k.lower() in allowed_headers
}
return Response(
content=json.dumps(response_json).encode(),
status_code=response.status_code,
headers=response_headers,
media_type="application/json",
)
except json.JSONDecodeError as e:
logger.error(
"Failed to parse JSON from upstream response",
extra={
"error": str(e),
"key_hash": key.hashed_key[:8] + "...",
"content_preview": content[:200].decode(errors="ignore")
if content
else "empty",
},
)
raise
except Exception as e:
logger.error(
"Error processing non-streaming chat completion",
extra={
"error": str(e),
"error_type": type(e).__name__,
"key_hash": key.hashed_key[:8] + "...",
},
)
raise
async def handle_streaming_responses_completion(
self, response: httpx.Response, key: ApiKey, max_cost_for_model: int
) -> StreamingResponse:
"""Handle streaming Responses API responses with token usage tracking and cost adjustment.
Args:
response: Streaming response from upstream
key: API key for the authenticated user
max_cost_for_model: Maximum cost deducted upfront for the model
Returns:
StreamingResponse with cost data injected at the end
"""
logger.info(
"Processing streaming Responses API completion",
extra={
"key_hash": key.hashed_key[:8] + "...",
"key_balance": key.balance,
"response_status": response.status_code,
},
)
async def stream_with_responses_cost(
max_cost_for_model: int,
) -> AsyncGenerator[bytes, None]:
usage_finalized: bool = False
last_model_seen: str | None = None
reasoning_tokens: int = 0
usage_chunk_data: dict | None = None
done_seen: bool = False
async def finalize_db_only() -> None:
nonlocal usage_finalized
if usage_finalized:
return
async with create_session() as new_session:
fresh_key = await new_session.get(key.__class__, key.hashed_key)
if not fresh_key:
return
try:
await adjust_payment_for_tokens(
fresh_key,
{"model": last_model_seen or "unknown", "usage": None},
new_session,
max_cost_for_model,
)
usage_finalized = True
except Exception:
pass
try:
async for chunk in response.aiter_bytes():
# Split chunk into SSE events
parts = re.split(b"data: ", chunk)
for i, part in enumerate(parts):
if not part:
continue
stripped_part = part.strip()
if not stripped_part:
continue
if stripped_part == b"[DONE]":
done_seen = True
continue
try:
obj = json.loads(part)
if isinstance(obj, dict):
if obj.get("model"):
last_model_seen = str(obj.get("model"))
# Track reasoning tokens for Responses API
if usage := obj.get("usage", {}):
if (
isinstance(usage, dict)
and "reasoning_tokens" in usage
):
reasoning_tokens += usage.get(
"reasoning_tokens", 0
)
# Responses API usage is in response.completed/incomplete events
chunk_type = obj.get("type", "")
if chunk_type in (
"response.completed",
"response.incomplete",
):
usage_chunk_data = obj
continue
except json.JSONDecodeError:
pass
prefix = (
b"data: " if (i > 0 or chunk.startswith(b"data: ")) else b""
)
yield prefix + part
# Stream finished, process usage if found
if usage_chunk_data:
async with create_session() as session:
fresh_key = await session.get(key.__class__, key.hashed_key)
if fresh_key:
try:
cost_data = await adjust_payment_for_tokens(
fresh_key,
usage_chunk_data,
session,
max_cost_for_model,
)
remaining_balance_msats = fresh_key.balance
# Merge cost into usage chunk
if (
"response" in usage_chunk_data
and "usage" in usage_chunk_data["response"]
):
usage_chunk_data["response"]["usage"]["cost"] = (
cost_data.get("total_usd", 0.0)
)
usage_chunk_data["response"]["usage"][
"cost_sats"
] = cost_data.get("total_msats", 0) // 1000
usage_chunk_data["response"]["usage"][
"remaining_balance_msats"
] = remaining_balance_msats
elif "usage" in usage_chunk_data:
usage_chunk_data["usage"]["cost"] = cost_data.get(
"total_usd", 0.0
)
usage_chunk_data["usage"]["cost_sats"] = (
cost_data.get("total_msats", 0) // 1000
)
usage_chunk_data["usage"][
"remaining_balance_msats"
] = remaining_balance_msats
# Keep detailed cost in metadata
usage_chunk_data["metadata"] = usage_chunk_data.get(
"metadata", {}
)
usage_chunk_data["metadata"]["routstr"] = {
"cost": cost_data
}
usage_chunk_data["metadata"]["routstr"]["cost"][
"sats_cost"
] = cost_data.get("total_msats", 0) // 1000
usage_chunk_data["metadata"]["routstr"]["cost"][
"remaining_balance_msats"
] = remaining_balance_msats
yield f"data: {json.dumps(usage_chunk_data)}\n\n".encode()
usage_finalized = True
except Exception:
# Fallback: yield original usage chunk if adjustment fails
yield f"data: {json.dumps(usage_chunk_data)}\n\n".encode()
if not usage_finalized:
await finalize_db_only()
if done_seen:
yield b"data: [DONE]\n\n"
except Exception as stream_error:
logger.warning(
"Responses API streaming interrupted; finalizing in background",
extra={
"error": str(stream_error),
"key_hash": key.hashed_key[:8] + "...",
},
)
raise
finally:
if not usage_finalized:
await finalize_db_only()
# Remove inaccurate encoding headers from upstream response
response_headers = dict(response.headers)
response_headers.pop("content-encoding", None)
response_headers.pop("content-length", None)
return StreamingResponse(
stream_with_responses_cost(max_cost_for_model),
status_code=response.status_code,
headers=response_headers,
)
async def handle_non_streaming_responses_completion(
self,
response: httpx.Response,
key: ApiKey,
session: AsyncSession,
deducted_max_cost: int,
) -> Response:
"""Handle non-streaming Responses API responses with token usage tracking and cost adjustment.
Args:
response: Response from upstream
key: API key for the authenticated user
session: Database session for updating balance
deducted_max_cost: Maximum cost deducted upfront
Returns:
Response with cost data added to JSON body
"""
logger.info(
"Processing non-streaming Responses API completion",
extra={
"key_hash": key.hashed_key[:8] + "...",
"key_balance": key.balance,
"response_status": response.status_code,
},
)
content: bytes | None = None
try:
content = await response.aread()
response_json = json.loads(content)
logger.debug(
"Parsed Responses API response JSON",
extra={
"key_hash": key.hashed_key[:8] + "...",
"model": response_json.get("model", "unknown"),
"has_usage": "usage" in response_json,
"has_reasoning_tokens": "usage" in response_json
and isinstance(response_json.get("usage"), dict)
and "reasoning_tokens" in response_json["usage"],
},
)
cost_data = await adjust_payment_for_tokens(
key, response_json, session, deducted_max_cost
)
await session.refresh(key)
remaining_balance_msats = key.balance
# Merge cost into usage for OpenCode
if "usage" in response_json:
response_json["usage"]["cost"] = cost_data.get("total_usd", 0.0)
response_json["usage"]["cost_sats"] = (
cost_data.get("total_msats", 0) // 1000
)
response_json["usage"]["remaining_balance_msats"] = (
remaining_balance_msats
)
# Keep detailed cost
response_json["metadata"] = response_json.get("metadata", {})
response_json["metadata"]["routstr"] = {"cost": cost_data}
response_json["metadata"]["routstr"]["cost"]["sats_cost"] = (
cost_data.get("total_msats", 0) // 1000
)
response_json["metadata"]["routstr"]["cost"]["remaining_balance_msats"] = (
remaining_balance_msats
)
response_json["cost"] = cost_data
response_json["cost"]["sats_cost"] = cost_data.get("total_msats", 0) // 1000
response_json["cost"]["remaining_balance_msats"] = remaining_balance_msats
logger.info(
"Payment adjustment completed for non-streaming Responses API",
extra={
"key_hash": key.hashed_key[:8] + "...",
"cost_data": cost_data,
"model": response_json.get("model", "unknown"),
"balance_after_adjustment": key.balance,
},
)
allowed_headers = {
"content-type",
"cache-control",
"date",
"vary",
"access-control-allow-origin",
"access-control-allow-methods",
"access-control-allow-headers",
"access-control-allow-credentials",
"access-control-expose-headers",
"access-control-max-age",
}
response_headers = {
k: v
for k, v in response.headers.items()
if k.lower() in allowed_headers
}
return Response(
content=json.dumps(response_json).encode(),
status_code=response.status_code,
headers=response_headers,
media_type="application/json",
)
except json.JSONDecodeError as e:
logger.error(
"Failed to parse JSON from upstream Responses API response",
extra={
"error": str(e),
"key_hash": key.hashed_key[:8] + "...",
"content_preview": content[:200].decode(errors="ignore")
if content
else "empty",
},
)
raise
except Exception as e:
logger.error(
"Error processing non-streaming Responses API completion",
extra={
"error": str(e),
"error_type": type(e).__name__,
"key_hash": key.hashed_key[:8] + "...",
},
)
raise
async def _finalize_generic_streaming_payment(
self, key_hash: str, max_cost: int, path: str
) -> None:
"""Background task to finalize payment for generic streaming requests."""
async with create_session() as session:
key = await session.get(ApiKey, key_hash)
if not key:
logger.warning(
"Key not found during background payment finalization",
extra={"key_hash": key_hash[:8] + "..."},
)
return
try:
# Finalize with "unknown" model and no usage to release reservation/charge max cost
await adjust_payment_for_tokens(
key,
{"model": "unknown", "usage": None},
session,
max_cost,
)
logger.info(
"Finalized generic streaming payment in background",
extra={
"path": path,
"key_hash": key_hash[:8] + "...",
},
)
except Exception as e:
logger.error(
"Error finalizing generic streaming payment in background",
extra={
"error": str(e),
"key_hash": key_hash[:8] + "...",
"path": path,
},
)
async def forward_request(
self,
request: Request,
path: str,
headers: dict,
request_body: bytes | None,
key: ApiKey,
max_cost_for_model: int,
session: AsyncSession,
model_obj: Model,
) -> Response | StreamingResponse:
"""Forward authenticated request to upstream service with cost tracking.
Args:
request: Original FastAPI request
path: Request path
headers: Prepared headers for upstream
request_body: Request body bytes, if any
key: API key for authenticated user
max_cost_for_model: Maximum cost deducted upfront
session: Database session for balance updates
Returns:
Response or StreamingResponse from upstream with cost tracking
"""
if path.startswith("v1/"):
path = path.replace("v1/", "")
url = f"{self.base_url}/{path}"
transformed_body = self.prepare_request_body(request_body, model_obj)
logger.info(
"Forwarding request to upstream",
extra={
"url": url,
"method": request.method,
"path": path,
"key_hash": key.hashed_key[:8] + "...",
"key_balance": key.balance,
"has_request_body": request_body is not None,
},
)
client = httpx.AsyncClient(
transport=httpx.AsyncHTTPTransport(retries=1),
timeout=None,
)
try:
if transformed_body is not None:
response = await client.send(
client.build_request(
request.method,
url,
headers=headers,
content=transformed_body,
params=self.prepare_params(path, request.query_params),
),
stream=True,
)
else:
response = await client.send(
client.build_request(
request.method,
url,
headers=headers,
content=request.stream(),
params=self.prepare_params(path, request.query_params),
),
stream=True,
)
logger.info(
"Received upstream response",
extra={
"status_code": response.status_code,
"path": path,
"key_hash": key.hashed_key[:8] + "...",
"content_type": response.headers.get("content-type", "unknown"),
},
)
if response.status_code != 200:
if response.status_code >= 500:
await response.aclose()
await client.aclose()
raise UpstreamError(
f"Upstream returned status {response.status_code}",
status_code=response.status_code,
)
try:
mapped_error = await self.map_upstream_error_response(
request, path, response
)
finally:
await response.aclose()
await client.aclose()
return mapped_error
if path.endswith("chat/completions") or path.endswith("embeddings"):
if path.endswith("chat/completions"):
client_wants_streaming = False
if request_body:
try:
request_data = json.loads(request_body)
client_wants_streaming = request_data.get("stream", False)
logger.debug(
"Chat completion request analysis",
extra={
"client_wants_streaming": client_wants_streaming,
"model": request_data.get("model", "unknown"),
"key_hash": key.hashed_key[:8] + "...",
},
)
except json.JSONDecodeError:
logger.warning(
"Failed to parse request body JSON for streaming detection"
)
content_type = response.headers.get("content-type", "")
upstream_is_streaming = "text/event-stream" in content_type
is_streaming = client_wants_streaming and upstream_is_streaming
logger.debug(
"Response type analysis",
extra={
"is_streaming": is_streaming,
"client_wants_streaming": client_wants_streaming,
"upstream_is_streaming": upstream_is_streaming,
"content_type": content_type,
"key_hash": key.hashed_key[:8] + "...",
},
)
if is_streaming and response.status_code == 200:
background_tasks = BackgroundTasks()
background_tasks.add_task(response.aclose)
background_tasks.add_task(client.aclose)
result = await self.handle_streaming_chat_completion(
response, key, max_cost_for_model, background_tasks
)
result.background = background_tasks
return result
# Handle both non-streaming chat completions and embeddings
if response.status_code == 200:
try:
return await self.handle_non_streaming_chat_completion(
response, key, session, max_cost_for_model
)
finally:
await response.aclose()
await client.aclose()
background_tasks = BackgroundTasks()
background_tasks.add_task(response.aclose)
background_tasks.add_task(client.aclose)
background_tasks.add_task(
self._finalize_generic_streaming_payment,
key.hashed_key,
max_cost_for_model,
path,
)
logger.debug(
"Streaming non-chat response",
extra={
"path": path,
"status_code": response.status_code,
"key_hash": key.hashed_key[:8] + "...",
},
)
return StreamingResponse(
response.aiter_bytes(),
status_code=response.status_code,
headers=dict(response.headers),
background=background_tasks,
)
except UpstreamError:
raise
except httpx.RequestError as exc:
await client.aclose()
error_type = type(exc).__name__
error_details = str(exc)
logger.error(
"HTTP request error to upstream",
extra={
"error_type": error_type,
"error_details": error_details,
"method": request.method,
"url": url,
"path": path,
"query_params": dict(request.query_params),
"key_hash": key.hashed_key[:8] + "...",
},
)
await revert_pay_for_request(key, session, max_cost_for_model)
if isinstance(exc, httpx.ConnectError):
error_message = "Unable to connect to upstream service"
elif isinstance(exc, httpx.TimeoutException):
error_message = "Upstream service request timed out"
elif isinstance(exc, httpx.NetworkError):
error_message = "Network error while connecting to upstream service"
else:
error_message = f"Error connecting to upstream service: {error_type}"
raise UpstreamError(error_message, status_code=502)
except Exception as exc:
await client.aclose()
tb = traceback.format_exc()
logger.error(
"Unexpected error in upstream forwarding",
extra={
"error": str(exc),
"error_type": type(exc).__name__,
"method": request.method,
"url": url,
"path": path,
"query_params": dict(request.query_params),
"key_hash": key.hashed_key[:8] + "...",
"traceback": tb,
},
)
await revert_pay_for_request(key, session, max_cost_for_model)
return create_error_response(
"internal_error",
"An unexpected server error occurred",
500,
request=request,
)
async def forward_responses_request(
self,
request: Request,
path: str,
headers: dict,
request_body: bytes | None,
key: ApiKey,
max_cost_for_model: int,
session: AsyncSession,
model_obj: Model,
) -> Response | StreamingResponse:
"""Forward authenticated Responses API request to upstream service with cost tracking.
Args:
request: Original FastAPI request
path: Request path
headers: Prepared headers for upstream
request_body: Request body bytes, if any
key: API key for authenticated user
max_cost_for_model: Maximum cost deducted upfront
session: Database session for balance updates
model_obj: Model object for the request
Returns:
Response or StreamingResponse from upstream with cost tracking
"""
# Remove v1/ prefix if present for Responses API
if path.startswith("v1/"):
path = path.replace("v1/", "")
url = f"{self.base_url}/{path}"
transformed_body = self.prepare_responses_request_body(request_body, model_obj)
logger.info(
"Forwarding Responses API request to upstream",
extra={
"url": url,
"method": request.method,
"path": path,
"key_hash": key.hashed_key[:8] + "...",
"key_balance": key.balance,
"has_request_body": request_body is not None,
},
)
client = httpx.AsyncClient(
transport=httpx.AsyncHTTPTransport(retries=1),
timeout=None,
)
try:
if transformed_body is not None:
response = await client.send(
client.build_request(
request.method,
url,
headers=headers,
content=transformed_body,
params=self.prepare_params(path, request.query_params),
),
stream=True,
)
else:
response = await client.send(
client.build_request(
request.method,
url,
headers=headers,
content=request.stream(),
params=self.prepare_params(path, request.query_params),
),
stream=True,
)
logger.info(
"Received upstream Responses API response",
extra={
"status_code": response.status_code,
"path": path,
"key_hash": key.hashed_key[:8] + "...",
"content_type": response.headers.get("content-type", "unknown"),
},
)
if response.status_code != 200:
if response.status_code >= 500:
await response.aclose()
await client.aclose()
raise UpstreamError(
f"Upstream returned status {response.status_code}",
status_code=response.status_code,
)
try:
mapped_error = await self.map_upstream_error_response(
request, path, response
)
finally:
await response.aclose()
await client.aclose()
return mapped_error
if path.startswith("responses"):
content_type = response.headers.get("content-type", "")
is_streaming = "text/event-stream" in content_type
logger.debug(
"Responses API response type analysis",
extra={
"is_streaming": is_streaming,
"content_type": content_type,
"key_hash": key.hashed_key[:8] + "...",
},
)
if is_streaming and response.status_code == 200:
result = await self.handle_streaming_responses_completion(
response, key, max_cost_for_model
)
background_tasks = BackgroundTasks()
background_tasks.add_task(response.aclose)
background_tasks.add_task(client.aclose)
result.background = background_tasks
return result
if response.status_code == 200:
try:
return await self.handle_non_streaming_responses_completion(
response, key, session, max_cost_for_model
)
finally:
await response.aclose()
await client.aclose()
background_tasks = BackgroundTasks()
background_tasks.add_task(response.aclose)
background_tasks.add_task(client.aclose)
background_tasks.add_task(
self._finalize_generic_streaming_payment,
key.hashed_key,
max_cost_for_model,
path,
)
logger.debug(
"Streaming non-Responses API response",
extra={
"path": path,
"status_code": response.status_code,
"key_hash": key.hashed_key[:8] + "...",
},
)
return StreamingResponse(
response.aiter_bytes(),
status_code=response.status_code,
headers=dict(response.headers),
background=background_tasks,
)
except UpstreamError:
raise
except httpx.RequestError as exc:
await client.aclose()
error_type = type(exc).__name__
error_details = str(exc)
logger.error(
"HTTP request error to upstream Responses API",
extra={
"error_type": error_type,
"error_details": error_details,
"method": request.method,
"url": url,
"path": path,
"query_params": dict(request.query_params),
"key_hash": key.hashed_key[:8] + "...",
},
)
await revert_pay_for_request(key, session, max_cost_for_model)
if isinstance(exc, httpx.ConnectError):
error_message = "Unable to connect to upstream service"
elif isinstance(exc, httpx.TimeoutException):
error_message = "Upstream service request timed out"
elif isinstance(exc, httpx.NetworkError):
error_message = "Network error while connecting to upstream service"
else:
error_message = f"Error connecting to upstream service: {error_type}"
raise UpstreamError(error_message, status_code=502)
except Exception as exc:
await client.aclose()
tb = traceback.format_exc()
logger.error(
"Unexpected error in upstream Responses API forwarding",
extra={
"error": str(exc),
"error_type": type(exc).__name__,
"method": request.method,
"url": url,
"path": path,
"query_params": dict(request.query_params),
"key_hash": key.hashed_key[:8] + "...",
"traceback": tb,
},
)
await revert_pay_for_request(key, session, max_cost_for_model)
return create_error_response(
"internal_error",
"An unexpected server error occurred",
500,
request=request,
)
async def forward_get_request(
self,
request: Request,
path: str,
headers: dict,
) -> Response | StreamingResponse:
"""Forward unauthenticated GET request to upstream service.
Args:
request: Original FastAPI request
path: Request path
headers: Prepared headers for upstream
Returns:
StreamingResponse from upstream
"""
if path.startswith("v1/"):
path = path.replace("v1/", "")
url = f"{self.base_url}/{path}"
logger.info(
"Forwarding GET request to upstream",
extra={"url": url, "method": request.method, "path": path},
)
async with httpx.AsyncClient(
transport=httpx.AsyncHTTPTransport(retries=1),
timeout=None,
) as client:
try:
response = await client.send(
client.build_request(
request.method,
url,
headers=headers,
content=request.stream(),
params=self.prepare_params(path, request.query_params),
),
)
logger.info(
"GET request forwarded successfully",
extra={"path": path, "status_code": response.status_code},
)
if response.status_code != 200:
try:
mapped = await self.map_upstream_error_response(
request, path, response
)
finally:
await response.aclose()
return mapped
return StreamingResponse(
response.aiter_bytes(),
status_code=response.status_code,
headers=dict(response.headers),
)
except Exception as exc:
tb = traceback.format_exc()
logger.error(
"Error forwarding GET request",
extra={
"error": str(exc),
"error_type": type(exc).__name__,
"method": request.method,
"url": url,
"path": path,
"query_params": dict(request.query_params),
"traceback": tb,
},
)
return create_error_response(
"internal_error",
"An unexpected server error occurred",
500,
request=request,
)
async def get_x_cashu_cost(
self, response_data: dict, max_cost_for_model: int
) -> MaxCostData | CostData | None:
"""Calculate cost for X-Cashu payment based on response data.
Args:
response_data: Response data containing model and usage information
max_cost_for_model: Maximum cost for the model
Returns:
Cost data object (MaxCostData or CostData) or None if calculation fails
"""
model = response_data.get("model", None)
logger.debug(
"Calculating cost for response",
extra={"model": model, "has_usage": "usage" in response_data},
)
async with create_session() as session:
match await calculate_cost(response_data, max_cost_for_model, session):
case MaxCostData() as cost:
logger.debug(
"Using max cost pricing",
extra={"model": model, "max_cost_msats": cost.total_msats},
)
return cost
case CostData() as cost:
logger.debug(
"Using token-based pricing",
extra={
"model": model,
"total_cost_msats": cost.total_msats,
"input_msats": cost.input_msats,
"output_msats": cost.output_msats,
},
)
return cost
case CostDataError() as error:
logger.error(
"Cost calculation error",
extra={
"model": model,
"error_message": error.message,
"error_code": error.code,
},
)
raise HTTPException(
status_code=400,
detail={
"error": {
"message": error.message,
"type": "invalid_request_error",
"code": error.code,
}
},
)
return None
async def send_refund(self, amount: int, unit: str, mint: str | None = None) -> str:
"""Create and send a refund token to the user.
Args:
amount: Refund amount
unit: Unit of the refund (sat or msat)
mint: Optional mint URL for the refund token
Returns:
Refund token string
"""
logger.debug(
"Creating refund token",
extra={"amount": amount, "unit": unit, "mint": mint},
)
max_retries = 3
last_exception = None
for attempt in range(max_retries):
try:
refund_token = await send_token(amount, unit=unit, mint_url=mint)
logger.info(
"Refund token created successfully",
extra={
"amount": amount,
"unit": unit,
"mint": mint,
"attempt": attempt + 1,
"token_preview": refund_token[:20] + "..."
if len(refund_token) > 20
else refund_token,
},
)
return refund_token
except Exception as e:
last_exception = e
if attempt < max_retries - 1:
logger.warning(
"Refund token creation failed, retrying",
extra={
"error": str(e),
"error_type": type(e).__name__,
"attempt": attempt + 1,
"max_retries": max_retries,
"amount": amount,
"unit": unit,
"mint": mint,
},
)
else:
logger.error(
"Failed to create refund token after all retries",
extra={
"error": str(e),
"error_type": type(e).__name__,
"attempt": attempt + 1,
"max_retries": max_retries,
"amount": amount,
"unit": unit,
"mint": mint,
},
)
raise HTTPException(
status_code=401,
detail={
"error": {
"message": f"failed to create refund after {max_retries} attempts: {str(last_exception)}",
"type": "invalid_request_error",
"code": "send_token_failed",
}
},
)
async def handle_x_cashu_streaming_response(
self,
content_str: str,
response: httpx.Response,
amount: int,
unit: str,
max_cost_for_model: int,
mint: str | None = None,
) -> StreamingResponse:
"""Handle streaming response for X-Cashu payment, calculating refund if needed.
Args:
content_str: Response content as string
response: Original httpx response
amount: Payment amount received
unit: Payment unit (sat or msat)
max_cost_for_model: Maximum cost for the model
Returns:
StreamingResponse with refund token in header if applicable
"""
logger.debug(
"Processing streaming response",
extra={
"amount": amount,
"unit": unit,
"content_lines": len(content_str.strip().split("\n")),
},
)
response_headers = dict(response.headers)
if "transfer-encoding" in response_headers:
del response_headers["transfer-encoding"]
if "content-encoding" in response_headers:
del response_headers["content-encoding"]
usage_data = None
model = None
lines = content_str.strip().split("\n")
for line in lines:
if line.startswith("data: "):
try:
data_json = json.loads(line[6:])
if "usage" in data_json:
usage_data = data_json["usage"]
model = data_json.get("model")
elif "model" in data_json and not model:
model = data_json["model"]
except json.JSONDecodeError:
continue
if usage_data and model:
logger.debug(
"Found usage data in streaming response",
extra={
"model": model,
"usage_data": usage_data,
"amount": amount,
"unit": unit,
},
)
response_data = {"usage": usage_data, "model": model}
try:
cost_data = await self.get_x_cashu_cost(
response_data, max_cost_for_model
)
if cost_data:
if unit == "msat":
refund_amount = amount - cost_data.total_msats
elif unit == "sat":
refund_amount = amount - (cost_data.total_msats + 999) // 1000
else:
raise ValueError(f"Invalid unit: {unit}")
if refund_amount > 0:
logger.info(
"Processing refund for streaming response",
extra={
"original_amount": amount,
"cost_msats": cost_data.total_msats,
"refund_amount": refund_amount,
"unit": unit,
"model": model,
},
)
refund_token = await self.send_refund(refund_amount, unit, mint)
response_headers["X-Cashu"] = refund_token
logger.info(
"Refund processed for streaming response",
extra={
"refund_amount": refund_amount,
"unit": unit,
"refund_token_preview": refund_token[:20] + "..."
if len(refund_token) > 20
else refund_token,
},
)
else:
logger.debug(
"No refund needed for streaming response",
extra={
"amount": amount,
"cost_msats": cost_data.total_msats,
"model": model,
},
)
except Exception as e:
logger.error(
"Error calculating cost for streaming response",
extra={
"error": str(e),
"error_type": type(e).__name__,
"model": model,
"amount": amount,
"unit": unit,
},
)
async def generate() -> AsyncGenerator[bytes, None]:
for line in lines:
yield (line + "\n").encode("utf-8")
return StreamingResponse(
generate(),
status_code=response.status_code,
headers=response_headers,
media_type="text/plain",
)
async def handle_x_cashu_non_streaming_response(
self,
content_str: str,
response: httpx.Response,
amount: int,
unit: str,
max_cost_for_model: int,
mint: str | None = None,
) -> Response:
"""Handle non-streaming response for X-Cashu payment, calculating refund if needed.
Args:
content_str: Response content as string
response: Original httpx response
amount: Payment amount received
unit: Payment unit (sat or msat)
max_cost_for_model: Maximum cost for the model
Returns:
Response with refund token in header if applicable
"""
logger.debug(
"Processing non-streaming response",
extra={"amount": amount, "unit": unit, "content_length": len(content_str)},
)
try:
response_json = json.loads(content_str)
cost_data = await self.get_x_cashu_cost(response_json, max_cost_for_model)
if not cost_data:
logger.error(
"Failed to calculate cost for response",
extra={
"amount": amount,
"unit": unit,
"response_model": response_json.get("model", "unknown"),
},
)
return Response(
content=json.dumps(
{
"error": {
"message": "Error forwarding request to upstream",
"type": "upstream_error",
"code": response.status_code,
}
}
),
status_code=response.status_code,
media_type="application/json",
)
response_headers = dict(response.headers)
if "transfer-encoding" in response_headers:
del response_headers["transfer-encoding"]
if "content-encoding" in response_headers:
del response_headers["content-encoding"]
if unit == "msat":
refund_amount = amount - cost_data.total_msats
elif unit == "sat":
refund_amount = amount - (cost_data.total_msats + 999) // 1000
else:
raise ValueError(f"Invalid unit: {unit}")
logger.info(
"Processing non-streaming response cost calculation",
extra={
"original_amount": amount,
"cost_msats": cost_data.total_msats,
"refund_amount": refund_amount,
"unit": unit,
"model": response_json.get("model", "unknown"),
},
)
if refund_amount > 0:
refund_token = await self.send_refund(refund_amount, unit, mint)
response_headers["X-Cashu"] = refund_token
logger.info(
"Refund processed for non-streaming response",
extra={
"refund_amount": refund_amount,
"unit": unit,
"refund_token_preview": refund_token[:20] + "..."
if len(refund_token) > 20
else refund_token,
},
)
return Response(
content=content_str,
status_code=response.status_code,
headers=response_headers,
media_type="application/json",
)
except json.JSONDecodeError as e:
logger.error(
"Failed to parse JSON from upstream response",
extra={
"error": str(e),
"content_preview": content_str[:200] + "..."
if len(content_str) > 200
else content_str,
"amount": amount,
"unit": unit,
},
)
emergency_refund = amount
refund_token = await send_token(emergency_refund, unit=unit, mint_url=mint)
response.headers["X-Cashu"] = refund_token
logger.warning(
"Emergency refund issued due to JSON parse error",
extra={
"original_amount": amount,
"refund_amount": emergency_refund,
"deduction": 60,
},
)
return Response(
content=content_str,
status_code=response.status_code,
headers=dict(response.headers),
media_type="application/json",
)
async def handle_x_cashu_chat_completion(
self,
response: httpx.Response,
amount: int,
unit: str,
max_cost_for_model: int,
mint: str | None = None,
) -> StreamingResponse | Response:
"""Handle chat completion response for X-Cashu payment, detecting streaming vs non-streaming.
Args:
response: Response from upstream
amount: Payment amount received
unit: Payment unit (sat or msat)
max_cost_for_model: Maximum cost for the model
Returns:
StreamingResponse or Response depending on response type
"""
logger.debug(
"Handling chat completion response",
extra={"amount": amount, "unit": unit, "status_code": response.status_code},
)
try:
content = await response.aread()
content_str = (
content.decode("utf-8") if isinstance(content, bytes) else content
)
is_streaming = content_str.startswith("data:") or "data:" in content_str
logger.debug(
"Chat completion response analysis",
extra={
"is_streaming": is_streaming,
"content_length": len(content_str),
"amount": amount,
"unit": unit,
},
)
if is_streaming:
return await self.handle_x_cashu_streaming_response(
content_str, response, amount, unit, max_cost_for_model, mint
)
else:
return await self.handle_x_cashu_non_streaming_response(
content_str, response, amount, unit, max_cost_for_model, mint
)
except Exception as e:
logger.error(
"Error processing chat completion response",
extra={
"error": str(e),
"error_type": type(e).__name__,
"amount": amount,
"unit": unit,
},
)
return StreamingResponse(
response.aiter_bytes(),
status_code=response.status_code,
headers=dict(response.headers),
)
async def forward_x_cashu_request(
self,
request: Request,
path: str,
headers: dict,
amount: int,
unit: str,
max_cost_for_model: int,
model_obj: Model,
mint: str | None = None,
) -> Response | StreamingResponse:
"""Forward request paid with X-Cashu token to upstream service.
Args:
request: Original FastAPI request
path: Request path
headers: Prepared headers for upstream
amount: Payment amount from X-Cashu token
unit: Payment unit (sat or msat)
max_cost_for_model: Maximum cost for the model
model_obj: Model object for the request
Returns:
Response or StreamingResponse with refund if applicable
"""
if path.startswith("v1/"):
path = path.replace("v1/", "")
url = f"{self.base_url}/{path}"
request_body = await request.body()
transformed_body = self.prepare_request_body(request_body, model_obj)
logger.debug(
"Forwarding request to upstream",
extra={
"url": url,
"method": request.method,
"path": path,
"amount": amount,
"unit": unit,
},
)
async with httpx.AsyncClient(
transport=httpx.AsyncHTTPTransport(retries=1),
timeout=None,
) as client:
try:
response = await client.send(
client.build_request(
request.method,
url,
headers=headers,
content=transformed_body if transformed_body else request_body,
params=self.prepare_params(path, request.query_params),
),
stream=True,
)
logger.debug(
"Received upstream response",
extra={
"status_code": response.status_code,
"path": path,
"response_headers": dict(response.headers),
},
)
if response.status_code != 200:
logger.warning(
"Upstream request failed, processing refund",
extra={
"status_code": response.status_code,
"path": path,
"amount": amount,
"unit": unit,
},
)
refund_token = await self.send_refund(amount - 60, unit, mint)
logger.info(
"Refund processed for failed upstream request",
extra={
"status_code": response.status_code,
"refund_amount": amount,
"unit": unit,
"refund_token_preview": refund_token[:20] + "..."
if len(refund_token) > 20
else refund_token,
},
)
error_response = Response(
content=json.dumps(
{
"error": {
"message": "Error forwarding request to upstream",
"type": "upstream_error",
"code": response.status_code,
"refund_token": refund_token,
}
}
),
status_code=response.status_code,
media_type="application/json",
)
error_response.headers["X-Cashu"] = refund_token
return error_response
if path.endswith("chat/completions") or path.endswith("embeddings"):
logger.debug(
"Processing completion/embeddings response",
extra={"path": path, "amount": amount, "unit": unit},
)
result = await self.handle_x_cashu_chat_completion(
response, amount, unit, max_cost_for_model, mint
)
background_tasks = BackgroundTasks()
background_tasks.add_task(response.aclose)
result.background = background_tasks
return result
background_tasks = BackgroundTasks()
background_tasks.add_task(response.aclose)
background_tasks.add_task(client.aclose)
logger.debug(
"Streaming non-chat response",
extra={"path": path, "status_code": response.status_code},
)
return StreamingResponse(
response.aiter_bytes(),
status_code=response.status_code,
headers=dict(response.headers),
background=background_tasks,
)
except Exception as exc:
tb = traceback.format_exc()
logger.error(
"Unexpected error in upstream forwarding",
extra={
"error": str(exc),
"error_type": type(exc).__name__,
"method": request.method,
"url": url,
"path": path,
"query_params": dict(request.query_params),
"traceback": tb,
},
)
return create_error_response(
"internal_error",
"An unexpected server error occurred",
500,
request=request,
)
async def handle_x_cashu_responses(
self,
request: Request,
x_cashu_token: str,
path: str,
max_cost_for_model: int,
model_obj: Model,
) -> Response | StreamingResponse:
"""Handle X-Cashu payment for Responses API requests.
Args:
request: Original FastAPI request
x_cashu_token: X-Cashu token from request header
path: Request path
max_cost_for_model: Maximum cost for the model
model_obj: Model object for the request
Returns:
Response or StreamingResponse from upstream with refund if applicable
"""
logger.info(
"Processing X-Cashu payment for Responses API",
extra={
"path": path,
"method": request.method,
"token_preview": x_cashu_token[:20] + "..."
if len(x_cashu_token) > 20
else x_cashu_token,
},
)
try:
headers = dict(request.headers)
amount, unit, mint = await recieve_token(x_cashu_token)
headers = self.prepare_headers(dict(request.headers))
logger.info(
"X-Cashu token redeemed for Responses API",
extra={"amount": amount, "unit": unit, "path": path, "mint": mint},
)
return await self.forward_x_cashu_responses_request(
request,
path,
headers,
amount,
unit,
max_cost_for_model,
model_obj,
mint,
)
except Exception as e:
error_message = str(e)
logger.error(
"X-Cashu payment for Responses API failed",
extra={
"error": error_message,
"error_type": type(e).__name__,
"path": path,
"method": request.method,
},
)
# Use same error handling as regular X-Cashu
if "already spent" in error_message.lower():
return create_error_response(
"token_already_spent",
"The provided CASHU token has already been spent",
400,
request=request,
token=x_cashu_token,
)
if "invalid token" in error_message.lower():
return create_error_response(
"invalid_token",
"The provided CASHU token is invalid",
400,
request=request,
token=x_cashu_token,
)
if "mint error" in error_message.lower():
return create_error_response(
"mint_error",
f"CASHU mint error: {error_message}",
422,
request=request,
token=x_cashu_token,
)
return create_error_response(
"cashu_error",
f"CASHU token processing failed: {error_message}",
400,
request=request,
token=x_cashu_token,
)
async def forward_x_cashu_responses_request(
self,
request: Request,
path: str,
headers: dict,
amount: int,
unit: str,
max_cost_for_model: int,
model_obj: Model,
mint: str | None = None,
) -> Response | StreamingResponse:
"""Forward Responses API request paid with X-Cashu token to upstream service.
Args:
request: Original FastAPI request
path: Request path
headers: Prepared headers for upstream
amount: Payment amount from X-Cashu token
unit: Payment unit (sat or msat)
max_cost_for_model: Maximum cost for the model
model_obj: Model object for the request
mint: Mint URL for refund tokens
Returns:
Response or StreamingResponse with refund if applicable
"""
if path.startswith("v1/"):
path = path.replace("v1/", "")
url = f"{self.base_url}/{path}"
request_body = await request.body()
transformed_body = self.prepare_responses_request_body(request_body, model_obj)
logger.debug(
"Forwarding Responses API request to upstream with X-Cashu payment",
extra={
"url": url,
"method": request.method,
"path": path,
"amount": amount,
"unit": unit,
},
)
async with httpx.AsyncClient(
transport=httpx.AsyncHTTPTransport(retries=1),
timeout=None,
) as client:
try:
response = await client.send(
client.build_request(
request.method,
url,
headers=headers,
content=transformed_body if transformed_body else request_body,
params=self.prepare_params(path, request.query_params),
),
stream=True,
)
logger.debug(
"Received upstream Responses API response",
extra={
"status_code": response.status_code,
"path": path,
"response_headers": dict(response.headers),
},
)
if response.status_code != 200:
logger.warning(
"Upstream Responses API request failed, processing refund",
extra={
"status_code": response.status_code,
"path": path,
"amount": amount,
"unit": unit,
},
)
refund_token = await self.send_refund(amount - 60, unit, mint)
logger.info(
"Refund processed for failed upstream Responses API request",
extra={
"status_code": response.status_code,
"refund_amount": amount,
"unit": unit,
"refund_token_preview": refund_token[:20] + "..."
if len(refund_token) > 20
else refund_token,
},
)
error_response = Response(
content=json.dumps(
{
"error": {
"message": "Error forwarding Responses API request to upstream",
"type": "upstream_error",
"code": response.status_code,
"refund_token": refund_token,
}
}
),
status_code=response.status_code,
media_type="application/json",
)
error_response.headers["X-Cashu"] = refund_token
return error_response
if path.startswith("responses"):
logger.debug(
"Processing Responses API response",
extra={"path": path, "amount": amount, "unit": unit},
)
result = await self.handle_x_cashu_responses_completion(
response, amount, unit, max_cost_for_model, mint
)
background_tasks = BackgroundTasks()
background_tasks.add_task(response.aclose)
result.background = background_tasks
return result
background_tasks = BackgroundTasks()
background_tasks.add_task(response.aclose)
background_tasks.add_task(client.aclose)
logger.debug(
"Streaming non-responses response",
extra={"path": path, "status_code": response.status_code},
)
return StreamingResponse(
response.aiter_bytes(),
status_code=response.status_code,
headers=dict(response.headers),
background=background_tasks,
)
except Exception as exc:
tb = traceback.format_exc()
logger.error(
"Unexpected error in upstream Responses API forwarding",
extra={
"error": str(exc),
"error_type": type(exc).__name__,
"method": request.method,
"url": url,
"path": path,
"query_params": dict(request.query_params),
"traceback": tb,
},
)
return create_error_response(
"internal_error",
"An unexpected server error occurred",
500,
request=request,
)
async def handle_x_cashu_responses_completion(
self,
response: httpx.Response,
amount: int,
unit: str,
max_cost_for_model: int,
mint: str | None = None,
) -> StreamingResponse | Response:
"""Handle Responses API completion response for X-Cashu payment.
Args:
response: Response from upstream
amount: Payment amount received
unit: Payment unit (sat or msat)
max_cost_for_model: Maximum cost for the model
mint: Mint URL for refund tokens
Returns:
StreamingResponse or Response depending on response type
"""
logger.debug(
"Handling Responses API completion response",
extra={"amount": amount, "unit": unit, "status_code": response.status_code},
)
try:
content = await response.aread()
content_str = (
content.decode("utf-8") if isinstance(content, bytes) else content
)
is_streaming = content_str.startswith("data:") or "data:" in content_str
logger.debug(
"Responses API completion response analysis",
extra={
"is_streaming": is_streaming,
"content_length": len(content_str),
"amount": amount,
"unit": unit,
},
)
if is_streaming:
return await self.handle_x_cashu_streaming_responses_response(
content_str, response, amount, unit, max_cost_for_model, mint
)
else:
return await self.handle_x_cashu_non_streaming_responses_response(
content_str, response, amount, unit, max_cost_for_model, mint
)
except Exception as e:
logger.error(
"Error processing Responses API completion response",
extra={
"error": str(e),
"error_type": type(e).__name__,
"amount": amount,
"unit": unit,
},
)
return StreamingResponse(
response.aiter_bytes(),
status_code=response.status_code,
headers=dict(response.headers),
)
async def handle_x_cashu_streaming_responses_response(
self,
content_str: str,
response: httpx.Response,
amount: int,
unit: str,
max_cost_for_model: int,
mint: str | None = None,
) -> StreamingResponse:
"""Handle streaming Responses API response for X-Cashu payment.
Similar to regular streaming but handles Responses API specific tokens like reasoning_tokens.
"""
logger.debug(
"Processing streaming Responses API response",
extra={
"amount": amount,
"unit": unit,
"content_lines": len(content_str.strip().split("\\n")),
},
)
response_headers = dict(response.headers)
if "transfer-encoding" in response_headers:
del response_headers["transfer-encoding"]
if "content-encoding" in response_headers:
del response_headers["content-encoding"]
usage_data = None
model = None
reasoning_tokens = 0
lines = content_str.strip().split("\\n")
for line in lines:
if line.startswith("data: "):
try:
data_json = json.loads(line[6:])
if "usage" in data_json:
usage_data = data_json["usage"]
model = data_json.get("model")
# Track reasoning tokens for Responses API
if (
isinstance(usage_data, dict)
and "reasoning_tokens" in usage_data
):
reasoning_tokens = usage_data.get("reasoning_tokens", 0)
elif "model" in data_json and not model:
model = data_json["model"]
except json.JSONDecodeError:
continue
if usage_data and model:
logger.debug(
"Found usage data in streaming Responses API response",
extra={
"model": model,
"usage_data": usage_data,
"reasoning_tokens": reasoning_tokens,
"amount": amount,
"unit": unit,
},
)
response_data = {"usage": usage_data, "model": model}
try:
cost_data = await self.get_x_cashu_cost(
response_data, max_cost_for_model
)
if cost_data:
if unit == "msat":
refund_amount = amount - cost_data.total_msats
elif unit == "sat":
refund_amount = amount - (cost_data.total_msats + 999) // 1000
else:
raise ValueError(f"Invalid unit: {unit}")
if refund_amount > 0:
logger.info(
"Processing refund for streaming Responses API response",
extra={
"original_amount": amount,
"cost_msats": cost_data.total_msats,
"refund_amount": refund_amount,
"unit": unit,
"model": model,
"reasoning_tokens": reasoning_tokens,
},
)
refund_token = await self.send_refund(refund_amount, unit, mint)
response_headers["X-Cashu"] = refund_token
logger.info(
"Refund processed for streaming Responses API response",
extra={
"refund_amount": refund_amount,
"unit": unit,
"refund_token_preview": refund_token[:20] + "..."
if len(refund_token) > 20
else refund_token,
},
)
else:
logger.debug(
"No refund needed for streaming Responses API response",
extra={
"amount": amount,
"cost_msats": cost_data.total_msats,
"model": model,
},
)
except Exception as e:
logger.error(
"Error calculating cost for streaming Responses API response",
extra={
"error": str(e),
"error_type": type(e).__name__,
"model": model,
"amount": amount,
"unit": unit,
},
)
async def generate() -> AsyncGenerator[bytes, None]:
for line in lines:
yield (line + "\\n").encode("utf-8")
return StreamingResponse(
generate(),
status_code=response.status_code,
headers=response_headers,
media_type="text/plain",
)
async def handle_x_cashu_non_streaming_responses_response(
self,
content_str: str,
response: httpx.Response,
amount: int,
unit: str,
max_cost_for_model: int,
mint: str | None = None,
) -> Response:
"""Handle non-streaming Responses API response for X-Cashu payment."""
logger.debug(
"Processing non-streaming Responses API response",
extra={"amount": amount, "unit": unit, "content_length": len(content_str)},
)
try:
response_json = json.loads(content_str)
cost_data = await self.get_x_cashu_cost(response_json, max_cost_for_model)
if not cost_data:
logger.error(
"Failed to calculate cost for Responses API response",
extra={
"amount": amount,
"unit": unit,
"response_model": response_json.get("model", "unknown"),
},
)
return Response(
content=json.dumps(
{
"error": {
"message": "Error forwarding Responses API request to upstream",
"type": "upstream_error",
"code": response.status_code,
}
}
),
status_code=response.status_code,
media_type="application/json",
)
response_headers = dict(response.headers)
if "transfer-encoding" in response_headers:
del response_headers["transfer-encoding"]
if "content-encoding" in response_headers:
del response_headers["content-encoding"]
if unit == "msat":
refund_amount = amount - cost_data.total_msats
elif unit == "sat":
refund_amount = amount - (cost_data.total_msats + 999) // 1000
else:
raise ValueError(f"Invalid unit: {unit}")
logger.info(
"Processing non-streaming Responses API cost calculation",
extra={
"original_amount": amount,
"cost_msats": cost_data.total_msats,
"refund_amount": refund_amount,
"unit": unit,
"model": response_json.get("model", "unknown"),
},
)
if refund_amount > 0:
refund_token = await self.send_refund(refund_amount, unit, mint)
response_headers["X-Cashu"] = refund_token
logger.info(
"Refund processed for non-streaming Responses API response",
extra={
"refund_amount": refund_amount,
"unit": unit,
"refund_token_preview": refund_token[:20] + "..."
if len(refund_token) > 20
else refund_token,
},
)
return Response(
content=content_str,
status_code=response.status_code,
headers=response_headers,
media_type="application/json",
)
except json.JSONDecodeError as e:
logger.error(
"Failed to parse JSON from upstream Responses API response",
extra={
"error": str(e),
"content_preview": content_str[:200] + "..."
if len(content_str) > 200
else content_str,
"amount": amount,
"unit": unit,
},
)
emergency_refund = amount
refund_token = await send_token(emergency_refund, unit=unit, mint_url=mint)
response.headers["X-Cashu"] = refund_token
logger.warning(
"Emergency refund issued for Responses API due to JSON parse error",
extra={
"original_amount": amount,
"refund_amount": emergency_refund,
"deduction": 60,
},
)
return Response(
content=content_str,
status_code=response.status_code,
headers=dict(response.headers),
media_type="application/json",
)
async def handle_x_cashu(
self,
request: Request,
x_cashu_token: str,
path: str,
max_cost_for_model: int,
model_obj: Model,
) -> Response | StreamingResponse:
"""Handle request with X-Cashu token payment, redeeming token and forwarding request.
Args:
request: Original FastAPI request
x_cashu_token: X-Cashu token from request header
path: Request path
max_cost_for_model: Maximum cost for the model
model_obj: Model object for the request
Returns:
Response or StreamingResponse from upstream with refund if applicable
"""
logger.info(
"Processing X-Cashu payment request",
extra={
"path": path,
"method": request.method,
"token_preview": x_cashu_token[:20] + "..."
if len(x_cashu_token) > 20
else x_cashu_token,
},
)
try:
headers = dict(request.headers)
amount, unit, mint = await recieve_token(x_cashu_token)
headers = self.prepare_headers(dict(request.headers))
logger.info(
"X-Cashu token redeemed successfully",
extra={"amount": amount, "unit": unit, "path": path, "mint": mint},
)
return await self.forward_x_cashu_request(
request,
path,
headers,
amount,
unit,
max_cost_for_model,
model_obj,
mint,
)
except Exception as e:
error_message = str(e)
logger.error(
"X-Cashu payment request failed",
extra={
"error": error_message,
"error_type": type(e).__name__,
"path": path,
"method": request.method,
},
)
if "already spent" in error_message.lower():
return create_error_response(
"token_already_spent",
"The provided CASHU token has already been spent",
400,
request=request,
token=x_cashu_token,
)
if "invalid token" in error_message.lower():
return create_error_response(
"invalid_token",
"The provided CASHU token is invalid",
400,
request=request,
token=x_cashu_token,
)
if "mint error" in error_message.lower():
return create_error_response(
"mint_error",
f"CASHU mint error: {error_message}",
422,
request=request,
token=x_cashu_token,
)
return create_error_response(
"cashu_error",
f"CASHU token processing failed: {error_message}",
400,
request=request,
token=x_cashu_token,
)
def _apply_provider_fee_to_model(self, model: Model) -> Model:
"""Apply provider fee to model's USD pricing and calculate max costs.
Args:
model: Model object to update
Returns:
Model with provider fee applied to pricing and max costs calculated
"""
adjusted_pricing = Pricing.parse_obj(
{k: v * self.provider_fee for k, v in model.pricing.dict().items()}
)
temp_model = Model(
id=model.id,
name=model.name,
created=model.created,
description=model.description,
context_length=model.context_length,
architecture=model.architecture,
pricing=adjusted_pricing,
sats_pricing=None,
per_request_limits=model.per_request_limits,
top_provider=model.top_provider,
enabled=model.enabled,
upstream_provider_id=model.upstream_provider_id,
canonical_slug=model.canonical_slug,
alias_ids=model.alias_ids,
)
(
adjusted_pricing.max_prompt_cost,
adjusted_pricing.max_completion_cost,
adjusted_pricing.max_cost,
) = _calculate_usd_max_costs(temp_model)
return Model(
id=model.id,
name=model.name,
created=model.created,
description=model.description,
context_length=model.context_length,
architecture=model.architecture,
pricing=adjusted_pricing,
sats_pricing=model.sats_pricing,
per_request_limits=model.per_request_limits,
top_provider=model.top_provider,
enabled=model.enabled,
upstream_provider_id=model.upstream_provider_id,
canonical_slug=model.canonical_slug,
alias_ids=model.alias_ids,
)
async def fetch_models(self) -> list[Model]:
"""Fetch available models from upstream API and update cache.
Returns:
List of Model objects with pricing
"""
try:
or_models, provider_models_response = await asyncio.gather(
self._fetch_openrouter_models(),
self._fetch_provider_models(),
)
provider_model_ids = self._parse_model_ids(provider_models_response)
found_models = []
not_found_models = []
for model_id in provider_model_ids:
or_model = self._match_model(model_id, or_models)
if or_model:
try:
model = Model(**or_model) # type: ignore
found_models.append(model)
except Exception as e:
logger.warning(
f"Failed to parse model {model_id}",
extra={"error": str(e), "error_type": type(e).__name__},
)
else:
not_found_models.append(model_id)
if not_found_models:
logger.debug(
f"({len(not_found_models)}/{len(provider_model_ids)}) unmatched models for {self.provider_type or self.base_url}",
extra={"not_found_models": not_found_models},
)
return found_models
except Exception as e:
logger.error(
f"Error fetching models for {self.provider_type or self.base_url}",
extra={"error": str(e), "error_type": type(e).__name__},
)
return []
async def _fetch_openrouter_models(self) -> list[dict]:
"""Fetch models from OpenRouter API."""
url = "https://openrouter.ai/api/v1/models"
embeddings_url = "https://openrouter.ai/api/v1/embeddings/models"
async with httpx.AsyncClient(timeout=30.0) as client:
models_response, embeddings_response = await asyncio.gather(
client.get(url), client.get(embeddings_url), return_exceptions=True
)
all_models = []
def process_models_response(
response: httpx.Response | BaseException,
) -> list[dict]:
if not isinstance(response, BaseException):
response.raise_for_status()
data = response.json()
return [
model
for model in data.get("data", [])
if ":free" not in model.get("id", "").lower()
]
return []
all_models.extend(process_models_response(models_response))
all_models.extend(process_models_response(embeddings_response))
return all_models
async def _fetch_provider_models(self) -> dict:
"""Fetch models from provider's API."""
url = f"{self.base_url.rstrip('/')}/models"
headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else None
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(url, headers=headers)
response.raise_for_status()
return response.json()
def _parse_model_ids(self, response: dict) -> list[str]:
"""Parse model IDs from provider response."""
return [model.get("id") for model in response.get("data", []) if "id" in model]
def _match_model(self, model_id: str, or_models: list[dict]) -> dict | None:
"""Match provider model ID with OpenRouter model."""
return next(
(
model
for model in or_models
if (model.get("id") == model_id)
or (model.get("id", "").split("/")[-1] == model_id)
or (model.get("canonical_slug") == model_id)
or (model.get("canonical_slug", "").split("/")[-1] == model_id)
),
None,
)
async def refresh_models_cache(self) -> None:
"""Refresh the in-memory models cache from upstream API."""
try:
async with create_session() as session:
stmt = select(UpstreamProviderRow).where(
UpstreamProviderRow.base_url == self.base_url,
UpstreamProviderRow.api_key == self.api_key
)
result = await session.exec(stmt)
# .first() returns the object or None if not found
provider = result.first()
if not provider or not provider.id:
raise HTTPException(status_code=404, detail="Provider not found")
db_models = await list_models(
session=session,
upstream_id=provider.id,
include_disabled=False,
apply_fees=False,
)
db_model_ids: set[str] = {model.id for model in db_models}
models = await self.fetch_models()
model_ids = [model.id for model in models]
diff = set(db_model_ids) - set(model_ids)
for db_model_id in diff:
found_db_model = next((model_obj for model_obj in db_models if model_obj.id == db_model_id))
models.append(found_db_model)
models_with_fees = [self._apply_provider_fee_to_model(m) for m in models]
print([mode.id for mode in models_with_fees])
try:
sats_to_usd = sats_usd_price()
self._models_cache = [
_update_model_sats_pricing(m, sats_to_usd) for m in models_with_fees
]
except Exception:
self._models_cache = models_with_fees
self._models_by_id = {m.id: m for m in self._models_cache}
except Exception as e:
logger.error(
f"Failed to refresh models cache for {self.provider_type or self.base_url}",
extra={"error": str(e), "error_type": type(e).__name__},
)
def get_cached_models(self) -> list[Model]:
"""Get cached models for this provider.
Returns:
List of cached Model objects
"""
return self._models_cache
def get_cached_model_by_id(self, model_id: str) -> Model | None:
"""Get a specific cached model by ID.
Args:
model_id: Model identifier
Returns:
Model object or None if not found
"""
return self._models_by_id.get(model_id)
@classmethod
async def create_account_static(cls) -> dict[str, object]:
"""Create a new account with the provider (class method, no instance needed).
Returns:
Dict with account creation details including api_key
Raises:
NotImplementedError: If provider does not support account creation
"""
raise NotImplementedError(
f"Provider {cls.provider_type} does not support account creation"
)
async def create_account(self) -> dict[str, object]:
"""Create a new account with the provider.
Returns:
Dict with account creation details including api_key
Raises:
NotImplementedError: If provider does not support account creation
"""
raise NotImplementedError(
f"Provider {self.provider_type} does not support account creation"
)
async def initiate_topup(self, amount: int) -> TopupData:
"""Initiate a Lightning Network top-up for the provider account.
Args:
amount: Amount in currency units to top up
Returns:
TopupData with standardized invoice information
Raises:
NotImplementedError: If provider does not support top-up
"""
raise NotImplementedError(
f"Provider {self.provider_type} does not support top-up"
)
async def get_balance(self) -> float | None:
"""Get the current account balance from the provider.
Returns:
Float representing the balance amount, or None if not supported/available.
Typically in USD or the provider's credit unit.
Raises:
NotImplementedError: If provider does not support balance checking (default behavior)
"""
raise NotImplementedError(
f"Provider {self.provider_type} does not support balance checking"
)
Reference in New Issue View Git Blame Copy Permalink