""" title: Anthropic Function (Simplified) author: Seyed Yahya Shirazi author_url: neuromechanist.github.io date: 2025-07-20 version: 2.1 license: MIT description: A simplified function for generating text and processing images using the Anthropic API. requirements: aiohttp, pydantic environment_variables: ANTHROPIC_API_KEY """ import os import json import asyncio from typing import List, Union, Dict, Optional, AsyncIterator from pydantic import BaseModel import aiohttp from open_webui.utils.misc import pop_system_message class Pipe: class Valves(BaseModel): ANTHROPIC_API_KEY: str = "" def __init__(self): self.type = "manifold" self.id = "anthropic_simple" self.valves = self.Valves() def pipes(self) -> List[dict]: return [ {"id": "anthropic/claude-sonnet-4-20250514", "name": "claude-4.0-sonnet"}, {"id": "anthropic/claude-3-7-sonnet-latest", "name": "claude-3.7-sonnet"}, ] def process_image(self, image_data): if image_data["url"].startswith("data:image"): mime_type, base64_data = image_data["url"].split(",", 1) media_type = mime_type.split(":")[1].split(";")[0] return { "type": "image", "source": { "type": "base64", "media_type": media_type, "data": base64_data, }, } else: return { "type": "image", "source": {"type": "url", "url": image_data["url"]}, } async def pipe( self, body: Dict, __event_emitter__=None ) -> Union[str, AsyncIterator[str]]: if not self.valves.ANTHROPIC_API_KEY: error_msg = "Error: ANTHROPIC_API_KEY is required" if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": error_msg, "done": True}, } ) return error_msg try: system_message, messages = pop_system_message(body["messages"]) # Extract actual model name from "anthropic/model-name" format model_id = body.get("model", "anthropic/claude-3-5-sonnet-20241022").split( "/" )[-1] # Process messages for Anthropic format processed_messages = [] 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["image_url"]) processed_content.append(processed_image) else: processed_content = [ {"type": "text", "text": message.get("content", "")} ] processed_messages.append( {"role": message["role"], "content": processed_content} ) # Prepare payload payload = { "model": model_id, "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), "stream": body.get("stream", False), } if system_message: payload["system"] = str(system_message) headers = { "x-api-key": self.valves.ANTHROPIC_API_KEY, "anthropic-version": "2023-06-01", "content-type": "application/json", } if body.get("stream", False): return self._stream_response(headers, payload, __event_emitter__) else: return await self._get_completion(headers, payload, __event_emitter__) except Exception as e: error_msg = f"Error: {str(e)}" if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": error_msg, "done": True}, } ) return error_msg async def _stream_response( self, headers: dict, payload: dict, __event_emitter__=None ) -> AsyncIterator[str]: try: async with aiohttp.ClientSession() as session: timeout = aiohttp.ClientTimeout(total=300) async with session.post( "https://api.anthropic.com/v1/messages", headers=headers, json=payload, timeout=timeout, ) as response: if response.status != 200: error_text = await response.text() error_msg = f"Error: HTTP {response.status}: {error_text}" if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": error_msg, "done": True}, } ) yield error_msg return async for line in response.content: if line and line.startswith(b"data: "): try: line_text = line[6:].decode("utf-8").strip() if line_text == "[DONE]": break data = json.loads(line_text) if data["type"] == "content_block_start": yield data["content_block"]["text"] elif data["type"] == "content_block_delta": yield data["delta"]["text"] elif data["type"] == "message_stop": break except json.JSONDecodeError: continue except KeyError: continue except Exception as e: error_msg = f"Stream error: {str(e)}" if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": error_msg, "done": True}, } ) yield error_msg async def _get_completion( self, headers: dict, payload: dict, __event_emitter__=None ) -> str: try: async with aiohttp.ClientSession() as session: timeout = aiohttp.ClientTimeout(total=300) async with session.post( "https://api.anthropic.com/v1/messages", headers=headers, json=payload, timeout=timeout, ) as response: if response.status != 200: error_text = await response.text() error_msg = f"Error: HTTP {response.status}: {error_text}" if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": error_msg, "done": True}, } ) return error_msg result = await response.json() return ( result["content"][0]["text"] if "content" in result and result["content"] else "" ) except Exception as e: error_msg = f"Completion error: {str(e)}" if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": error_msg, "done": True}, } ) return error_msg