""" title: Anthropic API Integration for OpenWebUI (Incremental Caching + Image/PDF support) author: Balaxxe (modified by alphanumeric0) version: 2.4-cached license: MIT requirements: pydantic>=2.0.0, requests>=2.0.0 environment_variables: - ANTHROPIC_API_KEY (required) # For Anthropic access - ANTHROPIC_CACHE_STRATEGY (optional) # e.g. none|incremental|system ------------------------------------------------------------------------------------ OVERVIEW OF THIS SCRIPT: 1) By default, it uses an "incremental" (wandering) strategy to gradually expand what’s in Anthropic’s cache. - "Incremental" / "Wandering" Strategy: (a) On the first request, we mark the system block with cache control (if any), plus the last block of the user’s message. (b) On subsequent requests, we only mark the last block of the newest user message, since older text is presumably cached from prior calls. 2) Env variable toggles for caching approaches: - none - system - incremental (default) 3) Edge Cases: - Marking too many blocks with cache_control at once can exceed Anthropic’s 4-block limit. 4) Streaming vs. Non-streaming: - `_stream_with_ui` uses `aiohttp` for streaming partial responses. - Otherwise, we do a single request via `requests`. 5) Error Handling: - If Anthropic complains about “A maximum of 4 blocks with cache_control may be provided,” we append a note to the returned error text. ------------------------------------------------------------------------------------ """ import os import requests import json import time import logging import aiohttp from typing import List, Dict, Any, Union, Generator, Iterator, Optional from pydantic import BaseModel, Field from open_webui.utils.misc import pop_system_message class Pipe: API_VERSION = "2023-06-01" MODEL_URL = "https://api.anthropic.com/v1/messages" # Known token limits per model MODEL_MAX_TOKENS = { "claude-3-opus-20240229": 4096, "claude-3-sonnet-20240229": 4096, "claude-3-haiku-20240307": 4096, "claude-3-5-sonnet-20240620": 8192, "claude-3-5-sonnet-20241022": 8192, "claude-3-5-haiku-20241022": 8192, "claude-3-opus-latest": 4096, "claude-3-5-sonnet-latest": 8192, "claude-3-5-haiku-latest": 8192, } # Optional image/PDF constraints SUPPORTED_IMAGE_TYPES = ["image/jpeg", "image/png", "image/gif", "image/webp"] MAX_IMAGE_SIZE = 5 * 1024 * 1024 # 5 MB MAX_PDF_SIZE = 32 * 1024 * 1024 # 32 MB class Valves(BaseModel): ANTHROPIC_API_KEY: str = Field( default=os.getenv("ANTHROPIC_API_KEY", ""), description="Your Anthropic API key", ) ANTHROPIC_CACHE_STRATEGY: str = Field( default=os.getenv("ANTHROPIC_CACHE_STRATEGY", "incremental"), description="Caching approach: none|incremental|system", ) def __init__(self): logging.basicConfig(level=logging.INFO) self.type = "manifold" self.id = "anthropic" self.valves = self.Valves() self.request_id = None # --------------------------------------------------------- # AVAILABLE MODELS # --------------------------------------------------------- def pipes(self) -> List[dict]: """Return a static list of pipe descriptors for each Anthropic model.""" return [ { "id": f"anthropic/{name}", "name": name, "context_length": 200000, "supports_vision": True, } for name in self.MODEL_MAX_TOKENS ] # --------------------------------------------------------- # IMAGE/PDF HELPERS # --------------------------------------------------------- def _process_image(self, image_data: dict) -> dict: """ Handle image URLs vs. base64 data. Check MIME type & size if desired. """ url_str = image_data["image_url"]["url"] if url_str.startswith("data:image"): # parse data URI mime_type_part, base64_data = url_str.split(",", 1) media_type = mime_type_part.split(":")[1].split(";")[0] if media_type not in self.SUPPORTED_IMAGE_TYPES: raise ValueError(f"Unsupported image media type: {media_type}") # Approx. check for size (Base64 grows ~4/3 overhead) image_size = (len(base64_data) * 3) // 4 if image_size > self.MAX_IMAGE_SIZE: raise ValueError( f"Image size exceeds {self.MAX_IMAGE_SIZE / (1024 * 1024):.2f}MB limit." ) return { "type": "image", "source": { "type": "base64", "media_type": media_type, "data": base64_data, }, } else: # Just pass it as a URL return { "type": "image", "source": {"type": "url", "url": url_str}, } def _process_pdf(self, pdf_data: dict) -> dict: """ Handle PDF URLs vs. base64 data. (Size check optional.) """ url_str = pdf_data["pdf_url"]["url"] if url_str.startswith("data:application/pdf"): # parse data URI _, base64_data = url_str.split(",", 1) # Approx. check for size pdf_size = (len(base64_data) * 3) // 4 if pdf_size > self.MAX_PDF_SIZE: raise ValueError( f"PDF size exceeds {self.MAX_PDF_SIZE / (1024 * 1024):.2f}MB limit." ) return { "type": "document", "source": { "type": "base64", "media_type": "application/pdf", "data": base64_data, }, } else: return { "type": "document", "source": {"type": "url", "url": url_str}, } # --------------------------------------------------------- # CONVERT USER CONTENT INTO BLOCKS # --------------------------------------------------------- def _process_content(self, content: Union[str, List[dict]]) -> List[dict]: """ Convert user-provided content into Anthropic-friendly blocks. Now includes image/pdf handling. """ if isinstance(content, str): return [{"type": "text", "text": content}] processed = [] for item in content: block_type = item.get("type", "text") if block_type == "text": processed.append({"type": "text", "text": item["text"]}) elif block_type == "image_url": processed.append(self._process_image(item)) elif block_type == "pdf_url": processed.append(self._process_pdf(item)) else: # If user passed custom blocks, or new types, pass them through processed.append(item) return processed def _process_messages(self, messages: List[dict]) -> List[dict]: """ Transform each message dict into the required format: { "role": "user"/"assistant", "content": [ {"type": "text", "text": "..."} ... ] } """ result = [] for msg in messages: blocks = self._process_content(msg["content"]) result.append({"role": msg["role"], "content": blocks}) return result # --------------------------------------------------------- # STRATEGIES FOR ADDING CACHE CONTROL # --------------------------------------------------------- def _mark_cache_control( self, strategy: str, system_str: Optional[str], messages: List[dict], is_first_request: bool, ): """ Depending on the strategy, mark blocks with `cache_control`. - 'none': do nothing - 'system': always mark system - 'incremental': if first request => mark system + last block of single user message else => mark last block of newest user message Returns (system_value, messages_value). """ # STRATEGY = none if strategy == "none": return system_str, messages if strategy == "system": if system_str is not None: if isinstance(system_str, str): system_blocks = [ { "type": "text", "text": system_str, "cache_control": {"type": "ephemeral"}, } ] return system_blocks, messages if isinstance(system_str, list): for block in system_str: block["cache_control"] = {"type": "ephemeral"} return system_str, messages return system_str, messages if strategy == "incremental": # 1) If it's the first request and we have a system_str, mark it if is_first_request and system_str is not None: if isinstance(system_str, str): system_val = [ { "type": "text", "text": system_str, "cache_control": {"type": "ephemeral"}, } ] elif isinstance(system_str, list): # mark each block in system for block in system_str: block["cache_control"] = {"type": "ephemeral"} system_val = system_str else: system_val = system_str else: system_val = system_str # 2) Mark the last block of the newest (latest) user message user_indices = [i for i, m in enumerate(messages) if m["role"] == "user"] if user_indices: last_user_idx = user_indices[-1] if messages[last_user_idx]["content"]: messages[last_user_idx]["content"][-1]["cache_control"] = { "type": "ephemeral" } # 3) Mark the last block in the second-to-last user message (if present) if len(user_indices) > 1: second_last_user_idx = user_indices[-2] if messages[second_last_user_idx]["content"]: messages[second_last_user_idx]["content"][-1]["cache_control"] = { "type": "ephemeral" } return system_val, messages return system_str, messages # --------------------------------------------------------- # MAIN PIPE # --------------------------------------------------------- async def pipe( self, body: Dict, __event_emitter__=None ) -> Union[str, Generator, Iterator]: """ Main entry point for the OpenWebUI. 1. We pop system from body 2. Convert messages to blocks 3. Mark blocks with cache control based on strategy 4. Send to Anthropic 5. Return or stream """ # 1) Check for API key if not self.valves.ANTHROPIC_API_KEY: return {"content": "Error: ANTHROPIC_API_KEY is required", "format": "text"} # 2) Ensure we have 'messages' if "messages" not in body: return { "content": "Error: The request body must include 'messages'.", "format": "text", } # 3) Retrieve caching strategy strategy = self.valves.ANTHROPIC_CACHE_STRATEGY.lower().strip() # 4) Pop system from the conversation system_str, raw_messages = pop_system_message(body["messages"]) # 5) Convert messages -> blocks processed_msgs = self._process_messages(raw_messages) # 6) Decide if first request user_count = sum(1 for m in processed_msgs if m["role"] == "user") is_first_request = user_count <= 1 # 7) Mark with cache control new_system, new_msgs = self._mark_cache_control( strategy, system_str, processed_msgs, is_first_request ) # 8) Build payload model_id = body.get("model", "claude-2") model_name = model_id.split("/")[-1] if "/" in model_id else model_id max_tokens = self.MODEL_MAX_TOKENS.get(model_name, 4096) used_max_tokens = min(body.get("max_tokens", 1024), max_tokens) payload = { "model": model_name, "messages": new_msgs, "max_tokens": used_max_tokens, "temperature": ( float(body["temperature"]) if "temperature" in body else None ), "stream": body.get("stream", False), } # If system is a list => pass that (list of blocks). # If a single string => pass as string. # If None => exclude it entirely. if isinstance(new_system, list): payload["system"] = new_system elif isinstance(new_system, str): payload["system"] = new_system # Tools if "tools" in body: payload["tools"] = [ {"type": "function", "function": t} for t in body["tools"] ] # Response format if "response_format" in body: payload["response_format"] = {"type": body["response_format"].get("type")} # Cleanup None fields payload = {k: v for k, v in payload.items() if v is not None} # 9) Build headers (no beta headers needed anymore) headers = { "x-api-key": self.valves.ANTHROPIC_API_KEY, "anthropic-version": self.API_VERSION, "content-type": "application/json", } # 10) Send request try: if payload.get("stream"): return self._stream_with_ui( self.MODEL_URL, headers, payload, body, __event_emitter__ ) resp = requests.post( self.MODEL_URL, headers=headers, json=payload, timeout=60 ) if resp.status_code != 200: error_txt = resp.text if ( "A maximum of 4 blocks with cache_control may be provided" in error_txt ): error_txt += "\n(Anthropic cache_control limit exceeded!)" return { "content": f"Error {resp.status_code}: {error_txt}", "format": "text", } # Parse JSON rj = resp.json() # Usually rj["content"] is a list of blocks if ( "content" in rj and isinstance(rj["content"], list) and len(rj["content"]) > 0 ): first_block = rj["content"][0] if "text" in first_block: return first_block["text"] else: return "(No 'text' field in the first content block.)" else: return "(No content in response.)" except requests.exceptions.RequestException as e: return {"content": f"Request failed: {e}", "format": "text"} # --------------------------------------------------------- # STREAMING LOGIC # --------------------------------------------------------- async def _stream_with_ui( self, url: str, headers: dict, payload: dict, body: dict, __event_emitter__=None ) -> Generator: """ Stream partial responses from Anthropic using aiohttp. """ try: async with aiohttp.ClientSession() as session: async with session.post(url, headers=headers, json=payload) as response: self.request_id = response.headers.get("x-request-id") if response.status != 200: text_ = await response.text() if ( "A maximum of 4 blocks with cache_control may be provided" in text_ ): text_ += "\n(Anthropic cache_control limit exceeded!)" yield f"Error {response.status}: {text_}" 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" and "text" in data["delta"] ): yield data["delta"]["text"] elif data["type"] == "message_stop": break except json.JSONDecodeError: continue except Exception as e: yield f"Stream error: {e}"