""" title: Anthropic Function with Web Search author: Seyed Yahya Shirazi author_url: neuromechanist.github.io date: 2025-08-01 version: 2.2 license: MIT description: A function for generating text and processing images using the Anthropic API with optional web search. requirements: aiohttp, pydantic environment_variables: ANTHROPIC_API_KEY """ import json from typing import List, Union, Dict, 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 = "" ENABLE_WEB_SEARCH: bool = True MAX_SEARCH_USES: int = 5 DEBUG_CITATIONS: bool = False def __init__(self): self.type = "manifold" self.id = "anthropic_websearch" self.valves = self.Valves() def pipes(self) -> List[dict]: return [ {"id": "anthropic/claude-sonnet-4-20250514", "name": "claude-4.0-sonnet-web"}, {"id": "anthropic/claude-3-7-sonnet-latest", "name": "claude-3.7-sonnet-web"}, ] 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"]}, } def extract_citations_from_response(self, content_blocks: List[dict]) -> tuple[str, List[str]]: """Extract text content and citations from Anthropic response blocks.""" text_parts = [] citations = [] citation_map = {} # To track unique URLs if self.valves.DEBUG_CITATIONS: print(f"DEBUG: Processing {len(content_blocks)} content blocks") for i, block in enumerate(content_blocks): print(f"DEBUG: Block {i}: type={block.get('type')}, keys={list(block.keys())}") for block in content_blocks: if block.get("type") == "text": text_parts.append(block.get("text", "")) # Extract citations from this text block if present if "citations" in block: if self.valves.DEBUG_CITATIONS: print(f"DEBUG: Found {len(block['citations'])} citations in text block") for citation in block["citations"]: if citation.get("type") == "web_search_result_location": url = citation.get("url", "") title = citation.get("title", "") if url and url not in citation_map: citations.append(url) citation_map[url] = len(citations) if self.valves.DEBUG_CITATIONS: print(f"DEBUG: Added citation: {url}") elif block.get("type") == "server_tool_use": # This is Claude deciding to use a tool - we can skip this for citations if self.valves.DEBUG_CITATIONS: print(f"DEBUG: Found server_tool_use block: {block.get('name', 'unknown')}") continue elif block.get("type") == "web_search_tool_result": # Extract URLs from search results if "content" in block: if self.valves.DEBUG_CITATIONS: print(f"DEBUG: Found web_search_tool_result with {len(block['content'])} results") for result in block["content"]: if result.get("type") == "web_search_result": url = result.get("url", "") if url and url not in citation_map: citations.append(url) citation_map[url] = len(citations) if self.valves.DEBUG_CITATIONS: print(f"DEBUG: Added search result URL: {url}") if self.valves.DEBUG_CITATIONS: print(f"DEBUG: Final citations count: {len(citations)}") return " ".join(text_parts), citations def format_response_with_citations(self, content: str, citations: List[str]) -> str: """Format the response by appending citations at the end.""" if not citations: return content formatted_response = content + "\n\nReferences:\n" for i, url in enumerate(citations, 1): formatted_response += f"[{i}] {url}\n" return formatted_response 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) # Add web search tool if enabled if self.valves.ENABLE_WEB_SEARCH: payload["tools"] = [{ "type": "web_search_20250305", "name": "web_search", "max_uses": self.valves.MAX_SEARCH_USES }] 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 citations = [] citation_map = {} content_blocks = [] # Store all content blocks to extract citations at the end 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) # Handle different event types if data["type"] == "content_block_start": content_block = data["content_block"] if content_block["type"] == "text": yield content_block.get("text", "") # Store all content blocks for citation extraction content_blocks.append(content_block) elif data["type"] == "content_block_delta": delta = data["delta"] if delta["type"] == "text_delta": yield delta["text"] elif data["type"] == "content_block_stop": # Store the complete content block for citation extraction if "content_block" in data: content_blocks.append(data["content_block"]) else: # Sometimes the block data is in the event itself block_data = {k: v for k, v in data.items() if k != "type"} if block_data: content_blocks.append(block_data) elif data["type"] == "message_delta": # Capture any final content with citations if "delta" in data and "content" in data["delta"]: for block in data["delta"]["content"]: content_blocks.append(block) if block.get("type") == "text" and block.get("text"): yield block["text"] elif data["type"] == "message_stop": # Extract citations from all collected content blocks if content_blocks: _, citations = self.extract_citations_from_response(content_blocks) # Append citations at the end if any were found if citations: yield "\n\nReferences:\n" + "\n".join( f"[{i}] {url}" for i, url in enumerate(citations, 1) ) 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() if self.valves.DEBUG_CITATIONS: print(f"DEBUG: Full API response keys: {list(result.keys())}") if "content" in result: print(f"DEBUG: Response has {len(result['content'])} content blocks") if "content" in result and result["content"]: # Extract text and citations from content blocks content, citations = self.extract_citations_from_response(result["content"]) return self.format_response_with_citations(content, citations) else: return "" 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