""" title: Vertex AI Claude Models author: fcakyon version: 1.0.0 license: MIT description: Access Claude 4 Opus & Sonnet via Google Vertex AI with thinking mode, streaming, vision, and tool support. Works best when deployed in Google Cloud or as a Pipeline. """ import os import json import logging import asyncio import aiohttp import time import base64 from datetime import datetime, timedelta from typing import List, Union, Dict, Optional, AsyncGenerator from pydantic import BaseModel, Field from open_webui.utils.misc import pop_system_message class Pipe: ANTHROPIC_VERSION = "vertex-2023-10-16" SUPPORTED_IMAGE_TYPES = ["image/jpeg", "image/png", "image/gif", "image/webp"] MAX_IMAGE_SIZE = 5 * 1024 * 1024 MAX_PDF_SIZE = 32 * 1024 * 1024 TOTAL_MAX_IMAGE_SIZE = 100 * 1024 * 1024 MAX_IMAGES_PER_REQUEST = 100 REQUEST_TIMEOUT = 300 MODEL_MAX_TOKENS = { "claude-opus-4@20250514": 32000, "claude-sonnet-4@20250514": 64000, } class Valves(BaseModel): GOOGLE_CLOUD_PROJECT: str = Field( default="", description="🔧 Your Google Cloud Project ID (required)", ) GOOGLE_CLOUD_LOCATION: str = Field( default="us-east5", description="🌍 Google Cloud location (us-east5, global, europe-west1)", ) SERVICE_ACCOUNT_JSON: str = Field( default="", description="🔑 Service Account JSON credentials (paste full JSON content)", ) THINKING_BUDGET_TOKENS: int = Field( default=16000, ge=1000, le=96000, description="🧠 Max tokens for thinking mode (1000-96000)" ) ENABLE_THINKING: bool = Field( default=False, description="🤔 Enable Claude's extended thinking process by default", ) REQUEST_TIMEOUT: int = Field( default=300, ge=30, le=600, description="⏱️ Request timeout in seconds (30-600)" ) def __init__(self): self.type = "manifold" self.id = "vertex_ai_claude" self.valves = self.Valves() self._access_token = None self._token_expires_at = None def get_vertex_ai_models(self) -> List[dict]: return [ { "id": "claude-sonnet-4@20250514", "name": "🎯 Claude Sonnet 4 (Balanced Performance & Cost)", "context_length": 200000, "supports_vision": True, }, { "id": "claude-opus-4@20250514", "name": "🚀 Claude Opus 4 (Most Powerful & Advanced)", "context_length": 200000, "supports_vision": True, }, ] def pipes(self) -> List[dict]: return self.get_vertex_ai_models() def _get_access_token(self) -> str: """Get Google Cloud access token using service account JSON""" if not self.valves.SERVICE_ACCOUNT_JSON: raise Exception("SERVICE_ACCOUNT_JSON is required. Please paste your service account JSON in the valves.") # Check if we have a valid cached token if self._access_token and self._token_expires_at and datetime.now() < self._token_expires_at: return self._access_token try: sa_data = json.loads(self.valves.SERVICE_ACCOUNT_JSON) # Use service account key for direct authentication # This is a simplified approach for marketplace compatibility import urllib.request import urllib.parse # Create a basic authentication token # Note: This is a simplified implementation # For production, you should use proper JWT signing libraries # Try to get token from metadata server if running in GCP try: metadata_url = "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token" request = urllib.request.Request(metadata_url) request.add_header('Metadata-Flavor', 'Google') with urllib.request.urlopen(request, timeout=2) as response: result = json.loads(response.read().decode()) self._access_token = result["access_token"] self._token_expires_at = datetime.now() + timedelta(seconds=result.get("expires_in", 3600) - 300) return self._access_token except: # If metadata server fails, we need external libraries for proper JWT signing # This is a limitation of marketplace functions without external dependencies raise Exception( "❌ Authentication Failed: This function requires proper JWT signing libraries for service account authentication.\n\n" "💡 Solutions:\n" "1. Run OpenWebUI in Google Cloud (uses metadata server)\n" "2. Use the Pipeline version with full dependencies\n" "3. Use Application Default Credentials (gcloud auth application-default login)\n\n" "🔧 For full functionality, deploy this as a Pipeline instead of a Function." ) except json.JSONDecodeError: raise Exception("Invalid JSON in SERVICE_ACCOUNT_JSON. Please check the format.") except Exception as e: raise Exception(f"Authentication failed: {str(e)}") def get_endpoint_url(self, model_name: str) -> str: location = self.valves.GOOGLE_CLOUD_LOCATION project_id = self.valves.GOOGLE_CLOUD_PROJECT if location == "global": endpoint = "https://aiplatform.googleapis.com" else: endpoint = f"https://{location}-aiplatform.googleapis.com" return f"{endpoint}/v1/projects/{project_id}/locations/{location}/publishers/anthropic/models/{model_name}:streamRawPredict" def process_content(self, content: Union[str, List[dict]]) -> List[dict]: if isinstance(content, str): return [{"type": "text", "text": content}] processed_content = [] for item in content: if item["type"] == "text": processed_content.append({"type": "text", "text": item["text"]}) elif item["type"] == "image_url": processed_content.append(self.process_image(item)) elif item["type"] == "pdf_url": processed_content.append(self.process_pdf(item)) elif item["type"] == "document": processed_content.append(self.process_document(item)) elif item["type"] in ["tool_calls", "tool_results"]: processed_content.append(item) return processed_content def process_image(self, image_data): if image_data["image_url"]["url"].startswith("data:image"): mime_type, base64_data = image_data["image_url"]["url"].split(",", 1) media_type = mime_type.split(":")[1].split(";")[0] if media_type not in self.SUPPORTED_IMAGE_TYPES: raise ValueError(f"Unsupported media type: {media_type}") 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):.1f}MB limit") return { "type": "image", "source": { "type": "base64", "media_type": media_type, "data": base64_data, }, } return { "type": "image", "source": {"type": "url", "url": image_data["image_url"]["url"]}, } def process_pdf(self, pdf_data): if pdf_data["pdf_url"]["url"].startswith("data:application/pdf"): mime_type, base64_data = pdf_data["pdf_url"]["url"].split(",", 1) 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):.1f}MB limit") document = { "type": "document", "source": { "type": "base64", "media_type": "application/pdf", "data": base64_data, }, } if pdf_data.get("cache_control"): document["cache_control"] = pdf_data["cache_control"] return document document = { "type": "document", "source": {"type": "url", "url": pdf_data["pdf_url"]["url"]}, } if pdf_data.get("cache_control"): document["cache_control"] = pdf_data["cache_control"] return document def process_document(self, doc_data): document = {"type": "document", "source": doc_data.get("source")} for field in ["title", "context", "citations", "cache_control"]: if field in doc_data: document[field] = doc_data[field] return document def convert_openai_tools_to_anthropic(self, tools: List[dict]) -> List[dict]: anthropic_tools = [] for tool in tools: if tool.get("type") == "function": function = tool.get("function", {}) anthropic_tools.append({ "name": function.get("name"), "description": function.get("description"), "input_schema": function.get("parameters", {}), }) return anthropic_tools def validate_total_image_size(self, messages: List[dict]) -> None: total_size = image_count = 0 for message in messages: if isinstance(message.get("content"), list): for item in message["content"]: if (item.get("type") == "image_url" and item["image_url"]["url"].startswith("data:image")): _, base64_data = item["image_url"]["url"].split(",", 1) total_size += len(base64_data) * 3 / 4 image_count += 1 if total_size > self.TOTAL_MAX_IMAGE_SIZE: raise ValueError(f"Total image size exceeds {self.TOTAL_MAX_IMAGE_SIZE/(1024*1024):.1f}MB") if image_count > self.MAX_IMAGES_PER_REQUEST: raise ValueError(f"Too many images: {image_count}. Maximum: {self.MAX_IMAGES_PER_REQUEST}") def _model_supports_thinking(self, model_name: str) -> bool: return model_name in self.MODEL_MAX_TOKENS async def pipe(self, body: Dict) -> Union[str, AsyncGenerator[str, None]]: # Validation if not self.valves.GOOGLE_CLOUD_PROJECT: return "❌ Error: GOOGLE_CLOUD_PROJECT is required. Please configure it in the function settings." if not self.valves.SERVICE_ACCOUNT_JSON: return "❌ Error: SERVICE_ACCOUNT_JSON is required. Please paste your service account JSON in the function settings." try: system_message, messages = pop_system_message(body["messages"]) self.validate_total_image_size(messages) model_name = body["model"].replace("api/", "").split("/")[-1] max_tokens_limit = self.MODEL_MAX_TOKENS.get(model_name, 4096) # Build payload payload = { "anthropic_version": self.ANTHROPIC_VERSION, "messages": self._process_messages(messages), "max_tokens": min(body.get("max_tokens", max_tokens_limit), max_tokens_limit), "stream": body.get("stream", True), } # Add optional parameters for param in ["temperature", "top_k", "top_p"]: if body.get(param) is not None: payload[param] = float(body[param]) if param != "top_k" else int(body[param]) if system_message: payload["system"] = str(system_message) # Handle tools if "tools" in body: payload["tools"] = self.convert_openai_tools_to_anthropic(body["tools"]) tool_choice = body.get("tool_choice", "auto") if tool_choice == "none": payload["tool_choice"] = {"type": "none"} elif tool_choice == "auto": payload["tool_choice"] = {"type": "auto"} elif isinstance(tool_choice, dict) and tool_choice.get("function"): payload["tool_choice"] = {"type": "tool", "name": tool_choice["function"]["name"]} # Add thinking mode if (self._model_supports_thinking(model_name) and (self.valves.ENABLE_THINKING or body.get("thinking"))): payload["thinking"] = { "type": "enabled", "budget_tokens": self.valves.THINKING_BUDGET_TOKENS, } # Get authentication and make request access_token = self._get_access_token() headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", } url = self.get_endpoint_url(model_name) if payload["stream"]: return self._stream_response(url, headers, payload) else: return await self._sync_response(url, headers, payload) except Exception as e: return f"❌ Error: {str(e)}" async def _stream_response(self, url: str, headers: dict, payload: dict) -> AsyncGenerator[str, None]: try: async with aiohttp.ClientSession() as session: async with session.post( url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=self.valves.REQUEST_TIMEOUT) ) as response: if response.status != 200: error_text = await response.text() yield f"❌ HTTP {response.status}: {error_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": delta = data.get("delta", {}) if delta.get("type") == "text_delta": yield delta.get("text", "") elif delta.get("type") == "citations_delta": citation = delta.get("citation", {}) if citation: doc_index = citation.get("document_index", 0) doc_title = citation.get("document_title", "") yield f" [{doc_index}: {doc_title}]" if doc_title else f" [{doc_index}]" elif data["type"] == "message_stop": break except json.JSONDecodeError: continue except asyncio.TimeoutError: yield "❌ Request timed out" except Exception as e: yield f"❌ Stream error: {str(e)}" async def _sync_response(self, url: str, headers: dict, payload: dict) -> str: try: async with aiohttp.ClientSession() as session: async with session.post( url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=self.valves.REQUEST_TIMEOUT) ) as response: if response.status != 200: error_text = await response.text() return f"❌ HTTP {response.status}: {error_text}" result = await response.json() return result["content"][0]["text"] except Exception as e: return f"❌ Request error: {str(e)}" def _process_messages(self, messages: List[dict]) -> List[dict]: processed_messages = [] for message in messages: processed_content = [] for content in self.process_content(message["content"]): # Add cache control for tool calls/results if ((message.get("role") == "assistant" and content.get("type") == "tool_calls") or (message.get("role") == "user" and content.get("type") == "tool_results")): content["cache_control"] = {"type": "ephemeral"} processed_content.append(content) processed_messages.append({"role": message["role"], "content": processed_content}) return processed_messages