import json import logging import aiohttp from open_webui.env import ( AIOHTTP_CLIENT_SESSION_SSL, AIOHTTP_CLIENT_TIMEOUT_MODEL_LIST, ENABLE_FORWARD_USER_INFO_HEADERS, ) from open_webui.models.users import UserModel from open_webui.utils.headers import include_user_info_headers log = logging.getLogger(__name__) def is_anthropic_url(url: str) -> bool: """Check if the URL is an Anthropic API endpoint.""" return 'api.anthropic.com' in url async def get_anthropic_models(url: str, key: str, user: UserModel = None) -> dict: """ Fetch models from Anthropic's /v1/models endpoint with pagination. Normalizes the response to OpenAI format. """ timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_MODEL_LIST) all_models = [] after_id = None try: async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: headers = { 'x-api-key': key, 'anthropic-version': '2023-06-01', } if ENABLE_FORWARD_USER_INFO_HEADERS and user: headers = include_user_info_headers(headers, user) while True: params = {'limit': 1000} if after_id: params['after_id'] = after_id async with session.get( f'{url}/models', headers=headers, params=params, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as response: if response.status != 200: error_detail = f'HTTP Error: {response.status}' try: res = await response.json() if 'error' in res: error_detail = f'External Error: {res["error"]}' except Exception: pass return {'object': 'list', 'data': [], 'error': error_detail} data = await response.json() for model in data.get('data', []): all_models.append( { 'id': model.get('id'), 'object': 'model', 'created': 0, 'owned_by': 'anthropic', 'name': model.get('display_name', model.get('id')), } ) if not data.get('has_more', False): break after_id = data.get('last_id') except Exception as e: log.error(f'Anthropic connection error: {e}') return None return {'object': 'list', 'data': all_models} ############################## # # Anthropic Messages API Conversion Utilities # ############################## def convert_anthropic_to_openai_payload(anthropic_payload: dict) -> dict: """ Convert an Anthropic Messages API request to OpenAI Chat Completions format. Anthropic format: {model, messages: [{role, content}], system, max_tokens, ...} OpenAI format: {model, messages: [{role, content}], max_tokens, ...} """ openai_payload = {} # Model openai_payload['model'] = anthropic_payload.get('model', '') # Build messages list messages = [] # System prompt (Anthropic has it as top-level, OpenAI as a system message) system = anthropic_payload.get('system') if system: if isinstance(system, str): messages.append({'role': 'system', 'content': system}) elif isinstance(system, list): # Anthropic supports system as list of content blocks text_parts = [] for block in system: if isinstance(block, dict) and block.get('type') == 'text': text_parts.append(block.get('text', '')) elif isinstance(block, str): text_parts.append(block) messages.append({'role': 'system', 'content': '\n'.join(text_parts)}) # Convert messages for msg in anthropic_payload.get('messages', []): role = msg.get('role', 'user') content = msg.get('content') if isinstance(content, str): messages.append({'role': role, 'content': content}) elif isinstance(content, list): # Convert Anthropic content blocks to OpenAI format openai_content = [] tool_calls = [] for block in content: block_type = block.get('type', 'text') if block_type == 'text': openai_content.append( { 'type': 'text', 'text': block.get('text', ''), } ) elif block_type == 'image': source = block.get('source', {}) if source.get('type') == 'base64': media_type = source.get('media_type', 'image/png') data = source.get('data', '') openai_content.append( { 'type': 'image_url', 'image_url': { 'url': f'data:{media_type};base64,{data}', }, } ) elif source.get('type') == 'url': openai_content.append( { 'type': 'image_url', 'image_url': {'url': source.get('url', '')}, } ) elif block_type == 'tool_use': tool_calls.append( { 'id': block.get('id', ''), 'type': 'function', 'function': { 'name': block.get('name', ''), 'arguments': ( json.dumps(block.get('input', {})) if isinstance(block.get('input'), dict) else str(block.get('input', '{}')) ), }, } ) elif block_type == 'tool_result': # Tool results become separate tool messages in OpenAI format tool_content = block.get('content', '') if isinstance(tool_content, list): tool_text_parts = [] for tc in tool_content: if isinstance(tc, dict) and tc.get('type') == 'text': tool_text_parts.append(tc.get('text', '')) tool_content = '\n'.join(tool_text_parts) # Propagate error status if present if block.get('is_error'): tool_content = f'Error: {tool_content}' messages.append( { 'role': 'tool', 'tool_call_id': block.get('tool_use_id', ''), 'content': tool_content, } ) # Build the message if tool_calls: # Assistant message with tool calls msg_dict = {'role': role} if openai_content: # If there's only text, flatten it if len(openai_content) == 1 and openai_content[0]['type'] == 'text': msg_dict['content'] = openai_content[0]['text'] else: msg_dict['content'] = openai_content else: msg_dict['content'] = '' msg_dict['tool_calls'] = tool_calls messages.append(msg_dict) elif openai_content: # If there's only a single text block, flatten it to a string if len(openai_content) == 1 and openai_content[0]['type'] == 'text': messages.append({'role': role, 'content': openai_content[0]['text']}) else: messages.append({'role': role, 'content': openai_content}) else: messages.append({'role': role, 'content': str(content) if content else ''}) openai_payload['messages'] = messages # max_tokens if 'max_tokens' in anthropic_payload: openai_payload['max_tokens'] = anthropic_payload['max_tokens'] # Common parameters for param in ('temperature', 'top_p', 'stop_sequences', 'stream'): if param in anthropic_payload: if param == 'stop_sequences': openai_payload['stop'] = anthropic_payload[param] else: openai_payload[param] = anthropic_payload[param] # Tools conversion: Anthropic → OpenAI if 'tools' in anthropic_payload: openai_tools = [] for tool in anthropic_payload['tools']: openai_tools.append( { 'type': 'function', 'function': { 'name': tool.get('name', ''), 'description': tool.get('description', ''), 'parameters': tool.get('input_schema', {}), }, } ) openai_payload['tools'] = openai_tools # tool_choice if 'tool_choice' in anthropic_payload: tc = anthropic_payload['tool_choice'] if isinstance(tc, dict): tc_type = tc.get('type', 'auto') if tc_type == 'auto': openai_payload['tool_choice'] = 'auto' elif tc_type == 'any': openai_payload['tool_choice'] = 'required' elif tc_type == 'tool': openai_payload['tool_choice'] = { 'type': 'function', 'function': {'name': tc.get('name', '')}, } return openai_payload def convert_openai_to_anthropic_response(openai_response: dict, model: str = '') -> dict: """ Convert a non-streaming OpenAI Chat Completions response to Anthropic Messages format. """ import uuid as _uuid choice = {} if openai_response.get('choices'): choice = openai_response['choices'][0] message = choice.get('message', {}) finish_reason = choice.get('finish_reason', 'stop') # Map finish_reason to stop_reason stop_reason_map = { 'stop': 'end_turn', 'length': 'max_tokens', 'tool_calls': 'tool_use', 'content_filter': 'end_turn', } stop_reason = stop_reason_map.get(finish_reason, 'end_turn') # Build content blocks content = [] msg_content = message.get('content') if msg_content: content.append({'type': 'text', 'text': msg_content}) # Tool calls → tool_use blocks tool_calls = message.get('tool_calls', []) for tc in tool_calls: func = tc.get('function', {}) try: tool_input = json.loads(func.get('arguments', '{}')) except (json.JSONDecodeError, TypeError): tool_input = {} content.append( { 'type': 'tool_use', 'id': tc.get('id', f'toolu_{_uuid.uuid4().hex[:24]}'), 'name': func.get('name', ''), 'input': tool_input, } ) # Usage openai_usage = openai_response.get('usage', {}) usage = { 'input_tokens': openai_usage.get('prompt_tokens', 0), 'output_tokens': openai_usage.get('completion_tokens', 0), } return { 'id': openai_response.get('id', f'msg_{_uuid.uuid4().hex[:24]}'), 'type': 'message', 'role': 'assistant', 'content': content, 'model': model or openai_response.get('model', ''), 'stop_reason': stop_reason, 'stop_sequence': None, 'usage': usage, } async def openai_stream_to_anthropic_stream(openai_stream_generator, model: str = ''): """ Convert an OpenAI SSE streaming response to Anthropic Messages SSE format. OpenAI sends: data: {"choices": [{"delta": {"content": "..."}}]} Anthropic sends: event: content_block_delta\\ndata: {"type": "content_block_delta", ...} Handles text content, tool calls, and mixed content with proper multi-block indexing as required by Anthropic's streaming protocol. """ import uuid as _uuid msg_id = f'msg_{_uuid.uuid4().hex[:24]}' input_tokens = 0 output_tokens = 0 stop_reason = 'end_turn' # Track content blocks with a running index. # Each text block or tool_use block gets its own index. current_block_index = 0 text_block_open = False # Track tool call state: maps OpenAI tool_call index -> Anthropic block index # This allows handling multiple concurrent tool calls. tool_call_blocks = {} # {openai_tc_index: anthropic_block_index} tool_call_started = {} # {openai_tc_index: bool} # Emit message_start message_start = { 'type': 'message_start', 'message': { 'id': msg_id, 'type': 'message', 'role': 'assistant', 'content': [], 'model': model, 'stop_reason': None, 'stop_sequence': None, 'usage': {'input_tokens': 0, 'output_tokens': 0}, }, } yield f'event: message_start\ndata: {json.dumps(message_start)}\n\n'.encode() try: async for chunk in openai_stream_generator: if isinstance(chunk, bytes): chunk = chunk.decode('utf-8', errors='ignore') for line in chunk.strip().split('\n'): line = line.strip() if not line or not line.startswith('data:'): continue data_str = line[5:].strip() if data_str == '[DONE]': continue if data_str == '{}': continue try: data = json.loads(data_str) except (json.JSONDecodeError, TypeError): continue choices = data.get('choices', []) if not choices: # Check for usage in the final chunk if data.get('usage'): input_tokens = data['usage'].get('prompt_tokens', input_tokens) output_tokens = data['usage'].get('completion_tokens', output_tokens) continue delta = choices[0].get('delta', {}) finish_reason = choices[0].get('finish_reason') # Update usage if present if data.get('usage'): input_tokens = data['usage'].get('prompt_tokens', input_tokens) output_tokens = data['usage'].get('completion_tokens', output_tokens) # --- Handle text content --- content = delta.get('content') if content is not None: if not text_block_open: # Start a new text content block block_start = { 'type': 'content_block_start', 'index': current_block_index, 'content_block': {'type': 'text', 'text': ''}, } yield f'event: content_block_start\ndata: {json.dumps(block_start)}\n\n'.encode() text_block_open = True # Send text delta block_delta = { 'type': 'content_block_delta', 'index': current_block_index, 'delta': {'type': 'text_delta', 'text': content}, } yield f'event: content_block_delta\ndata: {json.dumps(block_delta)}\n\n'.encode() # --- Handle tool calls --- tool_calls = delta.get('tool_calls') if tool_calls: # Close text block if one is open (text comes before tools) if text_block_open: block_stop = { 'type': 'content_block_stop', 'index': current_block_index, } yield f'event: content_block_stop\ndata: {json.dumps(block_stop)}\n\n'.encode() text_block_open = False current_block_index += 1 for tc in tool_calls: tc_index = tc.get('index', 0) if tc_index not in tool_call_started: # First time seeing this tool call — emit content_block_start tool_call_blocks[tc_index] = current_block_index tool_call_started[tc_index] = True # Extract tool call ID and name from the first chunk tc_id = tc.get('id', f'toolu_{_uuid.uuid4().hex[:24]}') tc_name = tc.get('function', {}).get('name', '') block_start = { 'type': 'content_block_start', 'index': current_block_index, 'content_block': { 'type': 'tool_use', 'id': tc_id, 'name': tc_name, 'input': {}, }, } yield f'event: content_block_start\ndata: {json.dumps(block_start)}\n\n'.encode() current_block_index += 1 # Emit argument chunks as input_json_delta args_chunk = tc.get('function', {}).get('arguments', '') if args_chunk: block_delta = { 'type': 'content_block_delta', 'index': tool_call_blocks[tc_index], 'delta': { 'type': 'input_json_delta', 'partial_json': args_chunk, }, } yield f'event: content_block_delta\ndata: {json.dumps(block_delta)}\n\n'.encode() # --- Handle finish reason --- if finish_reason is not None: stop_reason_map = { 'stop': 'end_turn', 'length': 'max_tokens', 'tool_calls': 'tool_use', } stop_reason = stop_reason_map.get(finish_reason, 'end_turn') except Exception as e: log.error(f'Error in Anthropic stream conversion: {e}') # Close any open text block if text_block_open: block_stop = {'type': 'content_block_stop', 'index': current_block_index} yield f'event: content_block_stop\ndata: {json.dumps(block_stop)}\n\n'.encode() # Close any open tool call blocks for tc_index, block_index in tool_call_blocks.items(): block_stop = {'type': 'content_block_stop', 'index': block_index} yield f'event: content_block_stop\ndata: {json.dumps(block_stop)}\n\n'.encode() # Emit message_delta with stop reason message_delta = { 'type': 'message_delta', 'delta': { 'stop_reason': stop_reason, 'stop_sequence': None, }, 'usage': {'output_tokens': output_tokens}, } yield f'event: message_delta\ndata: {json.dumps(message_delta)}\n\n'.encode() # Emit message_stop yield f'event: message_stop\ndata: {json.dumps({"type": "message_stop"})}\n\n'.encode()