""" title: Anthropic API Integration author: grandx, based on Balaxxe & lavantien version: 0.2.0 license: MIT requirements: pydantic>=2.0.0 environment_variables: - ANTHROPIC_API_KEY (required) - THINKING_BUDGET_TOKENS Supports: - All Claude models (Opus 4, Sonnet 4, Sonnet 3.7, Sonnet 3.5) - Streaming responses - Image processing with validation - Prompt caching (5m and 1h) - Function calling (OpenAI format conversion) - PDF processing - Thinking - Citations - Files API references - Cache Control """ import os import json import logging import asyncio import aiohttp from datetime import datetime from typing import ( List, Union, Dict, Optional, AsyncIterator, AsyncGenerator, ) from pydantic import BaseModel, Field from open_webui.utils.misc import pop_system_message from types import SimpleNamespace class Pipe: API_VERSION = "2023-06-01" MODEL_URL = "https://api.anthropic.com/v1/messages" SUPPORTED_IMAGE_TYPES = ["image/jpeg", "image/png", "image/gif", "image/webp"] MAX_IMAGE_SIZE = 5 * 1024 * 1024 MAX_PDF_SIZE = 32 * 1024 * 1024 TOTAL_MAX_IMAGE_SIZE = 100 * 1024 * 1024 MAX_IMAGES_PER_REQUEST = 100 MODEL_MAX_TOKENS = { # Claude 4 models "claude-opus-4-20250514": 32000, "claude-opus-4-0": 32000, "claude-sonnet-4-20250514": 64000, "claude-sonnet-4-0": 64000, # Claude 3.7 models "claude-3-7-sonnet-20250219": 128000, "claude-3-7-sonnet-latest": 128000, # Claude 3.5 models "claude-3-5-sonnet-20241022": 8192, "claude-3-5-sonnet-20240620": 8192, "claude-3-5-sonnet-latest": 8192, } THINKING_BUDGET_TOKENS = 16000 PDF_HEADER = "pdfs-2024-09-25" PROMPT_CACHE_HEADER = "prompt-caching-2024-07-31" OUTPUT128K_HEADER = "output-128k-2025-02-19" EXTENDED_CACHE_HEADER = "extended-cache-ttl-2025-04-11" INTERLEAVED_THINKING_HEADER = "interleaved-thinking-2025-05-14" FILES_API_HEADER = "files-api-2025-04-14" REQUEST_TIMEOUT = 300 class Valves(BaseModel): ANTHROPIC_API_KEY: str = Field( default=os.getenv("ANTHROPIC_API_KEY", ""), description="Your Anthropic API key", ) THINKING_BUDGET_TOKENS: int = Field(default=16000, ge=0, le=96000) def __init__(self): logging.basicConfig(level=logging.INFO) self.type = "manifold" self.id = "anthropic" self.valves = self.Valves() self.request_id = None def get_anthropic_models(self) -> List[dict]: models = [] # Claude 4 models for name in [ "claude-opus-4-20250514", "claude-opus-4-0", "claude-opus-4-0-thinking", "claude-sonnet-4-20250514", "claude-sonnet-4-0", "claude-sonnet-4-0-thinking", ]: models.append( { "id": f"api/{name}", "name": name, "context_length": 200000, "supports_vision": True, } ) # Claude 3.7 models for name in [ "claude-3-7-sonnet-20250219", "claude-3-7-sonnet-latest", "claude-3-7-sonnet-latest-thinking", ]: models.append( { "id": f"api/{name}", "name": name, "context_length": 200000, "supports_vision": True, } ) # Claude 3.5 models for name in [ "claude-3-5-sonnet-20241022", "claude-3-5-sonnet-20240620", "claude-3-5-sonnet-latest", ]: models.append( { "id": f"api/{name}", "name": name, "context_length": 200000, "supports_vision": True, } ) return models def pipes(self) -> List[dict]: return self.get_anthropic_models() def process_content(self, content: Union[str, List[dict]]) -> List[dict]: if isinstance(content, str): return [{"type": "text", "text": content}] processed_content = [] for item in content: if item["type"] == "text": processed_content.append({"type": "text", "text": item["text"]}) elif item["type"] == "image_url": processed_content.append(self.process_image(item)) elif item["type"] == "pdf_url": processed_content.append(self.process_pdf(item)) elif item["type"] == "document": processed_content.append(self.process_document(item)) elif item["type"] == "file_reference": file_ref = self.process_file_reference(item) if file_ref: processed_content.append(file_ref) elif item["type"] == "tool_calls": processed_content.append(item) elif item["type"] == "tool_results": processed_content.append(item) return processed_content def process_image(self, image_data): if image_data["image_url"]["url"].startswith("data:image"): mime_type, base64_data = image_data["image_url"]["url"].split(",", 1) media_type = mime_type.split(":")[1].split(";")[0] if media_type not in self.SUPPORTED_IMAGE_TYPES: raise ValueError(f"Unsupported media type: {media_type}") # Check image size image_size = len(base64_data) * 3 / 4 # Approximate size of decoded base64 if image_size > self.MAX_IMAGE_SIZE: raise ValueError( f"Image size exceeds {self.MAX_IMAGE_SIZE/(1024*1024)}MB limit: {image_size/(1024*1024):.2f}MB" ) return { "type": "image", "source": { "type": "base64", "media_type": media_type, "data": base64_data, }, } else: return { "type": "image", "source": {"type": "url", "url": image_data["image_url"]["url"]}, } def process_pdf(self, pdf_data): if pdf_data["pdf_url"]["url"].startswith("data:application/pdf"): mime_type, base64_data = pdf_data["pdf_url"]["url"].split(",", 1) # Check PDF size pdf_size = len(base64_data) * 3 / 4 # Approximate size of decoded base64 if pdf_size > self.MAX_PDF_SIZE: raise ValueError( f"PDF size exceeds {self.MAX_PDF_SIZE/(1024*1024)}MB limit: {pdf_size/(1024*1024):.2f}MB" ) document = { "type": "document", "source": { "type": "base64", "media_type": "application/pdf", "data": base64_data, }, } if pdf_data.get("cache_control"): document["cache_control"] = pdf_data["cache_control"] return document else: document = { "type": "document", "source": {"type": "url", "url": pdf_data["pdf_url"]["url"]}, } if pdf_data.get("cache_control"): document["cache_control"] = pdf_data["cache_control"] return document def process_document(self, doc_data): """Process documents for citations""" document = { "type": "document", "source": doc_data.get("source"), } # Add optional fields if doc_data.get("title"): document["title"] = doc_data["title"] if doc_data.get("context"): document["context"] = doc_data["context"] # Enable citations if specified if doc_data.get("citations", {}).get("enabled"): document["citations"] = {"enabled": True} # Add cache_control if specified if doc_data.get("cache_control"): document["cache_control"] = doc_data["cache_control"] return document def process_file_reference(self, file_data): """Process file references through Files API""" if file_data.get("file_id"): return { "type": "document", "source": {"type": "file", "file_id": file_data["file_id"]}, } return None def convert_openai_tools_to_anthropic(self, tools: List[dict]) -> List[dict]: """Convert OpenAI format tools to Anthropic format""" anthropic_tools = [] for tool in tools: if tool.get("type") == "function": function = tool.get("function", {}) anthropic_tool = { "name": function.get("name"), "description": function.get("description"), "input_schema": function.get("parameters", {}), } anthropic_tools.append(anthropic_tool) return anthropic_tools def validate_total_image_size(self, messages: List[dict]) -> None: """Validate total size of all images""" total_size = 0 image_count = 0 for message in messages: if isinstance(message.get("content"), list): for item in message["content"]: if item.get("type") == "image_url" and item["image_url"][ "url" ].startswith("data:image"): _, base64_data = item["image_url"]["url"].split(",", 1) image_size = len(base64_data) * 3 / 4 total_size += image_size image_count += 1 if total_size > self.TOTAL_MAX_IMAGE_SIZE: raise ValueError( f"Total image size exceeds {self.TOTAL_MAX_IMAGE_SIZE/(1024*1024)}MB limit: {total_size/(1024*1024):.2f}MB" ) if image_count > self.MAX_IMAGES_PER_REQUEST: raise ValueError( f"Too many images: {image_count}. Maximum is {self.MAX_IMAGES_PER_REQUEST}." ) async def pipe(self, body: Dict) -> Union[str, AsyncGenerator[str, None]]: if not self.valves.ANTHROPIC_API_KEY: return {"content": "Error: ANTHROPIC_API_KEY is required", "format": "text"} try: system_message, messages = pop_system_message(body["messages"]) # Validate images self.validate_total_image_size(messages) # Determine if using thinking mode model_name = body["model"].split("/")[-1] is_thinking_mode = "-thinking" in body["model"] if is_thinking_mode: # Strip the "-thinking" suffix for API call model_name = model_name.replace("-thinking", "") max_tokens_limit = self.MODEL_MAX_TOKENS.get(model_name, 4096) payload = { "model": model_name, "messages": self._process_messages(messages), "max_tokens": min( body.get("max_tokens", max_tokens_limit), max_tokens_limit ), "temperature": ( float(body.get("temperature")) if body.get("temperature") is not None else None ), "top_k": ( int(body.get("top_k")) if body.get("top_k") is not None else None ), "top_p": ( float(body.get("top_p")) if body.get("top_p") is not None else None ), "stream": body.get("stream", True), "metadata": body.get("metadata", {}), } payload = {k: v for k, v in payload.items() if v is not None} if system_message: payload["system"] = str(system_message) # Convert OpenAI tools to Anthropic format if "tools" in body: payload["tools"] = self.convert_openai_tools_to_anthropic(body["tools"]) # Convert tool_choice tool_choice = body.get("tool_choice", "auto") if tool_choice == "none": payload["tool_choice"] = {"type": "none"} elif tool_choice == "auto": payload["tool_choice"] = {"type": "auto"} elif isinstance(tool_choice, dict) and tool_choice.get("function"): payload["tool_choice"] = { "type": "tool", "name": tool_choice["function"]["name"], } if "response_format" in body: payload["response_format"] = { "type": body["response_format"].get("type") } # Add thinking parameters if using thinking mode if is_thinking_mode: payload["thinking"] = { "type": "enabled", "budget_tokens": self.valves.THINKING_BUDGET_TOKENS, } headers = { "x-api-key": self.valves.ANTHROPIC_API_KEY, "anthropic-version": self.API_VERSION, "content-type": "application/json", } beta_headers = [] # Add 128K output beta header for Claude 3.7 if model_name in ["claude-3-7-sonnet-20250219", "claude-3-7-sonnet-latest"]: beta_headers.append(self.OUTPUT128K_HEADER) # Check for beta features in messages has_pdf = False has_cache = False has_1h_cache = False has_files = False for msg in body.get("messages", []): if isinstance(msg["content"], list): for item in msg["content"]: if item.get("type") == "pdf_url": has_pdf = True if item.get("cache_control"): has_cache = True if item.get("cache_control", {}).get("ttl") == "1h": has_1h_cache = True if item.get("type") == "file_reference": has_files = True if has_pdf: beta_headers.append(self.PDF_HEADER) if has_cache: beta_headers.append(self.PROMPT_CACHE_HEADER) if has_1h_cache: beta_headers.append(self.EXTENDED_CACHE_HEADER) if has_files: beta_headers.append(self.FILES_API_HEADER) # Add interleaved thinking if requested if is_thinking_mode and body.get("interleaved_thinking", False): beta_headers.append(self.INTERLEAVED_THINKING_HEADER) if beta_headers: headers["anthropic-beta"] = ",".join(beta_headers) try: if payload["stream"]: return self._stream_with_ui(self.MODEL_URL, headers, payload, body) response = await self._send_request(self.MODEL_URL, headers, payload) if response.status_code != 200: return { "content": f"Error: HTTP {response.status_code}: {response.text}", "format": "text", } result, cache_metrics = self._handle_response(response) response_text = result["content"][0]["text"] return response_text except aiohttp.ClientError as e: error_msg = f"Request failed: {str(e)}" if self.request_id: error_msg += f" (Request ID: {self.request_id})" return {"content": error_msg, "format": "text"} except Exception as e: error_msg = f"Error: {str(e)}" if self.request_id: error_msg += f" (Request ID: {self.request_id})" return {"content": error_msg, "format": "text"} async def _stream_with_ui( self, url: str, headers: dict, payload: dict, body: dict ) -> AsyncGenerator[str, None]: try: async with aiohttp.ClientSession() as session: async with session.post( url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=self.REQUEST_TIMEOUT), ) as response: self.request_id = response.headers.get("x-request-id") if response.status != 200: error_msg = ( f"Error: HTTP {response.status}: {await response.text()}" ) if self.request_id: error_msg += f" (Request ID: {self.request_id})" yield error_msg return async for line in response.content: if line and line.startswith(b"data: "): try: data = json.loads(line[6:]) if data["type"] == "content_block_delta": delta = data.get("delta", {}) # Handle text_delta if delta.get("type") == "text_delta": yield delta.get("text", "") # Handle citations_delta elif delta.get("type") == "citations_delta": citation = delta.get("citation", {}) if citation: doc_index = citation.get( "document_index", 0 ) doc_title = citation.get( "document_title", "" ) if doc_title: yield f" [{doc_index}: {doc_title}]" else: yield f" [{doc_index}]" elif data["type"] == "message_stop": break except json.JSONDecodeError as e: logging.error( f"Failed to parse streaming response: {e}" ) continue except asyncio.TimeoutError: error_msg = "Request timed out" if self.request_id: error_msg += f" (Request ID: {self.request_id})" yield error_msg except Exception as e: error_msg = f"Stream error: {str(e)}" if self.request_id: error_msg += f" (Request ID: {self.request_id})" yield error_msg def _process_messages(self, messages: List[dict]) -> List[dict]: processed_messages = [] for message in messages: processed_content = [] for content in self.process_content(message["content"]): if ( message.get("role") == "assistant" and content.get("type") == "tool_calls" ): content["cache_control"] = {"type": "ephemeral"} elif ( message.get("role") == "user" and content.get("type") == "tool_results" ): content["cache_control"] = {"type": "ephemeral"} processed_content.append(content) processed_messages.append( {"role": message["role"], "content": processed_content} ) return processed_messages async def _send_request( self, url: str, headers: dict, payload: dict ) -> SimpleNamespace: retry_count = 0 base_delay = 1 # Start with 1 second delay max_retries = 3 async with aiohttp.ClientSession() as session: while retry_count < max_retries: try: async with session.post( url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=self.REQUEST_TIMEOUT), ) as response: self.request_id = response.headers.get("x-request-id") if response.status == 429: retry_after = int( response.headers.get( "retry-after", base_delay * (2**retry_count) ) ) logging.warning( f"Rate limit hit. Retrying in {retry_after} seconds. Retry count: {retry_count + 1}" ) await asyncio.sleep(retry_after) retry_count += 1 continue response_text = await response.text() response_wrapper = SimpleNamespace( status_code=response.status, headers=response.headers, text=response_text, json=lambda: json.loads(response_text), ) return response_wrapper except aiohttp.ClientError as e: logging.error(f"Request failed: {str(e)}") raise e logging.error("Max retries exceeded for rate limit.") return SimpleNamespace( status_code=429, headers={}, text="Max retries exceeded", json=lambda: {"error": {"message": "Max retries exceeded"}}, ) def _handle_response(self, response): if response.status_code != 200: error_msg = f"Error: HTTP {response.status_code}" try: error_data = response.json().get("error", {}) error_msg += f": {error_data.get('message', response.text)}" error_type = error_data.get("type") if error_type: error_msg += f" (Type: {error_type})" except: error_msg += f": {response.text}" self.request_id = response.headers.get("x-request-id") if self.request_id: error_msg += f" (Request ID: {self.request_id})" return {"content": error_msg, "format": "text"}, None result = response.json() usage = result.get("usage", {}) cache_metrics = { "cache_creation_input_tokens": usage.get("cache_creation_input_tokens", 0), "cache_read_input_tokens": usage.get("cache_read_input_tokens", 0), "input_tokens": usage.get("input_tokens", 0), "output_tokens": usage.get("output_tokens", 0), } return result, cache_metrics