app.ai_providers.claude_provider

  1"""Anthropic Claude AI provider implementation.
  2
  3Uses the ``anthropic`` Python SDK to communicate with the Anthropic
  4Messages API. Supports synchronous and streaming generation, CSV file
  5attachments via content blocks, and automatic token-limit retry.
  6
  7Attributes:
  8    logger: Module-level logger for Claude provider operations.
  9"""
 10
 11from __future__ import annotations
 12
 13import base64
 14import logging
 15from pathlib import Path
 16from typing import Any, Callable, Iterator, Mapping
 17
 18from .base import (
 19    AIProvider,
 20    AIProviderError,
 21    DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS,
 22    DEFAULT_MAX_TOKENS,
 23    _is_attachment_unsupported_error,
 24    _is_anthropic_streaming_required_error,
 25    _is_context_length_error,
 26    _normalize_api_key_value,
 27    _resolve_completion_token_retry_limit,
 28    _resolve_timeout_seconds,
 29    _run_with_rate_limit_retries,
 30    _T,
 31)
 32from .utils import (
 33    _extract_anthropic_stream_text,
 34    _extract_anthropic_text,
 35    _inline_attachment_data_into_prompt,
 36)
 37
 38logger = logging.getLogger(__name__)
 39
 40DEFAULT_CLAUDE_MODEL = "claude-opus-4-6"
 41
 42
 43class ClaudeProvider(AIProvider):
 44    """Anthropic Claude provider implementation.
 45
 46    Supports both synchronous and streaming generation, CSV file attachments
 47    via content blocks (base64-encoded PDFs or inline text), and automatic
 48    token-limit retry when ``max_tokens`` exceeds the model's maximum.
 49
 50    Attributes:
 51        api_key (str): The Anthropic API key.
 52        model (str): The Claude model identifier.
 53        attach_csv_as_file (bool): Whether to upload CSV artifacts as
 54            content blocks.
 55        request_timeout_seconds (float): HTTP timeout in seconds.
 56        client: The ``anthropic.Anthropic`` SDK client instance.
 57    """
 58
 59    def __init__(
 60        self,
 61        api_key: str,
 62        model: str = DEFAULT_CLAUDE_MODEL,
 63        attach_csv_as_file: bool = True,
 64        request_timeout_seconds: float = DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS,
 65    ) -> None:
 66        """Initialize the Claude provider.
 67
 68        Args:
 69            api_key: Anthropic API key. Must be non-empty.
 70            model: Claude model identifier.
 71            attach_csv_as_file: If ``True``, send CSV artifacts as structured
 72                content blocks.
 73            request_timeout_seconds: HTTP timeout in seconds.
 74
 75        Raises:
 76            AIProviderError: If the ``anthropic`` SDK is not installed or
 77                the API key is empty.
 78        """
 79        try:
 80            import anthropic
 81        except ImportError as error:
 82            raise AIProviderError(
 83                "anthropic SDK is not installed. Install it with `pip install anthropic`."
 84            ) from error
 85
 86        normalized_api_key = _normalize_api_key_value(api_key)
 87        if not normalized_api_key:
 88            raise AIProviderError(
 89                "Claude API key is not configured. "
 90                "Set `ai.claude.api_key` in config.yaml or the ANTHROPIC_API_KEY environment variable."
 91            )
 92
 93        self._anthropic = anthropic
 94        self.api_key = normalized_api_key
 95        self.model = model
 96        self.attach_csv_as_file = bool(attach_csv_as_file)
 97        self._csv_attachment_supported: bool | None = None
 98        self.request_timeout_seconds = _resolve_timeout_seconds(
 99            request_timeout_seconds,
100            DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS,
101        )
102        self.client = anthropic.Anthropic(
103            api_key=normalized_api_key,
104            timeout=self.request_timeout_seconds,
105        )
106        logger.info("Initialized Claude provider with model %s (timeout %.1fs)", model, self.request_timeout_seconds)
107
108    def analyze(
109        self,
110        system_prompt: str,
111        user_prompt: str,
112        max_tokens: int = DEFAULT_MAX_TOKENS,
113    ) -> str:
114        """Send a prompt to Claude and return the generated text.
115
116        Args:
117            system_prompt: The system-level instruction text.
118            user_prompt: The user-facing prompt with investigation context.
119            max_tokens: Maximum completion tokens.
120
121        Returns:
122            The generated analysis text.
123
124        Raises:
125            AIProviderError: On any API or network failure.
126        """
127        return self.analyze_with_attachments(
128            system_prompt=system_prompt,
129            user_prompt=user_prompt,
130            attachments=None,
131            max_tokens=max_tokens,
132        )
133
134    def analyze_stream(
135        self,
136        system_prompt: str,
137        user_prompt: str,
138        max_tokens: int = DEFAULT_MAX_TOKENS,
139    ) -> Iterator[str]:
140        """Stream generated text chunks from Claude.
141
142        Args:
143            system_prompt: The system-level instruction text.
144            user_prompt: The user-facing prompt with investigation context.
145            max_tokens: Maximum completion tokens.
146
147        Yields:
148            Text chunk strings as they are generated.
149
150        Raises:
151            AIProviderError: On empty response or API failure.
152        """
153        def _stream() -> Iterator[str]:
154            request_kwargs: dict[str, Any] = {
155                "model": self.model,
156                "max_tokens": max_tokens,
157                "system": system_prompt,
158                "messages": [{"role": "user", "content": user_prompt}],
159                "stream": True,
160            }
161            try:
162                stream = _run_with_rate_limit_retries(
163                    request_fn=lambda: self._with_token_limit_retry(
164                        lambda kw: self.client.messages.create(**kw),
165                        request_kwargs,
166                    ),
167                    rate_limit_error_type=self._anthropic.RateLimitError,
168                    provider_name="Claude",
169                )
170                emitted = False
171                for event in stream:
172                    chunk_text = _extract_anthropic_stream_text(event)
173                    if not chunk_text:
174                        continue
175                    emitted = True
176                    yield chunk_text
177                if not emitted:
178                    raise AIProviderError("Claude returned an empty response.")
179            except AIProviderError:
180                raise
181            except self._anthropic.APIConnectionError as error:
182                raise AIProviderError(
183                    "Unable to connect to Claude API. Check network access and endpoint configuration."
184                ) from error
185            except self._anthropic.AuthenticationError as error:
186                raise AIProviderError(
187                    "Claude authentication failed. Check `ai.claude.api_key` or ANTHROPIC_API_KEY."
188                ) from error
189            except self._anthropic.BadRequestError as error:
190                if _is_context_length_error(error):
191                    raise AIProviderError(
192                        "Claude request exceeded the model context length. Reduce prompt size and retry."
193                    ) from error
194                raise AIProviderError(f"Claude request was rejected: {error}") from error
195            except self._anthropic.APIError as error:
196                raise AIProviderError(f"Claude API error: {error}") from error
197            except Exception as error:
198                raise AIProviderError(f"Unexpected Claude provider error: {error}") from error
199
200        return _stream()
201
202    def analyze_with_attachments(
203        self,
204        system_prompt: str,
205        user_prompt: str,
206        attachments: list[Mapping[str, str]] | None,
207        max_tokens: int = DEFAULT_MAX_TOKENS,
208    ) -> str:
209        """Analyze with optional CSV file attachments via Claude content blocks.
210
211        Args:
212            system_prompt: The system-level instruction text.
213            user_prompt: The user-facing prompt with investigation context.
214            attachments: Optional list of attachment descriptors.
215            max_tokens: Maximum completion tokens.
216
217        Returns:
218            The generated analysis text.
219
220        Raises:
221            AIProviderError: On any API or network failure.
222        """
223        def _request() -> str:
224            attachment_response = self._request_with_csv_attachments(
225                system_prompt=system_prompt,
226                user_prompt=user_prompt,
227                max_tokens=max_tokens,
228                attachments=attachments,
229            )
230            if attachment_response:
231                return attachment_response
232
233            effective_prompt = user_prompt
234            if attachments:
235                effective_prompt, inlined = _inline_attachment_data_into_prompt(
236                    user_prompt=user_prompt,
237                    attachments=attachments,
238                )
239                if inlined:
240                    logger.info("Claude attachment fallback inlined attachment data into prompt.")
241
242            response = self._create_message_with_stream_fallback(
243                system_prompt=system_prompt,
244                messages=[{"role": "user", "content": effective_prompt}],
245                max_tokens=max_tokens,
246            )
247            text = _extract_anthropic_text(response)
248            if not text:
249                raise AIProviderError("Claude returned an empty response.")
250            return text
251
252        return self._run_claude_request(_request)
253
254    def _run_claude_request(self, request_fn: Callable[[], _T]) -> _T:
255        """Execute a Claude request with rate-limit retries and error mapping.
256
257        Args:
258            request_fn: A zero-argument callable that performs the API request.
259
260        Returns:
261            The return value of ``request_fn`` on success.
262
263        Raises:
264            AIProviderError: On any Anthropic SDK error.
265        """
266        try:
267            return _run_with_rate_limit_retries(
268                request_fn=request_fn,
269                rate_limit_error_type=self._anthropic.RateLimitError,
270                provider_name="Claude",
271            )
272        except AIProviderError:
273            raise
274        except self._anthropic.APIConnectionError as error:
275            raise AIProviderError(
276                "Unable to connect to Claude API. Check network access and endpoint configuration."
277            ) from error
278        except self._anthropic.AuthenticationError as error:
279            raise AIProviderError(
280                "Claude authentication failed. Check `ai.claude.api_key` or ANTHROPIC_API_KEY."
281            ) from error
282        except self._anthropic.BadRequestError as error:
283            if _is_context_length_error(error):
284                raise AIProviderError(
285                    "Claude request exceeded the model context length. Reduce prompt size and retry."
286                ) from error
287            raise AIProviderError(f"Claude request was rejected: {error}") from error
288        except self._anthropic.APIError as error:
289            raise AIProviderError(f"Claude API error: {error}") from error
290        except Exception as error:
291            raise AIProviderError(f"Unexpected Claude provider error: {error}") from error
292
293    def _request_with_csv_attachments(
294        self,
295        system_prompt: str,
296        user_prompt: str,
297        max_tokens: int,
298        attachments: list[Mapping[str, str]] | None,
299    ) -> str | None:
300        """Attempt to send a request with CSV files as Claude content blocks.
301
302        Args:
303            system_prompt: The system-level instruction text.
304            user_prompt: The user-facing prompt text.
305            max_tokens: Maximum completion tokens.
306            attachments: Optional list of attachment descriptors.
307
308        Returns:
309            The generated text if attachment mode succeeded, or ``None``
310            if attachments were skipped or unsupported.
311        """
312        normalized_attachments = self._prepare_csv_attachments(attachments)
313        if not normalized_attachments:
314            return None
315
316        try:
317            content_blocks: list[dict[str, Any]] = [{"type": "text", "text": user_prompt}]
318            for attachment in normalized_attachments:
319                attachment_path = Path(attachment["path"])
320                mime_type = attachment["mime_type"].lower()
321                if mime_type == "application/pdf":
322                    encoded_data = base64.b64encode(attachment_path.read_bytes()).decode("ascii")
323                    content_blocks.append(
324                        {
325                            "type": "document",
326                            "source": {
327                                "type": "base64",
328                                "media_type": "application/pdf",
329                                "data": encoded_data,
330                            },
331                        }
332                    )
333                else:
334                    attachment_name = attachment.get("name", attachment_path.name)
335                    try:
336                        attachment_text = attachment_path.read_text(
337                            encoding="utf-8-sig", errors="replace"
338                        )
339                    except OSError:
340                        continue
341                    content_blocks.append(
342                        {
343                            "type": "text",
344                            "text": (
345                                f"--- BEGIN ATTACHMENT: {attachment_name} ---\n"
346                                f"{attachment_text.rstrip()}\n"
347                                f"--- END ATTACHMENT: {attachment_name} ---"
348                            ),
349                        }
350                    )
351
352            response = self._create_message_with_stream_fallback(
353                system_prompt=system_prompt,
354                messages=[{"role": "user", "content": content_blocks}],
355                max_tokens=max_tokens,
356            )
357            text = _extract_anthropic_text(response)
358            if not text:
359                raise AIProviderError("Claude returned an empty response for file-attachment mode.")
360
361            self._csv_attachment_supported = True
362            return text
363        except Exception as error:
364            if _is_attachment_unsupported_error(error):
365                self._csv_attachment_supported = False
366                logger.info(
367                    "Claude endpoint does not support CSV attachments; "
368                    "falling back to standard text mode."
369                )
370                return None
371            raise
372
373    def _create_message_with_stream_fallback(
374        self,
375        system_prompt: str,
376        messages: list[dict[str, Any]],
377        max_tokens: int,
378    ) -> Any:
379        """Create a Claude message, falling back to streaming for long requests.
380
381        Args:
382            system_prompt: The system-level instruction text.
383            messages: The conversation messages list.
384            max_tokens: Maximum completion tokens.
385
386        Returns:
387            The Anthropic ``Message`` response object.
388        """
389        request_kwargs: dict[str, Any] = {
390            "model": self.model,
391            "max_tokens": max_tokens,
392            "system": system_prompt,
393            "messages": messages,
394        }
395        try:
396            return self._with_token_limit_retry(
397                lambda kw: self.client.messages.create(**kw),
398                request_kwargs,
399            )
400        except ValueError as error:
401            if not _is_anthropic_streaming_required_error(error):
402                raise
403            logger.info(
404                "Claude SDK requires streaming for long request; retrying with messages.stream()."
405            )
406            return self._with_token_limit_retry(
407                lambda kw: self._stream_and_collect(**kw),
408                request_kwargs,
409            )
410
411    def _stream_and_collect(self, **kwargs: Any) -> Any:
412        """Stream a Claude request and return the final message.
413
414        Args:
415            **kwargs: Keyword arguments for ``client.messages.stream``.
416
417        Returns:
418            The final Anthropic ``Message`` response object.
419        """
420        with self.client.messages.stream(**kwargs) as stream:
421            return stream.get_final_message()
422
423    def _with_token_limit_retry(
424        self,
425        create_fn: Callable[[dict[str, Any]], Any],
426        request_kwargs: dict[str, Any],
427    ) -> Any:
428        """Execute a Claude API call with automatic token-limit retry.
429
430        If the initial request is rejected because ``max_tokens`` exceeds
431        the model's supported maximum, retries once with the lower limit
432        extracted from the error message.
433
434        This single method replaces the three near-identical retry methods
435        that existed previously.
436
437        Args:
438            create_fn: A callable that takes the request kwargs dict and
439                performs the API call.
440            request_kwargs: Keyword arguments for the API call.
441
442        Returns:
443            The API response object.
444
445        Raises:
446            anthropic.BadRequestError: If the request fails for a reason
447                other than token limits, or if the retry also fails.
448        """
449        effective_kwargs: dict[str, Any] = dict(request_kwargs)
450        for _ in range(2):
451            try:
452                return create_fn(effective_kwargs)
453            except self._anthropic.BadRequestError as error:
454                requested_tokens = int(effective_kwargs.get("max_tokens", 0))
455                retry_token_count = _resolve_completion_token_retry_limit(
456                    error=error,
457                    requested_tokens=requested_tokens,
458                )
459                if retry_token_count is None:
460                    raise
461                logger.warning(
462                    "Claude rejected max_tokens=%d; retrying with max_tokens=%d.",
463                    requested_tokens,
464                    retry_token_count,
465                )
466                effective_kwargs["max_tokens"] = retry_token_count
467        return create_fn(effective_kwargs)
468
469    def get_model_info(self) -> dict[str, str]:
470        """Return Claude provider and model metadata.
471
472        Returns:
473            A dictionary with ``"provider"`` and ``"model"`` keys.
474        """
475        return {"provider": "claude", "model": self.model}