diff --git a/example.py b/example.py deleted file mode 100644 index 4732859d..00000000 --- a/example.py +++ /dev/null @@ -1,37 +0,0 @@ -import os - -import openai - -client = openai.OpenAI( - api_key=os.environ["CASHU_TOKEN"], - base_url=os.environ.get("ROUTSTR_API_URL", "https://api.routstr.com/v1"), - # base_url="http://roustrjfsdgfiueghsklchg.onion/v1", - # client=httpx.AsyncClient( - # proxies={"http": "socks5://localhost:9050"}, - # ), # to use onion proxy (tor) -) -history: list = [] - - -def chat() -> None: - while True: - user_msg = {"role": "user", "content": input("\nYou: ")} - history.append(user_msg) - ai_msg = {"role": "assistant", "content": ""} - - for chunk in client.chat.completions.create( - model=os.environ.get("MODEL", "openai/gpt-4o-mini"), - messages=history, - stream=True, - ): - if len(chunk.choices) > 0: - content = chunk.choices[0].delta.content - if content is not None: - ai_msg["content"] += content - print(content, end="", flush=True) - print() - history.append(ai_msg) - - -if __name__ == "__main__": - chat() diff --git a/examples/balance/check_balance.py b/examples/balance/check_balance.py new file mode 100644 index 00000000..5a2a7c25 --- /dev/null +++ b/examples/balance/check_balance.py @@ -0,0 +1,11 @@ +import os + +import httpx + +# Use your Cashu token or API key as the Bearer token, +# cashu token is hashed on the server and acts as an Temporary API key +headers = {"Authorization": f"Bearer {os.environ.get('TOKEN')}"} +base_url = os.environ.get("API_URL", "https://api.routstr.com/v1") + +resp = httpx.get(f"{base_url}/balance/info", headers=headers) +print(resp.json()) diff --git a/examples/balance/create_balance.py b/examples/balance/create_balance.py new file mode 100644 index 00000000..0f88fc58 --- /dev/null +++ b/examples/balance/create_balance.py @@ -0,0 +1,15 @@ +import os + +import httpx + +# Send a Cashu token to the /create endpoint to get a persistent API key +token = os.environ.get("TOKEN") +if not token: + print("Please set TOKEN environment variable with a Cashu token") + exit(1) + +base_url = os.environ.get("API_URL", "https://api.routstr.com/v1") + +resp = httpx.get(f"{base_url}/balance/create", params={"initial_balance_token": token}) + +print(resp.json()) diff --git a/examples/balance/refund_balance.py b/examples/balance/refund_balance.py new file mode 100644 index 00000000..95a11ae5 --- /dev/null +++ b/examples/balance/refund_balance.py @@ -0,0 +1,12 @@ +import os + +import httpx + +# Use your Cashu token or API key as the Bearer token +headers = {"Authorization": f"Bearer {os.environ.get('TOKEN')}"} +base_url = os.environ.get("API_URL", "https://api.routstr.com/v1") + +resp = httpx.post(f"{base_url}/balance/refund", headers=headers) + +print("Refund successful!") +print(resp.json()) diff --git a/examples/balance/topup_balance.py b/examples/balance/topup_balance.py new file mode 100644 index 00000000..6f47d240 --- /dev/null +++ b/examples/balance/topup_balance.py @@ -0,0 +1,16 @@ +import os + +import httpx + +# Use your Cashu token or API key as the Bearer token +headers = {"Authorization": f"Bearer {os.environ.get('TOKEN')}"} +base_url = os.environ.get("API_URL", "https://api.routstr.com/v1") + +# The Cashu token to top up with +cashu_token = input("Enter Cashu token to top up: ") + +resp = httpx.post( + f"{base_url}/balance/topup", headers=headers, json={"cashu_token": cashu_token} +) + +print(resp.json()) diff --git a/examples/chat_completions.py b/examples/chat_completions.py new file mode 100644 index 00000000..f7b4b9a4 --- /dev/null +++ b/examples/chat_completions.py @@ -0,0 +1,15 @@ +import os + +from openai import OpenAI + +client = OpenAI( + api_key=os.environ.get("TOKEN"), + base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"), +) + +response = client.chat.completions.create( + model=os.environ.get("MODEL", "gpt-5-nano"), + messages=[{"role": "user", "content": "Hello!"}], +) + +print(response.choices[0].message.content) diff --git a/examples/list_models.py b/examples/list_models.py new file mode 100644 index 00000000..8aba55e3 --- /dev/null +++ b/examples/list_models.py @@ -0,0 +1,19 @@ +import os + +import httpx +from openai import OpenAI + +client = OpenAI( + api_key=os.environ.get("TOKEN", ""), + base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"), +) + +for model in client.models.list(): + print(model.id) + +# OR + +models = httpx.get( + f"{client.base_url}/v1/models", + headers={"Authorization": f"Bearer {client.api_key}"}, +).json() diff --git a/examples/responses/conversation.py b/examples/responses/conversation.py new file mode 100644 index 00000000..2f2d7ee0 --- /dev/null +++ b/examples/responses/conversation.py @@ -0,0 +1,31 @@ +import os + +from openai import OpenAI + +client = OpenAI( + api_key=os.environ.get("TOKEN"), + base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"), +) + +conversation = [] # type: ignore + +# First turn +response1 = client.responses.create( # type: ignore + model="o4-mini", + input="Hi, my name is Alice.", + conversation=conversation, +) +print("Response 1:", response1.output) + +# Note: The 'conversation' parameter might need to be constructed differently +# depending on exact SDK/API spec. Typically, you pass back the previous turn's data. +# Assuming the SDK manages or returns a conversation object/ID: +# conversation.append(response1) + +# Second turn - demonstrating intent, actual implementation depends on strict API spec +# response2 = client.responses.create( +# model="openai/gpt-4o-mini", +# input="What is my name?", +# conversation=conversation, +# ) +# print("Response 2:", response2.output) diff --git a/examples/responses/create.py b/examples/responses/create.py new file mode 100644 index 00000000..8d96ed2d --- /dev/null +++ b/examples/responses/create.py @@ -0,0 +1,17 @@ +import os + +from openai import OpenAI + +# The OpenAI SDK handles the 'responses' endpoint if it's updated to the latest version +# and the base_url points to a compatible proxy like Routstr. +client = OpenAI( + api_key=os.environ.get("TOKEN"), + base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"), +) + +response = client.responses.create( + model="gpt-5-mini", + input="Tell me a three sentence bedtime story about a unicorn.", +) + +print(response.output) diff --git a/examples/responses/streaming_response.py b/examples/responses/streaming_response.py new file mode 100644 index 00000000..ea3b54c5 --- /dev/null +++ b/examples/responses/streaming_response.py @@ -0,0 +1,20 @@ +import os + +from openai import OpenAI + +client = OpenAI( + api_key=os.environ.get("TOKEN"), + base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"), +) + +stream = client.responses.create( + model="claude-4.5-sonnet", + input="Write a short poem about rust.", + stream=True, +) + +for event in stream: + # Note: Depending on the SDK version and response structure, + # you might access event.output_delta or similar fields + print(event, end="", flush=True) +print() diff --git a/examples/responses/web_search.py b/examples/responses/web_search.py new file mode 100644 index 00000000..e7fb548f --- /dev/null +++ b/examples/responses/web_search.py @@ -0,0 +1,16 @@ +import os + +from openai import OpenAI + +client = OpenAI( + api_key=os.environ.get("TOKEN"), + base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"), +) + +response = client.responses.create( + model="gpt-5-mini", + input="What is the latest news about AI?", + tools=[{"type": "web_search"}], # type: ignore +) + +print(response.output) diff --git a/examples/streaming.py b/examples/streaming.py new file mode 100644 index 00000000..3c0bacf2 --- /dev/null +++ b/examples/streaming.py @@ -0,0 +1,28 @@ +import os + +from openai import OpenAI + +client = OpenAI( + api_key=os.environ.get("TOKEN"), + base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"), +) + +messages = [] +while True: + messages.append({"role": "user", "content": input("\nYou: ")}) + + stream = client.chat.completions.create( + model=os.environ.get("MODEL", "gpt-5.1-mini"), + messages=messages, # type: ignore + stream=True, + ) + + print("AI: ", end="") + response_content = "" + for chunk in stream: + if content := chunk.choices[0].delta.content: # type: ignore + print(content, end="", flush=True) + response_content += content + print() + + messages.append({"role": "assistant", "content": response_content}) diff --git a/examples/tor.py b/examples/tor.py new file mode 100644 index 00000000..7e67bb87 --- /dev/null +++ b/examples/tor.py @@ -0,0 +1,20 @@ +import os + +import httpx +from openai import OpenAI + +# Requires `pip install "httpx[socks]"` and a running Tor proxy on port 9050 +client = OpenAI( + api_key=os.environ.get("TOKEN"), + base_url=os.environ.get("ONION_URL", "http://roustrjfsdgfiueghsklchg.onion/v1"), + http_client=httpx.Client(proxies="socks5://localhost:9050"), +) + +print( + client.chat.completions.create( + model="openai/gpt-4o-mini", + messages=[{"role": "user", "content": "Hello from Tor!"}], + ) + .choices[0] + .message.content +) diff --git a/pyproject.toml b/pyproject.toml index a25cbbaf..0aa4e101 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -73,6 +73,7 @@ packages = ["routstr"] [tool.ruff.lint] select = ["E", "F", "I"] ignore = ["E501"] +exclude = ["examples"] [tool.mypy] python_version = "3.11" diff --git a/routstr/payment/cost_calculation.py b/routstr/payment/cost_calculation.py index 03b37362..2a3f9e8a 100644 --- a/routstr/payment/cost_calculation.py +++ b/routstr/payment/cost_calculation.py @@ -191,6 +191,10 @@ async def calculate_cost( # todo: can be sync output_tokens if output_tokens != 0 else usage_data.get("output_tokens", 0) ) + # added for response api + input_tokens = input_tokens if input_tokens != 0 else response_data.get("usage", {}).get("input_tokens", 0) + output_tokens = output_tokens if output_tokens != 0 else response_data.get("usage", {}).get("output_tokens", 0) + input_msats = round(input_tokens / 1000 * MSATS_PER_1K_INPUT_TOKENS, 3) output_msats = round(output_tokens / 1000 * MSATS_PER_1K_OUTPUT_TOKENS, 3) diff --git a/routstr/proxy.py b/routstr/proxy.py index f79fef1c..ee758390 100644 --- a/routstr/proxy.py +++ b/routstr/proxy.py @@ -137,20 +137,14 @@ async def proxy( "unauthorized", "Unauthorized", 401, request=request ) - logger.info( # TODO: move to middleware, async - "Received proxy request", - extra={ - "method": request.method, - "path": path, - "client_host": request.client.host if request.client else "unknown", - "user_agent": request.headers.get("user-agent", "unknown")[:100], - }, - ) - + is_responses_api = path.startswith("v1/responses") or path.startswith("responses") request_body = await request.body() request_body_dict = parse_request_body_json(request_body, path) - model_id = request_body_dict.get("model", "unknown") + if is_responses_api: + model_id = extract_model_from_responses_request(request_body_dict) + else: + model_id = request_body_dict.get("model", "unknown") model_obj = get_model_instance(model_id) if not model_obj: @@ -176,9 +170,14 @@ async def proxy( check_token_balance(headers, request_body_dict, max_cost_for_model) if x_cashu := headers.get("x-cashu", None): - return await upstream.handle_x_cashu( - request, x_cashu, path, max_cost_for_model, model_obj - ) + if is_responses_api: + return await upstream.handle_x_cashu_responses( + request, x_cashu, path, max_cost_for_model, model_obj + ) + else: + return await upstream.handle_x_cashu( + request, x_cashu, path, max_cost_for_model, model_obj + ) elif auth := headers.get("authorization", None): key = await get_bearer_token_key(headers, path, session, auth) @@ -193,28 +192,36 @@ async def proxy( ) logger.debug("Processing unauthenticated GET request", extra={"path": path}) - # TODO: why is this needed? can we remove it? headers = upstream.prepare_headers(dict(request.headers)) return await upstream.forward_get_request(request, path, headers) - # Only pay for request if we have request body data (for completions endpoints) if request_body_dict: await pay_for_request(key, max_cost_for_model, session) - # Prepare headers for upstream headers = upstream.prepare_headers(dict(request.headers)) - # Forward to upstream and handle response - response = await upstream.forward_request( - request, - path, - headers, - request_body, - key, - max_cost_for_model, - session, - model_obj, - ) + if is_responses_api: + response = await upstream.forward_responses_request( + request, + path, + headers, + request_body, + key, + max_cost_for_model, + session, + model_obj, + ) + else: + response = await upstream.forward_request( + request, + path, + headers, + request_body, + key, + max_cost_for_model, + session, + model_obj, + ) if response.status_code != 200: await revert_pay_for_request(key, session, max_cost_for_model) @@ -317,6 +324,24 @@ async def get_bearer_token_key( raise +def extract_model_from_responses_request(request_body_dict: dict[str, Any]) -> str: + if model := request_body_dict.get("model"): + return model + + if input_data := request_body_dict.get("input"): + if isinstance(input_data, dict) and (model := input_data.get("model")): + return model + + if request_body_dict.get("messages"): + return "unknown" + + logger.warning( + "No model found in Responses API request", + extra={"body_keys": list(request_body_dict.keys())} + ) + return "unknown" + + def parse_request_body_json(request_body: bytes, path: str) -> dict[str, Any]: request_body_dict = {} if request_body: diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index ca7505cd..69ef48d9 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -198,6 +198,62 @@ class BaseUpstreamProvider: """ 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: @@ -641,6 +697,310 @@ class BaseUpstreamProvider: ) 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]: + stored_chunks: list[bytes] = [] + usage_finalized: bool = False + last_model_seen: str | None = None + reasoning_tokens: int = 0 + + async def finalize_without_usage() -> bytes | None: + nonlocal usage_finalized + if usage_finalized: + return None + async with create_session() as new_session: + fresh_key = await new_session.get(key.__class__, key.hashed_key) + if not fresh_key: + return None + try: + fallback: dict = { + "model": last_model_seen or "unknown", + "usage": None, + } + cost_data = await adjust_payment_for_tokens( + fresh_key, fallback, new_session, max_cost_for_model + ) + usage_finalized = True + logger.info( + "Finalized Responses API streaming payment without explicit usage", + extra={ + "key_hash": key.hashed_key[:8] + "...", + "cost_data": cost_data, + "balance_after_adjustment": fresh_key.balance, + }, + ) + return f"data: {json.dumps({'cost': cost_data})}\\n\\n".encode() + except Exception as cost_error: + logger.error( + "Error finalizing Responses API payment without usage", + extra={ + "error": str(cost_error), + "error_type": type(cost_error).__name__, + "key_hash": key.hashed_key[:8] + "...", + }, + ) + return None + + try: + async for chunk in response.aiter_bytes(): + stored_chunks.append(chunk) + try: + for part in re.split(b"data: ", chunk): + if not part or part.strip() in (b"[DONE]", b""): + 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) + except json.JSONDecodeError: + pass + except Exception: + pass + + yield chunk + + logger.debug( + "Responses API streaming completed, analyzing usage data", + extra={ + "key_hash": key.hashed_key[:8] + "...", + "chunks_count": len(stored_chunks), + "reasoning_tokens": reasoning_tokens, + }, + ) + + # Process final usage data + for i in range(len(stored_chunks) - 1, -1, -1): + chunk = stored_chunks[i] + if not chunk: + continue + try: + events = re.split(b"data: ", chunk) + for event_data in events: + if not event_data or event_data.strip() in (b"[DONE]", b""): + continue + try: + data = json.loads(event_data) + if isinstance(data, dict) and data.get("model"): + last_model_seen = str(data.get("model")) + if isinstance(data, dict) and isinstance( + data.get("usage"), dict + ): + # Include reasoning tokens in usage calculation + async with create_session() as new_session: + fresh_key = await new_session.get( + key.__class__, key.hashed_key + ) + if fresh_key: + try: + cost_data = ( + await adjust_payment_for_tokens( + fresh_key, + data, + new_session, + max_cost_for_model, + ) + ) + usage_finalized = True + logger.info( + "Payment adjustment completed for Responses API streaming", + extra={ + "key_hash": key.hashed_key[:8] + + "...", + "cost_data": cost_data, + "model": last_model_seen, + "reasoning_tokens": reasoning_tokens, + "balance_after_adjustment": fresh_key.balance, + }, + ) + yield f"data: {json.dumps({'cost': cost_data})}\\n\\n".encode() + except Exception as cost_error: + logger.error( + "Error adjusting payment for Responses API streaming tokens", + extra={ + "error": str(cost_error), + "error_type": type( + cost_error + ).__name__, + "key_hash": key.hashed_key[:8] + + "...", + }, + ) + break + except json.JSONDecodeError: + continue + except Exception as e: + logger.error( + "Error processing Responses API streaming response chunk", + extra={ + "error": str(e), + "error_type": type(e).__name__, + "key_hash": key.hashed_key[:8] + "...", + }, + ) + + if not usage_finalized: + maybe_cost_event = await finalize_without_usage() + if maybe_cost_event is not None: + yield maybe_cost_event + + except Exception as stream_error: + logger.warning( + "Responses API streaming interrupted; finalizing without usage", + extra={ + "error": str(stream_error), + "error_type": type(stream_error).__name__, + "key_hash": key.hashed_key[:8] + "...", + }, + ) + await finalize_without_usage() + raise + + # 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, + }, + ) + + 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 + ) + response_json["cost"] = cost_data + + 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 forward_request( self, request: Request, @@ -865,6 +1225,209 @@ class BaseUpstreamProvider: 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: + 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) + + 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 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] + "...", + }, + ) + + 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}" + + return create_error_response( + "upstream_error", error_message, 502, request=request + ) + + 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, + }, + ) + + return create_error_response( + "internal_error", + "An unexpected server error occurred", + 500, + request=request, + ) + async def forward_get_request( self, request: Request, @@ -1571,6 +2134,585 @@ class BaseUpstreamProvider: 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,