""" title: Anthropic SDK Pipe authors: italomg author_url: https://github.com/italomg funding_url: https://github.com/open-webui version: 0.1 required_open_webui_version: 0.3.17 license: MIT """ import os import requests from typing import List, Union, Generator, Iterator from pydantic import BaseModel, Field import anthropic import mimetypes import base64 from open_webui.utils.misc import pop_system_message class Pipe: class Valves(BaseModel): ANTHROPIC_API_KEY: str = Field(default="") def __init__(self): self.type = "manifold" self.valves = self.Valves( **{"ANTHROPIC_API_KEY": os.getenv("ANTHROPIC_API_KEY", "")} ) self.MAX_IMAGE_SIZE = 5 * 1024 * 1024 # 5MB per image self.anthropic_client = anthropic.Anthropic(api_key=self.valves.ANTHROPIC_API_KEY) pass def pipes(self) -> List[dict]: response = self.anthropic_client.models.list() models = [{"id": model.id, "name": model.display_name} for model in response] return models def _process_image(self, file_path): """Encode image to base64 and get media type""" with open(file_path, "rb") as image_file: media_type = mimetypes.guess_type(file_path)[0] if not media_type or not media_type.startswith("image"): raise ValueError("Unsupported image format") image_data = { "type": "base64", "media_type": media_type, "data": base64.b64encode(image_file.read()).decode("utf-8") } return {"type": "image", "source": image_data} def pipe(self, body: dict) -> Union[str, Generator, Iterator]: system_message, messages = pop_system_message(body["messages"]) processed_messages = [] total_image_size = 0 for message in messages: processed_content = [] if isinstance(message.get("content"), list): for item in message["content"]: if item["type"] == "text": processed_content.append({"type": "text", "text": item["text"]}) elif item["type"] == "image_url": processed_image = self._process_image(item) processed_content.append(processed_image) # Track total size for base64 images if processed_image["source"]["type"] == "base64": image_size = len(processed_image["source"]["data"]) * 3 / 4 total_image_size += image_size if ( total_image_size > 100 * 1024 * 1024 ): # 100MB total limit raise ValueError( "Total size of images exceeds 100 MB limit" ) else: processed_content = [ {"type": "text", "text": message.get("content", "")} ] processed_messages.append( {"role": message["role"], "content": processed_content} ) payload = { "model": body["model"], "messages": processed_messages, "max_tokens": body.get("max_tokens", 4096), "temperature": body.get("temperature", 0.8), "top_k": body.get("top_k", 40), "top_p": body.get("top_p", 0.9), "stop_sequences": body.get("stop", []), **({"system": str(system_message)} if system_message else {}), "stream": body.get("stream", False), } try: if body.get("stream", False): with self.anthropic_client.messages.stream(**payload) as stream: for text in stream.text_stream: yield text else: message = self.anthropic_client.messages.create(**payload) return message.content[0].text except requests.exceptions.RequestException as e: print(f"Request failed: {e}") return f"Error: Request failed: {e}" except Exception as e: print(f"Error in pipe method: {e}") return f"Error: {e}"