Coverage for src/local_deep_research/llm/providers/implementations/anthropic.py: 99%

1"""Anthropic LLM provider for Local Deep Research.""" 

2 

3from langchain_anthropic import ChatAnthropic 

4from ....security.secure_logging import logger 

5 

6# get_setting_from_snapshot and NoSettingsContextError are imported inside 

7# the methods that use them so test patches at the source module 

8# (`local_deep_research.config.thread_settings`) are picked up by the 

9# function-local imports at call time. 

10from ..base import OPTIONAL_API_KEY_PLACEHOLDER, Exposure 

11from ..openai_base import OpenAICompatibleProvider 

12from ....security.ssrf_validator import assert_base_url_safe 

13 

14 

15class AnthropicProvider(OpenAICompatibleProvider): 

16 """Anthropic provider for Local Deep Research. 

17 

18 This is the official Anthropic API provider. 

19 """ 

20 

21 provider_name = "Anthropic" 

22 api_key_setting = "llm.anthropic.api_key" 

23 default_model = "" # User must explicitly pick a model — no silent fallback 

24 default_base_url = "https://api.anthropic.com/v1" 

25 

26 # Metadata for auto-discovery 

27 provider_key = "ANTHROPIC" 

28 company_name = "Anthropic" 

29 # Annotated so subclasses (e.g. the custom-endpoint provider, which may be 

30 # local or cloud) can set is_cloud = None without a type conflict. 

31 is_cloud: bool | None = True 

32 # Egress exposure (ADR-0007): cloud inference sink — data leaves the box. 

33 egress_exposure = Exposure.EXPOSING 

34 

35 @classmethod 

36 def create_llm(cls, model_name=None, temperature=0.7, **kwargs): 

37 """Factory function for Anthropic LLMs. 

38 

39 Args: 

40 model_name: Name of the model to use 

41 temperature: Model temperature (0.0-1.0) 

42 **kwargs: Additional arguments including settings_snapshot 

43 

44 Returns: 

45 A configured ChatAnthropic instance 

46 

47 Raises: 

48 ValueError: If API key is not configured 

49 """ 

50 from ....config.thread_settings import NoSettingsContextError 

51 

52 settings_snapshot = kwargs.get("settings_snapshot") 

53 

54 # resolve_api_key raises ValueError when the required key is missing 

55 # (preserves the legacy behavior with a unified error message). When 

56 # api_key_optional is True (custom self-hosted endpoints), fall back 

57 # to the shared placeholder instead of raising — and pass it 

58 # explicitly so langchain_anthropic does not silently read a real 

59 # ANTHROPIC_API_KEY from the environment and ship it to the endpoint. 

60 api_key: str | None 

61 if cls.api_key_optional: 

62 api_key = cls.resolve_api_key_or_placeholder(settings_snapshot) 

63 else: 

64 api_key = cls.resolve_api_key(settings_snapshot) 

65 

66 # Require an explicit model — no silent fallback to a hardcoded default. 

67 if not model_name or not model_name.strip(): 

68 logger.error(f"{cls.provider_name} model name not provided") 

69 raise ValueError( 

70 f"{cls.provider_name} model not configured. " 

71 f"Please set llm.model in settings " 

72 f"(e.g. 'claude-3-5-sonnet-20241022')." 

73 ) 

74 

75 # Build Anthropic-specific parameters 

76 anthropic_params = { 

77 "model": model_name, 

78 "anthropic_api_key": api_key, 

79 "temperature": temperature, 

80 } 

81 

82 # Apply context-window-aware max_tokens cap (was previously only 

83 # applied in dead code in llm_config.get_llm). 

84 from .._helpers import ( 

85 compute_max_tokens, 

86 get_context_window_for_provider, 

87 ) 

88 

89 try: 

90 context_window_size = get_context_window_for_provider( 

91 "anthropic", settings_snapshot=settings_snapshot 

92 ) 

93 max_tokens = compute_max_tokens( 

94 settings_snapshot=settings_snapshot, 

95 context_window_size=context_window_size, 

96 ) 

97 if max_tokens: # Treat 0 as unset (matches legacy behavior) 

98 anthropic_params["max_tokens"] = max_tokens 

99 except NoSettingsContextError: 

100 pass # Optional parameter 

101 

102 # Operator-configurable base_url for self-hosted Anthropic-format 

103 # endpoints. No-op for the official cloud provider, whose url_setting 

104 # is None (so it always talks to api.anthropic.com via the SDK 

105 # default). Subclasses like CustomAnthropicEndpointProvider set 

106 # url_setting to opt in. 

107 if cls.url_setting: 

108 from ....config.thread_settings import get_setting_from_snapshot 

109 

110 custom_url = get_setting_from_snapshot( 

111 cls.url_setting, 

112 default=None, 

113 settings_snapshot=settings_snapshot, 

114 ) 

115 custom_url = str(custom_url).strip() if custom_url else "" 

116 if not custom_url: 

117 # No silent fallback to the cloud endpoint — a custom-endpoint 

118 # provider with no URL is a misconfiguration, not a default. 

119 raise ValueError( 

120 f"{cls.provider_name} requires a base URL. " 

121 f"Please set {cls.url_setting} in settings." 

122 ) 

123 # SSRF guard before constructing the client. Cloud-metadata IPs 

124 # stay blocked even under the permissive localhost/private-IP 

125 # posture that legitimate self-hosted endpoints rely on. 

126 anthropic_params["base_url"] = assert_base_url_safe( 

127 custom_url, setting_key=cls.url_setting 

128 ) 

129 

130 logger.info( 

131 f"Creating {cls.provider_name} LLM with model: {model_name}, " 

132 f"temperature: {temperature}" 

133 ) 

134 

135 return ChatAnthropic(**anthropic_params) 

136 

137 @classmethod 

138 def list_models_for_api(cls, api_key=None, base_url=None): 

139 """List models via the anthropic SDK. 

140 

141 Overrides the OpenAICompatibleProvider implementation, which uses the 

142 OpenAI SDK and cannot talk to an Anthropic endpoint — it sends 

143 ``Authorization: Bearer`` while Anthropic requires ``x-api-key``, so 

144 the inherited path returns []. Works for both the official cloud 

145 provider (``base_url`` None → the SDK's api.anthropic.com default) and 

146 the custom-endpoint subclass (``base_url`` from 

147 ``llm.anthropic_endpoint.url``, SSRF-guarded). Degrades to [] on any 

148 failure so model-listing never 500s. 

149 """ 

150 try: 

151 # A URL-based (custom-endpoint) provider with no URL configured has 

152 # no endpoint to query — return [] rather than falling back to the 

153 # cloud default. ``url_setting`` is None for the cloud provider, 

154 # where the cloud default IS the intended target. 

155 if cls.url_setting and not base_url: 

156 return [] 

157 

158 # SSRF-guard an operator-configured base_url (custom-endpoint 

159 # subclass). The cloud provider has url_setting=None and base_url 

160 # None, so it skips this and uses the SDK's cloud default. 

161 if base_url and cls.url_setting: 

162 try: 

163 base_url = assert_base_url_safe( 

164 base_url, setting_key=cls.url_setting 

165 ) 

166 except ValueError: 

167 logger.warning( 

168 f"{cls.provider_name} base_url failed SSRF " 

169 f"validation; check {cls.url_setting} config" 

170 ) 

171 return [] 

172 

173 from anthropic import Anthropic 

174 

175 # Pass the key explicitly (placeholder when keyless) so the SDK does 

176 # not read ANTHROPIC_API_KEY from the environment and ship a real 

177 # cloud key to a self-hosted endpoint. 

178 client = Anthropic( 

179 api_key=api_key or OPTIONAL_API_KEY_PLACEHOLDER, 

180 base_url=base_url or None, 

181 ) 

182 

183 logger.debug(f"Fetching models from {cls.provider_name}") 

184 models_response = client.models.list() 

185 

186 models = [] 

187 for model in models_response.data: 

188 model_id = getattr(model, "id", None) 

189 if model_id: 189 ↛ 187line 189 didn't jump to line 187 because the condition on line 189 was always true

190 label = getattr(model, "display_name", None) or model_id 

191 models.append({"value": model_id, "label": label}) 

192 

193 logger.info(f"Found {len(models)} models from {cls.provider_name}") 

194 return models 

195 

196 except Exception: 

197 # Connection failures are expected when the endpoint isn't running. 

198 logger.warning(f"Could not list models from {cls.provider_name}") 

199 return []