Spaces:
Configuration error
Configuration error
File size: 2,714 Bytes
aa15bce |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
from __future__ import annotations
import json
from typing import Any, Dict, List, Optional
import httpx
from ..config import get_settings
class OpenRouterError(RuntimeError):
"""Raised when the API returns an error response."""
def _headers(*, api_key: Optional[str] = None) -> Dict[str, str]:
settings = get_settings()
key = (api_key or settings.api_key or "").strip()
if not key:
raise OpenRouterError("Missing API key")
headers = {
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
"Accept": "application/json",
}
return headers
def _build_messages(messages: List[Dict[str, str]], system: Optional[str]) -> List[Dict[str, str]]:
if system:
return [{"role": "system", "content": system}, *messages]
return messages
def _handle_response_error(exc: httpx.HTTPStatusError) -> None:
response = exc.response
detail: str
try:
payload = response.json()
detail = payload.get("error") or payload.get("message") or json.dumps(payload)
except Exception:
detail = response.text
raise OpenRouterError(f"API request failed ({response.status_code}): {detail}") from exc
async def request_chat_completion(
*,
model: str,
messages: List[Dict[str, str]],
system: Optional[str] = None,
api_key: Optional[str] = None,
tools: Optional[List[Dict[str, Any]]] = None,
base_url: Optional[str] = None,
) -> Dict[str, Any]:
"""Request a chat completion and return the raw JSON payload."""
settings = get_settings()
base_url = base_url or settings.api_base_url
payload: Dict[str, object] = {
"model": model,
"messages": _build_messages(messages, system),
"stream": False,
}
if tools:
payload["tools"] = tools
url = f"{base_url.rstrip('/')}/chat/completions"
async with httpx.AsyncClient() as client:
try:
response = await client.post(
url,
headers=_headers(api_key=api_key),
json=payload,
timeout=60.0, # Set reasonable timeout instead of None
)
try:
response.raise_for_status()
except httpx.HTTPStatusError as exc:
_handle_response_error(exc)
return response.json()
except httpx.HTTPStatusError as exc: # pragma: no cover - handled above
_handle_response_error(exc)
except httpx.HTTPError as exc:
raise OpenRouterError(f"API request failed: {exc}") from exc
raise OpenRouterError("API request failed: unknown error")
__all__ = ["OpenRouterError", "request_chat_completion"]
|