Coverage for session_buddy / llm / providers / anthropic_provider.py: 27.40%

1"""Anthropic API provider implementation (Claude models). 

2 

3Uses anthropic.AsyncAnthropic client. Kept optional; if the package or 

4API key is unavailable, the provider reports as unavailable. 

5""" 

6 

7from __future__ import annotations 

8 

9from datetime import datetime 

10from typing import TYPE_CHECKING, Any 

11 

12from session_buddy.llm.base import LLMProvider 

13from session_buddy.llm.models import LLMMessage, LLMResponse 

14 

15if TYPE_CHECKING: 

16 from collections.abc import AsyncGenerator 

17 

18 

19class AnthropicProvider(LLMProvider): 

20 """Anthropic Claude API provider.""" 

21 

22 def __init__(self, config: dict[str, Any]) -> None: 

23 super().__init__(config) 

24 self.api_key = config.get("api_key") 

25 self.default_model = config.get("default_model", "claude-3-5-haiku-20241022") 

26 self._client: Any = None 

27 

28 async def _get_client(self) -> Any: 

29 if self._client is None: 

30 try: 

31 import anthropic 

32 

33 self._client = anthropic.AsyncAnthropic(api_key=self.api_key) 

34 except ImportError: # pragma: no cover - optional dependency 

35 msg = "Anthropic package not installed. Install with: pip install anthropic" 

36 raise ImportError(msg) 

37 return self._client 

38 

39 def _strip_thinking_blocks(self, content: str) -> str: 

40 """Remove thinking blocks from content before sending to API. 

41 

42 Anthropic API does not accept thinking blocks in request messages. 

43 They can only appear in responses from the API. 

44 """ 

45 import re 

46 

47 # Remove all <thinking>...</thinking> blocks (with any attributes) 

48 pattern = r"<thinking[^>]*>.*?</thinking>" 

49 cleaned = re.sub(pattern, "", content, flags=re.DOTALL | re.IGNORECASE) 

50 return cleaned.strip() 

51 

52 def _convert_messages(self, messages: list[LLMMessage]) -> list[dict[str, Any]]: 

53 """Convert to Anthropic messages format. 

54 

55 - Maps 'system' into top-level system field (handled in generate) 

56 - Converts user/assistant into human/assistant messages 

57 - Strips thinking blocks from assistant messages (not allowed in API requests) 

58 """ 

59 converted: list[dict[str, Any]] = [] 

60 for msg in messages: 

61 if msg.role == "user": 

62 converted.append({"role": "user", "content": msg.content}) 

63 elif msg.role == "assistant": 

64 # Remove thinking blocks - they cannot be in API requests 

65 cleaned_content = self._strip_thinking_blocks(msg.content) 

66 if cleaned_content: # Only add if there's content left after stripping 

67 converted.append({"role": "assistant", "content": cleaned_content}) 

68 # 'system' is handled separately 

69 return converted 

70 

71 async def generate( 

72 self, 

73 messages: list[LLMMessage], 

74 model: str | None = None, 

75 temperature: float = 0.7, 

76 max_tokens: int | None = None, 

77 **kwargs: Any, 

78 ) -> LLMResponse: 

79 if not await self.is_available(): 

80 msg = "Anthropic provider not available" 

81 raise RuntimeError(msg) 

82 

83 client = await self._get_client() 

84 model_name = model or self.default_model 

85 

86 # Extract a system prompt if present 

87 system_parts = [m.content for m in messages if m.role == "system"] 

88 system_prompt = "\n\n".join(system_parts) if system_parts else None 

89 converted = self._convert_messages(messages) 

90 

91 try: 

92 resp = await client.messages.create( 

93 model=model_name, 

94 system=system_prompt, 

95 messages=converted, 

96 temperature=temperature, 

97 max_tokens=max_tokens or 1024, 

98 ) 

99 

100 text = "".join( 

101 [ 

102 block.text 

103 for block in resp.content 

104 if hasattr(block, "text") and isinstance(block.text, str) 

105 ] 

106 ) 

107 usage = getattr(resp, "usage", None) 

108 return LLMResponse( 

109 content=text, 

110 model=model_name, 

111 provider="anthropic", 

112 usage={ 

113 "prompt_tokens": getattr(usage, "input_tokens", 0) if usage else 0, 

114 "completion_tokens": getattr(usage, "output_tokens", 0) 

115 if usage 

116 else 0, 

117 "total_tokens": ( 

118 getattr(usage, "input_tokens", 0) 

119 + getattr(usage, "output_tokens", 0) 

120 if usage 

121 else 0 

122 ), 

123 }, 

124 finish_reason="stop", 

125 timestamp=datetime.now().isoformat(), 

126 ) 

127 except Exception as e: 

128 self.logger.exception(f"Anthropic generation failed: {e}") 

129 raise 

130 

131 async def stream_generate( # type: ignore[override] 

132 self, 

133 messages: list[LLMMessage], 

134 model: str | None = None, 

135 temperature: float = 0.7, 

136 max_tokens: int | None = None, 

137 **kwargs: Any, 

138 ) -> AsyncGenerator[str]: 

139 # Streaming not essential for extraction; implement later as needed 

140 raise NotImplementedError 

141 

142 async def is_available(self) -> bool: 

143 if not self.api_key: 143 ↛ 145line 143 didn't jump to line 145 because the condition on line 143 was always true

144 return False 

145 try: 

146 await self._get_client() 

147 return True 

148 except Exception: 

149 return False 

150 

151 def get_models(self) -> list[str]: 

152 return [ 

153 "claude-3-5-haiku-20241022", 

154 "claude-3-5-sonnet-20241022", 

155 "claude-3-opus-20240229", 

156 ]