[{"id":"anthropic","name":"Anthropic","meta":{"description":"Anthropic Manifold Pipe","manifest":{"title":"Anthropic Manifold Pipe","authors":"justinh-rahb and christian-taillon","author_url":"https://github.com/justinh-rahb","funding_url":"https://github.com/open-webui","version":"0.2.5","required_open_webui_version":"0.3.17","license":"MIT"},"type":"pipe","user":{"id":"937b18cd-3dff-4680-a0c7-37eedb47cfe3","username":"justinrahb","name":"","createdAt":1719521880,"role":"mod","verified":false},"id":"60984ebf-cf4e-4822-98b8-0492463c2852"},"content":"\"\"\"\ntitle: Anthropic Manifold Pipe\nauthors: justinh-rahb and christian-taillon\nauthor_url: https://github.com/justinh-rahb\nfunding_url: https://github.com/open-webui\nversion: 0.2.5\nrequired_open_webui_version: 0.3.17\nlicense: MIT\n\"\"\"\n\nimport os\nimport requests\nimport json\nimport time\nfrom typing import List, Union, Generator, Iterator\nfrom pydantic import BaseModel, Field\nfrom open_webui.utils.misc import pop_system_message\n\n\nclass Pipe:\n class Valves(BaseModel):\n ANTHROPIC_API_KEY: str = Field(default=\"\")\n\n def __init__(self):\n self.type = \"manifold\"\n self.id = \"anthropic\"\n self.name = \"anthropic/\"\n self.valves = self.Valves(\n **{\"ANTHROPIC_API_KEY\": os.getenv(\"ANTHROPIC_API_KEY\", \"\")}\n )\n self.MAX_IMAGE_SIZE = 5 * 1024 * 1024 # 5MB per image\n pass\n\n def get_anthropic_models(self):\n return [\n {\"id\": \"claude-3-haiku-20240307\", \"name\": \"claude-3-haiku\"},\n {\"id\": \"claude-3-opus-20240229\", \"name\": \"claude-3-opus\"},\n {\"id\": \"claude-3-sonnet-20240229\", \"name\": \"claude-3-sonnet\"},\n {\"id\": \"claude-3-5-haiku-20241022\", \"name\": \"claude-3.5-haiku\"},\n {\"id\": \"claude-3-5-haiku-latest\", \"name\": \"claude-3.5-haiku\"},\n {\"id\": \"claude-3-5-sonnet-20240620\", \"name\": \"claude-3.5-sonnet\"},\n {\"id\": \"claude-3-5-sonnet-20241022\", \"name\": \"claude-3.5-sonnet\"},\n {\"id\": \"claude-3-5-sonnet-latest\", \"name\": \"claude-3.5-sonnet\"},\n {\"id\": \"claude-3-7-sonnet-20250219\", \"name\": \"claude-3.7-sonnet\"},\n {\"id\": \"claude-3-7-sonnet-latest\", \"name\": \"claude-3.7-sonnet\"},\n ]\n\n def pipes(self) -> List[dict]:\n return self.get_anthropic_models()\n\n def process_image(self, image_data):\n \"\"\"Process image data with size validation.\"\"\"\n if image_data[\"image_url\"][\"url\"].startswith(\"data:image\"):\n mime_type, base64_data = image_data[\"image_url\"][\"url\"].split(\",\", 1)\n media_type = mime_type.split(\":\")[1].split(\";\")[0]\n\n # Check base64 image size\n image_size = len(base64_data) * 3 / 4 # Convert base64 size to bytes\n if image_size > self.MAX_IMAGE_SIZE:\n raise ValueError(\n f\"Image size exceeds 5MB limit: {image_size / (1024 * 1024):.2f}MB\"\n )\n\n return {\n \"type\": \"image\",\n \"source\": {\n \"type\": \"base64\",\n \"media_type\": media_type,\n \"data\": base64_data,\n },\n }\n else:\n # For URL images, perform size check after fetching\n url = image_data[\"image_url\"][\"url\"]\n response = requests.head(url, allow_redirects=True)\n content_length = int(response.headers.get(\"content-length\", 0))\n\n if content_length > self.MAX_IMAGE_SIZE:\n raise ValueError(\n f\"Image at URL exceeds 5MB limit: {content_length / (1024 * 1024):.2f}MB\"\n )\n\n return {\n \"type\": \"image\",\n \"source\": {\"type\": \"url\", \"url\": url},\n }\n\n def pipe(self, body: dict) -> Union[str, Generator, Iterator]:\n system_message, messages = pop_system_message(body[\"messages\"])\n\n processed_messages = []\n total_image_size = 0\n\n for message in messages:\n processed_content = []\n if isinstance(message.get(\"content\"), list):\n for item in message[\"content\"]:\n if item[\"type\"] == \"text\":\n processed_content.append({\"type\": \"text\", \"text\": item[\"text\"]})\n elif item[\"type\"] == \"image_url\":\n processed_image = self.process_image(item)\n processed_content.append(processed_image)\n\n # Track total size for base64 images\n if processed_image[\"source\"][\"type\"] == \"base64\":\n image_size = len(processed_image[\"source\"][\"data\"]) * 3 / 4\n total_image_size += image_size\n if (\n total_image_size > 100 * 1024 * 1024\n ): # 100MB total limit\n raise ValueError(\n \"Total size of images exceeds 100 MB limit\"\n )\n else:\n processed_content = [\n {\"type\": \"text\", \"text\": message.get(\"content\", \"\")}\n ]\n\n processed_messages.append(\n {\"role\": message[\"role\"], \"content\": processed_content}\n )\n\n payload = {\n \"model\": body[\"model\"][body[\"model\"].find(\".\") + 1 :],\n \"messages\": processed_messages,\n \"max_tokens\": body.get(\"max_tokens\", 4096),\n \"temperature\": body.get(\"temperature\", 0.8),\n \"top_k\": body.get(\"top_k\", 40),\n \"top_p\": body.get(\"top_p\", 0.9),\n \"stop_sequences\": body.get(\"stop\", []),\n **({\"system\": str(system_message)} if system_message else {}),\n \"stream\": body.get(\"stream\", False),\n }\n\n headers = {\n \"x-api-key\": self.valves.ANTHROPIC_API_KEY,\n \"anthropic-version\": \"2023-06-01\",\n \"content-type\": \"application/json\",\n }\n\n url = \"https://api.anthropic.com/v1/messages\"\n\n try:\n if body.get(\"stream\", False):\n return self.stream_response(url, headers, payload)\n else:\n return self.non_stream_response(url, headers, payload)\n except requests.exceptions.RequestException as e:\n print(f\"Request failed: {e}\")\n return f\"Error: Request failed: {e}\"\n except Exception as e:\n print(f\"Error in pipe method: {e}\")\n return f\"Error: {e}\"\n\n def stream_response(self, url, headers, payload):\n try:\n with requests.post(\n url, headers=headers, json=payload, stream=True, timeout=(3.05, 60)\n ) as response:\n if response.status_code != 200:\n raise Exception(\n f\"HTTP Error {response.status_code}: {response.text}\"\n )\n\n for line in response.iter_lines():\n if line:\n line = line.decode(\"utf-8\")\n if line.startswith(\"data: \"):\n try:\n data = json.loads(line[6:])\n if data[\"type\"] == \"content_block_start\":\n yield data[\"content_block\"][\"text\"]\n elif data[\"type\"] == \"content_block_delta\":\n yield data[\"delta\"][\"text\"]\n elif data[\"type\"] == \"message_stop\":\n break\n elif data[\"type\"] == \"message\":\n for content in data.get(\"content\", []):\n if content[\"type\"] == \"text\":\n yield content[\"text\"]\n\n time.sleep(\n 0.01\n ) # Delay to avoid overwhelming the client\n\n except json.JSONDecodeError:\n print(f\"Failed to parse JSON: {line}\")\n except KeyError as e:\n print(f\"Unexpected data structure: {e}\")\n print(f\"Full data: {data}\")\n except requests.exceptions.RequestException as e:\n print(f\"Request failed: {e}\")\n yield f\"Error: Request failed: {e}\"\n except Exception as e:\n print(f\"General error in stream_response method: {e}\")\n yield f\"Error: {e}\"\n\n def non_stream_response(self, url, headers, payload):\n try:\n response = requests.post(\n url, headers=headers, json=payload, timeout=(3.05, 60)\n )\n if response.status_code != 200:\n raise Exception(f\"HTTP Error {response.status_code}: {response.text}\")\n\n res = response.json()\n return (\n res[\"content\"][0][\"text\"] if \"content\" in res and res[\"content\"] else \"\"\n )\n except requests.exceptions.RequestException as e:\n print(f\"Failed non-stream request: {e}\")\n return f\"Error: {e}\"\n"}]