Coverage for apps / ai / services / openrouter.py: 84%

1"""OpenRouter API service using the official SDK.""" 

2 

3import hashlib 

4import json 

5import logging 

6import time 

7from typing import Any 

8 

9from openrouter import OpenRouter 

10 

11from apps.core.models import AppSettings 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16class AIServiceError(Exception): 

17 """Base exception for AI service errors.""" 

18 

19 pass 

20 

21 

22class AIUnavailableError(AIServiceError): 

23 """Raised when AI service is not available (no API key).""" 

24 

25 pass 

26 

27 

28class AIResponseError(AIServiceError): 

29 """Raised when AI returns an invalid or unexpected response.""" 

30 

31 pass 

32 

33 

34class OpenRouterService: 

35 """Service for interacting with OpenRouter API.""" 

36 

37 # Class-level cache for API key validation: {key_hash: (is_valid, timestamp)} 

38 _key_validation_cache: dict[int, tuple[bool, float]] = {} 

39 KEY_VALIDATION_TTL = 300 # 5 minutes 

40 

41 def __init__(self, api_key: str | None = None): 

42 if api_key is None: 

43 settings = AppSettings.get() 

44 api_key = settings.openrouter_api_key 

45 

46 if not api_key: 

47 raise AIUnavailableError("OpenRouter API key not configured") 

48 

49 self.api_key = api_key 

50 

51 def _parse_json_response(self, content: str) -> dict: 

52 """Parse JSON from AI response, handling markdown code blocks.""" 

53 try: 

54 if content.startswith("```"): 

55 lines = content.split("\n") 

56 json_lines = [] 

57 in_block = False 

58 for line in lines: 

59 if line.startswith("```"): 

60 in_block = not in_block 

61 continue 

62 if in_block: 

63 json_lines.append(line) 

64 content = "\n".join(json_lines) 

65 

66 return json.loads(content) 

67 except json.JSONDecodeError as e: 

68 logger.error(f"Failed to parse AI response as JSON: {content}") 

69 raise AIResponseError(f"Invalid JSON in AI response: {e}") 

70 

71 def complete( 

72 self, 

73 system_prompt: str, 

74 user_prompt: str, 

75 model: str = "anthropic/claude-haiku-4.5", 

76 json_response: bool = True, 

77 timeout: int = 30, 

78 ) -> dict[str, Any]: 

79 """Send a completion request to OpenRouter.""" 

80 messages = [ 

81 {"role": "system", "content": system_prompt}, 

82 {"role": "user", "content": user_prompt}, 

83 ] 

84 

85 timeout_ms = timeout * 1000 

86 

87 try: 

88 with OpenRouter(api_key=self.api_key) as client: 

89 response = client.chat.send( 

90 messages=messages, 

91 model=model, 

92 stream=False, 

93 timeout_ms=timeout_ms, 

94 ) 

95 

96 if not response or not hasattr(response, "choices"): 

97 raise AIResponseError("Invalid response structure from OpenRouter") 

98 

99 if not response.choices: 

100 raise AIResponseError("No choices in OpenRouter response") 

101 

102 content = response.choices[0].message.content 

103 

104 if json_response: 

105 return self._parse_json_response(content) 

106 

107 return {"content": content} 

108 

109 except AIServiceError: 

110 raise 

111 except Exception as e: 

112 logger.exception("OpenRouter API error") 

113 raise AIResponseError(f"OpenRouter API error: {e}") 

114 

115 async def complete_async( 

116 self, 

117 system_prompt: str, 

118 user_prompt: str, 

119 model: str = "anthropic/claude-haiku-4.5", 

120 json_response: bool = True, 

121 timeout: int = 30, 

122 ) -> dict[str, Any]: 

123 """Async version of complete().""" 

124 messages = [ 

125 {"role": "system", "content": system_prompt}, 

126 {"role": "user", "content": user_prompt}, 

127 ] 

128 

129 timeout_ms = timeout * 1000 

130 

131 try: 

132 async with OpenRouter(api_key=self.api_key) as client: 

133 response = await client.chat.send_async( 

134 messages=messages, 

135 model=model, 

136 stream=False, 

137 timeout_ms=timeout_ms, 

138 ) 

139 

140 if not response or not hasattr(response, "choices"): 

141 raise AIResponseError("Invalid response structure from OpenRouter") 

142 

143 if not response.choices: 

144 raise AIResponseError("No choices in OpenRouter response") 

145 

146 content = response.choices[0].message.content 

147 

148 if json_response: 

149 return self._parse_json_response(content) 

150 

151 return {"content": content} 

152 

153 except AIServiceError: 

154 raise 

155 except Exception as e: 

156 logger.exception("OpenRouter API error") 

157 raise AIResponseError(f"OpenRouter API error: {e}") 

158 

159 @classmethod 

160 def is_available(cls) -> bool: 

161 """Check if AI service is available (API key configured).""" 

162 settings = AppSettings.get() 

163 return bool(settings.openrouter_api_key) 

164 

165 def get_available_models(self) -> list[dict[str, str]]: 

166 """Get list of available models from OpenRouter.""" 

167 try: 

168 with OpenRouter(api_key=self.api_key) as client: 

169 response = client.models.list() 

170 

171 if not response or not hasattr(response, "data"): 

172 raise AIResponseError("Invalid response from OpenRouter models API") 

173 

174 models = [ 

175 {"id": model.id, "name": model.name} 

176 for model in response.data 

177 if hasattr(model, "id") and hasattr(model, "name") 

178 ] 

179 return sorted(models, key=lambda m: m["name"].lower()) 

180 except AIServiceError: 

181 raise 

182 except Exception as e: 

183 logger.exception("Failed to fetch OpenRouter models") 

184 raise AIResponseError(f"Failed to fetch available models: {e}") 

185 

186 @classmethod 

187 def test_connection(cls, api_key: str) -> tuple[bool, str]: 

188 """Test if an API key is valid by making a minimal request.""" 

189 try: 

190 service = cls(api_key=api_key) 

191 service.complete( 

192 system_prompt='Respond with exactly: {"status": "ok"}', 

193 user_prompt="Test connection", 

194 model="anthropic/claude-3.5-haiku", 

195 json_response=True, 

196 timeout=10, 

197 ) 

198 return True, "Connection successful" 

199 except AIUnavailableError: 

200 return False, "API key not provided" 

201 except AIResponseError as e: 

202 return False, f"API error: {e}" 

203 except Exception as e: 

204 return False, f"Connection failed: {e}" 

205 

206 @classmethod 

207 def validate_key_cached(cls, api_key: str | None = None) -> tuple[bool, str | None]: 

208 """Validate API key with caching to avoid excessive API calls.""" 

209 if api_key is None: 

210 settings = AppSettings.get() 

211 api_key = settings.openrouter_api_key 

212 

213 if not api_key: 

214 return False, "No API key configured" 

215 

216 # SHA-256 used as a cache lookup key (not for password storage) 

217 cache_id = hashlib.sha256(api_key.encode()).hexdigest() # nosec 

218 now = time.time() 

219 

220 if cache_id in cls._key_validation_cache: 

221 is_valid, timestamp = cls._key_validation_cache[cache_id] 

222 if now - timestamp < cls.KEY_VALIDATION_TTL: 

223 return is_valid, None if is_valid else "API key is invalid or expired" 

224 

225 try: 

226 is_valid, message = cls.test_connection(api_key) 

227 cls._key_validation_cache[cache_id] = (is_valid, now) 

228 return is_valid, None if is_valid else message 

229 except Exception as e: 

230 logger.exception("Failed to validate API key") 

231 return False, f"Unable to verify API key: {e}" 

232 

233 @classmethod 

234 def invalidate_key_cache(cls): 

235 """Clear validation cache (call when key is updated).""" 

236 cls._key_validation_cache.clear()