| 1 | """Anthropic provider — direct SDK integration for Claude models.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import asyncio |
| 6 | import os |
| 7 | import re |
| 8 | import secrets |
| 9 | import string |
| 10 | from collections.abc import Awaitable, Callable |
| 11 | from typing import Any |
| 12 | |
| 13 | import json_repair |
| 14 | |
| 15 | from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest |
| 16 | |
| 17 | _ALNUM = string.ascii_letters + string.digits |
| 18 | |
| 19 | |
| 20 | def _gen_tool_id() -> str: |
| 21 | return "toolu_" + "".join(secrets.choice(_ALNUM) for _ in range(22)) |
| 22 | |
| 23 | |
| 24 | class AnthropicProvider(LLMProvider): |
| 25 | """LLM provider using the native Anthropic SDK for Claude models. |
| 26 | |
| 27 | Handles message format conversion (OpenAI → Anthropic Messages API), |
| 28 | prompt caching, extended thinking, tool calls, and streaming. |
| 29 | """ |
| 30 | |
| 31 | def __init__( |
| 32 | self, |
| 33 | api_key: str | None = None, |
| 34 | api_base: str | None = None, |
| 35 | default_model: str = "claude-sonnet-4-20250514", |
| 36 | extra_headers: dict[str, str] | None = None, |
| 37 | ): |
| 38 | super().__init__(api_key, api_base) |
| 39 | self.default_model = default_model |
| 40 | self.extra_headers = extra_headers or {} |
| 41 | |
| 42 | from anthropic import AsyncAnthropic |
| 43 | |
| 44 | client_kw: dict[str, Any] = {} |
| 45 | if api_key: |
| 46 | client_kw["api_key"] = api_key |
| 47 | if api_base: |
| 48 | client_kw["base_url"] = api_base |
| 49 | if extra_headers: |
| 50 | client_kw["default_headers"] = extra_headers |
| 51 | # Keep retries centralized in LLMProvider._run_with_retry to avoid retry amplification. |
| 52 | client_kw["max_retries"] = 0 |
| 53 | self._client = AsyncAnthropic(**client_kw) |
| 54 | |
| 55 | @classmethod |
| 56 | def _handle_error(cls, e: Exception) -> LLMResponse: |
| 57 | response = getattr(e, "response", None) |
| 58 | headers = getattr(response, "headers", None) |
| 59 | payload = ( |
| 60 | getattr(e, "body", None) |
| 61 | or getattr(e, "doc", None) |
| 62 | or getattr(response, "text", None) |
| 63 | ) |
| 64 | if payload is None and response is not None: |
| 65 | response_json = getattr(response, "json", None) |
| 66 | if callable(response_json): |
| 67 | try: |
| 68 | payload = response_json() |
| 69 | except Exception: |
| 70 | payload = None |
| 71 | payload_text = payload if isinstance(payload, str) else str(payload) if payload is not None else "" |
| 72 | msg = f"Error: {payload_text.strip()[:500]}" if payload_text.strip() else f"Error calling LLM: {e}" |
| 73 | retry_after = cls._extract_retry_after_from_headers(headers) |
| 74 | if retry_after is None: |
| 75 | retry_after = LLMProvider._extract_retry_after(msg) |
| 76 | |
| 77 | status_code = getattr(e, "status_code", None) |
| 78 | if status_code is None and response is not None: |
| 79 | status_code = getattr(response, "status_code", None) |
| 80 | |
| 81 | should_retry: bool | None = None |
| 82 | if headers is not None: |
| 83 | raw = headers.get("x-should-retry") |
| 84 | if isinstance(raw, str): |
| 85 | lowered = raw.strip().lower() |
| 86 | if lowered == "true": |
| 87 | should_retry = True |
| 88 | elif lowered == "false": |
| 89 | should_retry = False |
| 90 | |
| 91 | error_kind: str | None = None |
| 92 | error_name = e.__class__.__name__.lower() |
| 93 | if "timeout" in error_name: |
| 94 | error_kind = "timeout" |
| 95 | elif "connection" in error_name: |
| 96 | error_kind = "connection" |
| 97 | error_type, error_code = LLMProvider._extract_error_type_code(payload) |
| 98 | |
| 99 | return LLMResponse( |
| 100 | content=msg, |
| 101 | finish_reason="error", |
| 102 | retry_after=retry_after, |
| 103 | error_status_code=int(status_code) if status_code is not None else None, |
| 104 | error_kind=error_kind, |
| 105 | error_type=error_type, |
| 106 | error_code=error_code, |
| 107 | error_retry_after_s=retry_after, |
| 108 | error_should_retry=should_retry, |
| 109 | ) |
| 110 | |
| 111 | @staticmethod |
| 112 | def _strip_prefix(model: str) -> str: |
| 113 | if model.startswith("anthropic/"): |
| 114 | return model[len("anthropic/"):] |
| 115 | return model |
| 116 | |
| 117 | # ------------------------------------------------------------------ |
| 118 | # Message conversion: OpenAI chat format → Anthropic Messages API |
| 119 | # ------------------------------------------------------------------ |
| 120 | |
| 121 | def _convert_messages( |
| 122 | self, messages: list[dict[str, Any]], |
| 123 | ) -> tuple[str | list[dict[str, Any]], list[dict[str, Any]]]: |
| 124 | """Return ``(system, anthropic_messages)``.""" |
| 125 | system: str | list[dict[str, Any]] = "" |
| 126 | raw: list[dict[str, Any]] = [] |
| 127 | |
| 128 | for msg in messages: |
| 129 | role = msg.get("role", "") |
| 130 | content = msg.get("content") |
| 131 | |
| 132 | if role == "system": |
| 133 | system = content if isinstance(content, (str, list)) else str(content or "") |
| 134 | continue |
| 135 | |
| 136 | if role == "tool": |
| 137 | block = self._tool_result_block(msg) |
| 138 | if raw and raw[-1]["role"] == "user": |
| 139 | prev_c = raw[-1]["content"] |
| 140 | if isinstance(prev_c, list): |
| 141 | prev_c.append(block) |
| 142 | else: |
| 143 | raw[-1]["content"] = [ |
| 144 | {"type": "text", "text": prev_c or ""}, block, |
| 145 | ] |
| 146 | else: |
| 147 | raw.append({"role": "user", "content": [block]}) |
| 148 | continue |
| 149 | |
| 150 | if role == "assistant": |
| 151 | raw.append({"role": "assistant", "content": self._assistant_blocks(msg)}) |
| 152 | continue |
| 153 | |
| 154 | if role == "user": |
| 155 | raw.append({ |
| 156 | "role": "user", |
| 157 | "content": self._convert_user_content(content), |
| 158 | }) |
| 159 | continue |
| 160 | |
| 161 | return system, self._merge_consecutive(raw) |
| 162 | |
| 163 | @staticmethod |
| 164 | def _tool_result_block(msg: dict[str, Any]) -> dict[str, Any]: |
| 165 | content = msg.get("content") |
| 166 | block: dict[str, Any] = { |
| 167 | "type": "tool_result", |
| 168 | "tool_use_id": msg.get("tool_call_id", ""), |
| 169 | } |
| 170 | if isinstance(content, list): |
| 171 | block["content"] = AnthropicProvider._convert_user_content(content) |
| 172 | elif isinstance(content, str): |
| 173 | block["content"] = content |
| 174 | else: |
| 175 | block["content"] = str(content) if content else "" |
| 176 | return block |
| 177 | |
| 178 | @staticmethod |
| 179 | def _assistant_blocks(msg: dict[str, Any]) -> list[dict[str, Any]]: |
| 180 | blocks: list[dict[str, Any]] = [] |
| 181 | content = msg.get("content") |
| 182 | |
| 183 | for tb in msg.get("thinking_blocks") or []: |
| 184 | if isinstance(tb, dict) and tb.get("type") == "thinking": |
| 185 | blocks.append({ |
| 186 | "type": "thinking", |
| 187 | "thinking": tb.get("thinking", ""), |
| 188 | "signature": tb.get("signature", ""), |
| 189 | }) |
| 190 | |
| 191 | if isinstance(content, str) and content: |
| 192 | blocks.append({"type": "text", "text": content}) |
| 193 | elif isinstance(content, list): |
| 194 | for item in content: |
| 195 | blocks.append(item if isinstance(item, dict) else {"type": "text", "text": str(item)}) |
| 196 | |
| 197 | for tc in msg.get("tool_calls") or []: |
| 198 | if not isinstance(tc, dict): |
| 199 | continue |
| 200 | func = tc.get("function", {}) |
| 201 | args = func.get("arguments", "{}") |
| 202 | if isinstance(args, str): |
| 203 | args = json_repair.loads(args) |
| 204 | blocks.append({ |
| 205 | "type": "tool_use", |
| 206 | "id": tc.get("id") or _gen_tool_id(), |
| 207 | "name": func.get("name", ""), |
| 208 | "input": args, |
| 209 | }) |
| 210 | |
| 211 | return blocks or [{"type": "text", "text": ""}] |
| 212 | |
| 213 | @staticmethod |
| 214 | def _convert_user_content(content: Any) -> Any: |
| 215 | """Convert user message content, translating image_url blocks.""" |
| 216 | if isinstance(content, str) or content is None: |
| 217 | return content or "(empty)" |
| 218 | if not isinstance(content, list): |
| 219 | return str(content) |
| 220 | |
| 221 | result: list[dict[str, Any]] = [] |
| 222 | for item in content: |
| 223 | if not isinstance(item, dict): |
| 224 | result.append({"type": "text", "text": str(item)}) |
| 225 | continue |
| 226 | if item.get("type") == "image_url": |
| 227 | converted = AnthropicProvider._convert_image_block(item) |
| 228 | if converted: |
| 229 | result.append(converted) |
| 230 | continue |
| 231 | result.append(item) |
| 232 | return result or "(empty)" |
| 233 | |
| 234 | @staticmethod |
| 235 | def _convert_image_block(block: dict[str, Any]) -> dict[str, Any] | None: |
| 236 | """Convert OpenAI image_url block to Anthropic image block.""" |
| 237 | url = (block.get("image_url") or {}).get("url", "") |
| 238 | if not url: |
| 239 | return None |
| 240 | m = re.match(r"data:(image/\w+);base64,(.+)", url, re.DOTALL) |
| 241 | if m: |
| 242 | return { |
| 243 | "type": "image", |
| 244 | "source": {"type": "base64", "media_type": m.group(1), "data": m.group(2)}, |
| 245 | } |
| 246 | return { |
| 247 | "type": "image", |
| 248 | "source": {"type": "url", "url": url}, |
| 249 | } |
| 250 | |
| 251 | @staticmethod |
| 252 | def _has_tool_use(msg: dict[str, Any]) -> bool: |
| 253 | """True if ``msg.content`` carries any ``tool_use`` block. |
| 254 | |
| 255 | Anthropic forbids ``tool_use`` inside ``user`` turns, so messages that |
| 256 | issued a tool call cannot be safely rerouted when we patch the role. |
| 257 | """ |
| 258 | content = msg.get("content") |
| 259 | if not isinstance(content, list): |
| 260 | return False |
| 261 | return any( |
| 262 | isinstance(block, dict) and block.get("type") == "tool_use" |
| 263 | for block in content |
| 264 | ) |
| 265 | |
| 266 | @staticmethod |
| 267 | def _merge_consecutive(msgs: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| 268 | """Normalize a message sequence for Anthropic's ``/messages`` endpoint. |
| 269 | |
| 270 | Anthropic's contract is stricter than OpenAI's: |
| 271 | |
| 272 | 1. Consecutive same-role turns must be collapsed into one. |
| 273 | 2. The conversation cannot end with an ``assistant`` turn — Anthropic |
| 274 | does not support assistant-message prefill and returns 400. |
| 275 | 3. The conversation cannot start with an ``assistant`` turn — the |
| 276 | first message must be ``user``. |
| 277 | |
| 278 | Rules 2 and 3 mirror ``LLMProvider._enforce_role_alternation`` in |
| 279 | ``base.py``, which applies the equivalent invariants to OpenAI-compat |
| 280 | providers. The only Anthropic-specific wrinkle: ``tool_use`` blocks |
| 281 | live inside ``content`` (not a separate ``tool_calls`` field) and are |
| 282 | invalid inside ``user`` turns, so the recovery paths below must skip |
| 283 | any message carrying them rather than silently producing a malformed |
| 284 | request. |
| 285 | """ |
| 286 | merged: list[dict[str, Any]] = [] |
| 287 | for msg in msgs: |
| 288 | if merged and merged[-1]["role"] == msg["role"]: |
| 289 | prev_c = merged[-1]["content"] |
| 290 | cur_c = msg["content"] |
| 291 | if isinstance(prev_c, str): |
| 292 | prev_c = [{"type": "text", "text": prev_c}] |
| 293 | if isinstance(cur_c, str): |
| 294 | cur_c = [{"type": "text", "text": cur_c}] |
| 295 | if isinstance(cur_c, list): |
| 296 | prev_c.extend(cur_c) |
| 297 | merged[-1]["content"] = prev_c |
| 298 | else: |
| 299 | merged.append(msg) |
| 300 | |
| 301 | # Rule 2: strip trailing assistant turns — Anthropic rejects prefill. |
| 302 | last_popped: dict[str, Any] | None = None |
| 303 | while merged and merged[-1].get("role") == "assistant": |
| 304 | last_popped = merged.pop() |
| 305 | |
| 306 | # Recovery for rule 2: if stripping removed every turn, reroute the |
| 307 | # last popped assistant as a user turn so upstream code still gets a |
| 308 | # valid request instead of a secondary "messages array empty" 400. |
| 309 | # Skip when the message carried ``tool_use`` blocks (see _has_tool_use). |
| 310 | if ( |
| 311 | not merged |
| 312 | and last_popped is not None |
| 313 | and not AnthropicProvider._has_tool_use(last_popped) |
| 314 | ): |
| 315 | merged.append({"role": "user", "content": last_popped.get("content")}) |
| 316 | |
| 317 | # Rule 3: prepend a synthetic opener if the first surviving turn is an |
| 318 | # assistant (e.g. upstream history truncation dropped the original |
| 319 | # user request). ``tool_use``-carrying assistants are left alone — |
| 320 | # that message will still fail validation, but injecting an opener |
| 321 | # before it would orphan the tool_use/tool_result pair that follows, |
| 322 | # turning a recoverable 400 into a harder-to-diagnose one. |
| 323 | if ( |
| 324 | merged |
| 325 | and merged[0].get("role") == "assistant" |
| 326 | and not AnthropicProvider._has_tool_use(merged[0]) |
| 327 | ): |
| 328 | merged.insert(0, {"role": "user", "content": "(conversation continued)"}) |
| 329 | |
| 330 | return merged |
| 331 | |
| 332 | # ------------------------------------------------------------------ |
| 333 | # Tool definition conversion |
| 334 | # ------------------------------------------------------------------ |
| 335 | |
| 336 | @staticmethod |
| 337 | def _convert_tools(tools: list[dict[str, Any]] | None) -> list[dict[str, Any]] | None: |
| 338 | if not tools: |
| 339 | return None |
| 340 | result = [] |
| 341 | for tool in tools: |
| 342 | func = tool.get("function", tool) |
| 343 | entry: dict[str, Any] = { |
| 344 | "name": func.get("name", ""), |
| 345 | "input_schema": func.get("parameters", {"type": "object", "properties": {}}), |
| 346 | } |
| 347 | desc = func.get("description") |
| 348 | if desc: |
| 349 | entry["description"] = desc |
| 350 | if "cache_control" in tool: |
| 351 | entry["cache_control"] = tool["cache_control"] |
| 352 | result.append(entry) |
| 353 | return result |
| 354 | |
| 355 | @staticmethod |
| 356 | def _convert_tool_choice( |
| 357 | tool_choice: str | dict[str, Any] | None, |
| 358 | thinking_enabled: bool = False, |
| 359 | ) -> dict[str, Any] | None: |
| 360 | if thinking_enabled: |
| 361 | return {"type": "auto"} |
| 362 | if tool_choice is None or tool_choice == "auto": |
| 363 | return {"type": "auto"} |
| 364 | if tool_choice == "required": |
| 365 | return {"type": "any"} |
| 366 | if tool_choice == "none": |
| 367 | return None |
| 368 | if isinstance(tool_choice, dict): |
| 369 | name = tool_choice.get("function", {}).get("name") |
| 370 | if name: |
| 371 | return {"type": "tool", "name": name} |
| 372 | return {"type": "auto"} |
| 373 | |
| 374 | # ------------------------------------------------------------------ |
| 375 | # Prompt caching |
| 376 | # ------------------------------------------------------------------ |
| 377 | |
| 378 | @classmethod |
| 379 | def _apply_cache_control( |
| 380 | cls, |
| 381 | system: str | list[dict[str, Any]], |
| 382 | messages: list[dict[str, Any]], |
| 383 | tools: list[dict[str, Any]] | None, |
| 384 | ) -> tuple[str | list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]] | None]: |
| 385 | marker = {"type": "ephemeral"} |
| 386 | |
| 387 | if isinstance(system, str) and system: |
| 388 | system = [{"type": "text", "text": system, "cache_control": marker}] |
| 389 | elif isinstance(system, list) and system: |
| 390 | system = list(system) |
| 391 | system[-1] = {**system[-1], "cache_control": marker} |
| 392 | |
| 393 | new_msgs = list(messages) |
| 394 | if len(new_msgs) >= 3: |
| 395 | m = new_msgs[-2] |
| 396 | c = m.get("content") |
| 397 | if isinstance(c, str): |
| 398 | new_msgs[-2] = {**m, "content": [{"type": "text", "text": c, "cache_control": marker}]} |
| 399 | elif isinstance(c, list) and c: |
| 400 | nc = list(c) |
| 401 | nc[-1] = {**nc[-1], "cache_control": marker} |
| 402 | new_msgs[-2] = {**m, "content": nc} |
| 403 | |
| 404 | new_tools = tools |
| 405 | if tools: |
| 406 | new_tools = list(tools) |
| 407 | for idx in cls._tool_cache_marker_indices(new_tools): |
| 408 | new_tools[idx] = {**new_tools[idx], "cache_control": marker} |
| 409 | |
| 410 | return system, new_msgs, new_tools |
| 411 | |
| 412 | # ------------------------------------------------------------------ |
| 413 | # Build API kwargs |
| 414 | # ------------------------------------------------------------------ |
| 415 | |
| 416 | def _build_kwargs( |
| 417 | self, |
| 418 | messages: list[dict[str, Any]], |
| 419 | tools: list[dict[str, Any]] | None, |
| 420 | model: str | None, |
| 421 | max_tokens: int, |
| 422 | temperature: float, |
| 423 | reasoning_effort: str | None, |
| 424 | tool_choice: str | dict[str, Any] | None, |
| 425 | supports_caching: bool = True, |
| 426 | ) -> dict[str, Any]: |
| 427 | model_name = self._strip_prefix(model or self.default_model) |
| 428 | system, anthropic_msgs = self._convert_messages(self._sanitize_empty_content(messages)) |
| 429 | anthropic_tools = self._convert_tools(tools) |
| 430 | |
| 431 | if supports_caching: |
| 432 | system, anthropic_msgs, anthropic_tools = self._apply_cache_control( |
| 433 | system, anthropic_msgs, anthropic_tools, |
| 434 | ) |
| 435 | |
| 436 | max_tokens = max(1, max_tokens) |
| 437 | thinking_enabled = bool(reasoning_effort) |
| 438 | |
| 439 | # claude-opus-4-7 deprecated the `temperature` parameter entirely — the |
| 440 | # API returns 400 if it is present, on any code path. |
| 441 | omit_temperature = "opus-4-7" in model_name |
| 442 | |
| 443 | kwargs: dict[str, Any] = { |
| 444 | "model": model_name, |
| 445 | "messages": anthropic_msgs, |
| 446 | "max_tokens": max_tokens, |
| 447 | } |
| 448 | |
| 449 | if system: |
| 450 | kwargs["system"] = system |
| 451 | |
| 452 | if reasoning_effort == "adaptive": |
| 453 | # Adaptive thinking: model decides when and how much to think |
| 454 | # Supported on claude-sonnet-4-6 and claude-opus-4-6. |
| 455 | # Also auto-enables interleaved thinking between tool calls. |
| 456 | kwargs["thinking"] = {"type": "adaptive"} |
| 457 | if not omit_temperature: |
| 458 | kwargs["temperature"] = 1.0 |
| 459 | elif thinking_enabled: |
| 460 | budget_map = {"low": 1024, "medium": 4096, "high": max(8192, max_tokens)} |
| 461 | budget = budget_map.get(reasoning_effort.lower(), 4096) |
| 462 | kwargs["thinking"] = {"type": "enabled", "budget_tokens": budget} |
| 463 | kwargs["max_tokens"] = max(max_tokens, budget + 4096) |
| 464 | if not omit_temperature: |
| 465 | kwargs["temperature"] = 1.0 |
| 466 | elif not omit_temperature: |
| 467 | kwargs["temperature"] = temperature |
| 468 | |
| 469 | if anthropic_tools: |
| 470 | kwargs["tools"] = anthropic_tools |
| 471 | tc = self._convert_tool_choice(tool_choice, thinking_enabled) |
| 472 | if tc: |
| 473 | kwargs["tool_choice"] = tc |
| 474 | |
| 475 | if self.extra_headers: |
| 476 | kwargs["extra_headers"] = self.extra_headers |
| 477 | |
| 478 | return kwargs |
| 479 | |
| 480 | # ------------------------------------------------------------------ |
| 481 | # Response parsing |
| 482 | # ------------------------------------------------------------------ |
| 483 | |
| 484 | @staticmethod |
| 485 | def _parse_response(response: Any) -> LLMResponse: |
| 486 | content_parts: list[str] = [] |
| 487 | tool_calls: list[ToolCallRequest] = [] |
| 488 | thinking_blocks: list[dict[str, Any]] = [] |
| 489 | |
| 490 | for block in response.content: |
| 491 | if block.type == "text": |
| 492 | content_parts.append(block.text) |
| 493 | elif block.type == "tool_use": |
| 494 | tool_calls.append(ToolCallRequest( |
| 495 | id=block.id, |
| 496 | name=block.name, |
| 497 | arguments=block.input if isinstance(block.input, dict) else {}, |
| 498 | )) |
| 499 | elif block.type == "thinking": |
| 500 | thinking_blocks.append({ |
| 501 | "type": "thinking", |
| 502 | "thinking": block.thinking, |
| 503 | "signature": getattr(block, "signature", ""), |
| 504 | }) |
| 505 | |
| 506 | stop_map = {"tool_use": "tool_calls", "end_turn": "stop", "max_tokens": "length"} |
| 507 | finish_reason = stop_map.get(response.stop_reason or "", response.stop_reason or "stop") |
| 508 | |
| 509 | usage: dict[str, int] = {} |
| 510 | if response.usage: |
| 511 | input_tokens = response.usage.input_tokens |
| 512 | cache_creation = getattr(response.usage, "cache_creation_input_tokens", 0) or 0 |
| 513 | cache_read = getattr(response.usage, "cache_read_input_tokens", 0) or 0 |
| 514 | total_prompt_tokens = input_tokens + cache_creation + cache_read |
| 515 | usage = { |
| 516 | "prompt_tokens": total_prompt_tokens, |
| 517 | "completion_tokens": response.usage.output_tokens, |
| 518 | "total_tokens": total_prompt_tokens + response.usage.output_tokens, |
| 519 | } |
| 520 | for attr in ("cache_creation_input_tokens", "cache_read_input_tokens"): |
| 521 | val = getattr(response.usage, attr, 0) |
| 522 | if val: |
| 523 | usage[attr] = val |
| 524 | # Normalize to cached_tokens for downstream consistency. |
| 525 | if cache_read: |
| 526 | usage["cached_tokens"] = cache_read |
| 527 | |
| 528 | return LLMResponse( |
| 529 | content="".join(content_parts) or None, |
| 530 | tool_calls=tool_calls, |
| 531 | finish_reason=finish_reason, |
| 532 | usage=usage, |
| 533 | thinking_blocks=thinking_blocks or None, |
| 534 | ) |
| 535 | |
| 536 | # ------------------------------------------------------------------ |
| 537 | # Public API |
| 538 | # ------------------------------------------------------------------ |
| 539 | |
| 540 | async def chat( |
| 541 | self, |
| 542 | messages: list[dict[str, Any]], |
| 543 | tools: list[dict[str, Any]] | None = None, |
| 544 | model: str | None = None, |
| 545 | max_tokens: int = 4096, |
| 546 | temperature: float = 0.7, |
| 547 | reasoning_effort: str | None = None, |
| 548 | tool_choice: str | dict[str, Any] | None = None, |
| 549 | ) -> LLMResponse: |
| 550 | kwargs = self._build_kwargs( |
| 551 | messages, tools, model, max_tokens, temperature, |
| 552 | reasoning_effort, tool_choice, |
| 553 | ) |
| 554 | try: |
| 555 | response = await self._client.messages.create(**kwargs) |
| 556 | return self._parse_response(response) |
| 557 | except Exception as e: |
| 558 | return self._handle_error(e) |
| 559 | |
| 560 | async def chat_stream( |
| 561 | self, |
| 562 | messages: list[dict[str, Any]], |
| 563 | tools: list[dict[str, Any]] | None = None, |
| 564 | model: str | None = None, |
| 565 | max_tokens: int = 4096, |
| 566 | temperature: float = 0.7, |
| 567 | reasoning_effort: str | None = None, |
| 568 | tool_choice: str | dict[str, Any] | None = None, |
| 569 | on_content_delta: Callable[[str], Awaitable[None]] | None = None, |
| 570 | ) -> LLMResponse: |
| 571 | kwargs = self._build_kwargs( |
| 572 | messages, tools, model, max_tokens, temperature, |
| 573 | reasoning_effort, tool_choice, |
| 574 | ) |
| 575 | idle_timeout_s = int(os.environ.get("NANOBOT_STREAM_IDLE_TIMEOUT_S", "90")) |
| 576 | try: |
| 577 | async with self._client.messages.stream(**kwargs) as stream: |
| 578 | if on_content_delta: |
| 579 | stream_iter = stream.text_stream.__aiter__() |
| 580 | while True: |
| 581 | try: |
| 582 | text = await asyncio.wait_for( |
| 583 | stream_iter.__anext__(), |
| 584 | timeout=idle_timeout_s, |
| 585 | ) |
| 586 | except StopAsyncIteration: |
| 587 | break |
| 588 | await on_content_delta(text) |
| 589 | response = await asyncio.wait_for( |
| 590 | stream.get_final_message(), |
| 591 | timeout=idle_timeout_s, |
| 592 | ) |
| 593 | return self._parse_response(response) |
| 594 | except asyncio.TimeoutError: |
| 595 | return LLMResponse( |
| 596 | content=( |
| 597 | f"Error calling LLM: stream stalled for more than " |
| 598 | f"{idle_timeout_s} seconds" |
| 599 | ), |
| 600 | finish_reason="error", |
| 601 | error_kind="timeout", |
| 602 | ) |
| 603 | except Exception as e: |
| 604 | return self._handle_error(e) |
| 605 | |
| 606 | def get_default_model(self) -> str: |
| 607 | return self.default_model |
| 608 |