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 from open_webui.utils.json_codec import JSONCodec log = logging.getLogger(__name__) ANTHROPIC_CONVERTED_REQUEST_PARAMS = { 'model', 'messages', 'system', 'max_tokens', 'temperature', 'top_p', 'top_k', 'stop_sequences', 'stream', 'metadata', 'service_tier', 'tools', 'tool_choice', 'reasoning_effort', } 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 _copy_cache_control(source: dict, target: dict) -> dict: if isinstance(source, dict) and 'cache_control' in source: target['cache_control'] = source['cache_control'] return target def _has_cache_control(blocks: list) -> bool: return any(isinstance(block, dict) and 'cache_control' in block for block in blocks) def _finalize_openai_content(blocks: list) -> str | list: if not blocks: return '' if len(blocks) == 1 and blocks[0].get('type') == 'text' and not _has_cache_control(blocks): return blocks[0].get('text', '') return blocks def is_anthropic_messages_passthrough(url: str, api_config: dict | None = None) -> bool: api_config = api_config or {} provider = str(api_config.get('provider', '')).lower() return is_anthropic_url(url or '') or provider == 'litellm' def convert_anthropic_to_openai_payload( anthropic_payload: dict, passthrough_params: list[str] | str | None = None ) -> 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): openai_content = [] for block in system: if isinstance(block, dict) and block.get('type') == 'text': openai_content.append( _copy_cache_control( block, { 'type': 'text', 'text': block.get('text', ''), }, ) ) elif isinstance(block, str): openai_content.append({'type': 'text', 'text': block}) messages.append({'role': 'system', 'content': _finalize_openai_content(openai_content)}) # 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( _copy_cache_control( block, { 'type': 'text', 'text': block.get('text', ''), }, ) ) elif block_type in ('thinking', 'redacted_thinking'): openai_content.append(_copy_cache_control(block, dict(block))) 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( _copy_cache_control( block, { 'type': 'image_url', 'image_url': { 'url': f'data:{media_type};base64,{data}', }, }, ) ) elif source.get('type') == 'url': openai_content.append( _copy_cache_control( block, { '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': ( JSONCodec.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_result_content = block.get('content', '') tool_content: str | list = '' if isinstance(tool_result_content, str): tool_content = tool_result_content elif isinstance(tool_result_content, list): # Build a multimodal content array to preserve # images and other non-text content types. converted_parts = [] for content_block in tool_result_content: if not isinstance(content_block, dict): continue content_type = content_block.get('type', 'text') if content_type == 'text': converted_parts.append( _copy_cache_control( content_block, { 'type': 'text', 'text': content_block.get('text', ''), }, ) ) elif content_type == 'image': source = content_block.get('source', {}) if source.get('type') == 'base64': media_type = source.get('media_type', 'image/png') data = source.get('data', '') converted_parts.append( _copy_cache_control( content_block, { 'type': 'image_url', 'image_url': { 'url': f'data:{media_type};base64,{data}', }, }, ) ) elif source.get('type') == 'url': converted_parts.append( _copy_cache_control( content_block, { 'type': 'image_url', 'image_url': { 'url': source.get('url', ''), }, }, ) ) elif content_type == 'document': # Documents have no direct OpenAI equivalent; # convert to a text representation. document_source = content_block.get('source', {}) document_title = content_block.get('title', 'Document') document_context = content_block.get('context', '') document_text = f'[Document: {document_title}]' if document_context: document_text += f'\n{document_context}' if document_source.get('type') == 'text' and document_source.get('data'): document_text += f'\n{document_source["data"]}' converted_parts.append({'type': 'text', 'text': document_text}) elif content_type == 'search_result': # Convert search results to a text # representation with source attribution. search_title = content_block.get('title', '') search_url = content_block.get('source', '') search_content_blocks = content_block.get('content', []) search_texts = [] for search_block in search_content_blocks: if isinstance(search_block, dict) and search_block.get('type') == 'text': search_texts.append(search_block.get('text', '')) search_body = '\n'.join(search_texts) search_text = f'[Search Result: {search_title}]' if search_url: search_text += f'\nSource: {search_url}' if search_body: search_text += f'\n{search_body}' converted_parts.append({'type': 'text', 'text': search_text}) # Flatten to string when only text parts are present if all(part.get('type') == 'text' for part in converted_parts) and not _has_cache_control( converted_parts ): tool_content = '\n'.join(part.get('text', '') for part in converted_parts) elif converted_parts: tool_content = converted_parts else: tool_content = '' # Propagate error status if present if block.get('is_error'): if isinstance(tool_content, str): tool_content = f'Error: {tool_content}' elif isinstance(tool_content, list): tool_content.insert( 0, { 'type': 'text', 'text': 'Error: ', }, ) 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: msg_dict['content'] = _finalize_openai_content(openai_content) else: msg_dict['content'] = '' msg_dict['tool_calls'] = tool_calls messages.append(msg_dict) elif openai_content: messages.append({'role': role, 'content': _finalize_openai_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'] captured_passthrough_params = { param: value for param, value in anthropic_payload.items() if param not in ANTHROPIC_CONVERTED_REQUEST_PARAMS } if isinstance(passthrough_params, str): passthrough_params = passthrough_params.split(',') elif not isinstance(passthrough_params, (list, tuple, set)): passthrough_params = [] passthrough_param_names = {str(item).strip() for item in passthrough_params if str(item).strip()} if '*' in passthrough_param_names: openai_payload.update(captured_passthrough_params) else: for param in passthrough_param_names: if param in captured_passthrough_params: openai_payload[param] = captured_passthrough_params[param] output_config = anthropic_payload.get('output_config') if isinstance(output_config, dict): if 'effort' in output_config and 'reasoning_effort' not in anthropic_payload: openai_payload['reasoning_effort'] = output_config['effort'] format_config = output_config.get('format') if isinstance(format_config, dict): format_type = format_config.get('type') if format_type == 'json_schema': json_schema = { 'name': format_config.get('name', 'response_schema'), 'schema': format_config.get('schema', {}), } if 'description' in format_config: json_schema['description'] = format_config['description'] if 'strict' in format_config: json_schema['strict'] = format_config['strict'] openai_payload['response_format'] = { 'type': 'json_schema', 'json_schema': json_schema, } elif format_type == 'json_object': openai_payload['response_format'] = {'type': format_type} if 'reasoning_effort' in anthropic_payload: openai_payload['reasoning_effort'] = anthropic_payload['reasoning_effort'] # Common parameters for param in ('temperature', 'top_p', 'top_k', 'stop_sequences', 'stream', 'metadata', 'service_tier'): 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( _copy_cache_control( tool, { '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: tool_choice = anthropic_payload['tool_choice'] if isinstance(tool_choice, dict): tool_choice_type = tool_choice.get('type', 'auto') if tool_choice_type == 'auto': openai_payload['tool_choice'] = 'auto' elif tool_choice_type == 'any': openai_payload['tool_choice'] = 'required' elif tool_choice_type == 'tool': openai_payload['tool_choice'] = { 'type': 'function', 'function': {'name': tool_choice.get('name', '')}, } return openai_payload def convert_openai_to_anthropic_response( openai_response: dict, model: str = '', input_tokens: int | None = None ) -> 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 = [] message_thinking = message.get('thinking') thinking_blocks = message.get('thinking_blocks') or [] if not thinking_blocks and isinstance(message_thinking, dict): thinking_blocks = message_thinking.get('blocks') or [] has_thinking = False for block in thinking_blocks: if not isinstance(block, dict): continue if block.get('type') == 'redacted_thinking': content.append({k: v for k, v in block.items() if k in {'type', 'data'}}) has_thinking = True continue thinking = block.get('thinking') or block.get('content') or block.get('text') if not thinking: continue thinking_block = {'type': 'thinking', 'thinking': thinking} if block.get('signature'): thinking_block['signature'] = block['signature'] content.append(thinking_block) has_thinking = True reasoning_content = message.get('reasoning_content') or message.get('reasoning') if not reasoning_content and isinstance(message_thinking, str): reasoning_content = message_thinking if reasoning_content and not has_thinking: content.append({'type': 'thinking', 'thinking': reasoning_content}) message_content = message.get('content') if message_content: content.append({'type': 'text', 'text': message_content}) # Tool calls -> tool_use blocks tool_calls = message.get('tool_calls') or [] for tool_call in tool_calls: function = tool_call.get('function', {}) try: tool_input = JSONCodec.loads(function.get('arguments', '{}')) except (JSONCodec.JSONDecodeError, TypeError): tool_input = {} content.append( { 'type': 'tool_use', 'id': tool_call.get('id', f'toolu_{_uuid.uuid4().hex[:24]}'), 'name': function.get('name', ''), 'input': tool_input, } ) # Usage openai_usage = openai_response.get('usage') or {} cache_creation = openai_usage.get('cache_creation_input_tokens') cache_read = openai_usage.get('cache_read_input_tokens') prompt_details = openai_usage.get('prompt_tokens_details') if cache_read is None and isinstance(prompt_details, dict): cache_read = prompt_details.get('cached_tokens') usage_input = openai_usage.get('input_tokens') if usage_input is None: prompt_tokens = openai_usage.get('prompt_tokens') if prompt_tokens is not None: usage_input = max(prompt_tokens - (cache_creation or 0) - (cache_read or 0), 0) usage_output = openai_usage.get('output_tokens') if usage_output is None: usage_output = openai_usage.get('completion_tokens') usage = { 'input_tokens': usage_input if usage_input is not None else (input_tokens if input_tokens is not None else 0), 'output_tokens': usage_output if usage_output is not None else 0, } if cache_creation is not None: usage['cache_creation_input_tokens'] = cache_creation if cache_read is not None: usage['cache_read_input_tokens'] = cache_read if isinstance(openai_usage.get('output_tokens_details'), dict): usage['output_tokens_details'] = openai_usage['output_tokens_details'] if isinstance(openai_usage.get('server_tool_use'), dict): usage['server_tool_use'] = openai_usage['server_tool_use'] if openai_usage.get('service_tier') is not None: usage['service_tier'] = openai_usage['service_tier'] 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 = '', input_tokens: int | None = None): """ 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. Tool calls are tracked by their unique id (not OpenAI index) so that parallel calls sharing the same index get distinct Anthropic tool_use blocks. Each block follows the Anthropic lifecycle: start -> delta -> stop. """ import uuid as _uuid message_id = f'msg_{_uuid.uuid4().hex[:24]}' output_tokens = 0 cache_creation_input_tokens = None cache_read_input_tokens = None output_tokens_details = None server_tool_use = None service_tier = None 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 thinking_block_open = False text_block_open = False # Accumulated state for each tool call, keyed by tool call id. # Parallel calls that share the same OpenAI index get distinct entries. # Each entry: {id, name, arguments, block_index, started, stopped} tracked_tool_calls = {} # Map OpenAI tool call index -> tool call id for routing # argument-only deltas (deltas that carry arguments but no id). index_to_tool_id = {} # Whether any tool call block has been emitted (suppresses further text) has_tool_calls = False # Emit message_start message_start = { 'type': 'message_start', 'message': { 'id': message_id, 'type': 'message', 'role': 'assistant', 'content': [], 'model': model, 'stop_reason': None, 'stop_sequence': None, 'usage': {'input_tokens': input_tokens or 0, 'output_tokens': 0}, }, } yield f'event: message_start\ndata: {JSONCodec.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_string = line[5:].strip() if data_string == '[DONE]': continue if data_string == '{}': continue try: data = JSONCodec.loads(data_string) except (JSONCodec.JSONDecodeError, TypeError): continue usage_data = data.get('usage') if isinstance(usage_data, dict): cache_creation = usage_data.get('cache_creation_input_tokens') cache_read = usage_data.get('cache_read_input_tokens') prompt_details = usage_data.get('prompt_tokens_details') if cache_read is None and isinstance(prompt_details, dict): cache_read = prompt_details.get('cached_tokens') usage_input = usage_data.get('input_tokens') if usage_input is None: prompt_tokens = usage_data.get('prompt_tokens') if prompt_tokens is not None: usage_input = max(prompt_tokens - (cache_creation or 0) - (cache_read or 0), 0) usage_output = usage_data.get('output_tokens') if usage_output is None: usage_output = usage_data.get('completion_tokens') if usage_input is not None: input_tokens = usage_input if usage_output is not None: output_tokens = usage_output if cache_creation is not None: cache_creation_input_tokens = cache_creation if cache_read is not None: cache_read_input_tokens = cache_read if isinstance(usage_data.get('output_tokens_details'), dict): output_tokens_details = usage_data['output_tokens_details'] if isinstance(usage_data.get('server_tool_use'), dict): server_tool_use = usage_data['server_tool_use'] if usage_data.get('service_tier') is not None: service_tier = usage_data['service_tier'] choices = data.get('choices', []) if not choices: continue delta = choices[0].get('delta', {}) finish_reason = choices[0].get('finish_reason') message = choices[0].get('message') or {} reasoning_content = ( delta.get('reasoning_content') or delta.get('reasoning') or delta.get('thinking') or message.get('reasoning_content') or message.get('reasoning') ) if not reasoning_content: thinking_blocks = delta.get('thinking_blocks') or message.get('thinking_blocks') or [] for block in thinking_blocks: if isinstance(block, dict): reasoning_content = block.get('thinking') or block.get('content') or block.get('text') if reasoning_content: break if reasoning_content and not text_block_open and not has_tool_calls: if not thinking_block_open: block_start = { 'type': 'content_block_start', 'index': current_block_index, 'content_block': {'type': 'thinking', 'thinking': ''}, } yield f'event: content_block_start\ndata: {JSONCodec.dumps(block_start)}\n\n'.encode() thinking_block_open = True block_delta = { 'type': 'content_block_delta', 'index': current_block_index, 'delta': {'type': 'thinking_delta', 'thinking': reasoning_content}, } yield f'event: content_block_delta\ndata: {JSONCodec.dumps(block_delta)}\n\n'.encode() # --- Handle text content --- # Anthropic expects text blocks before tool blocks, so skip # text deltas once any tool call has started. content = delta.get('content') if content and not has_tool_calls: if thinking_block_open: block_stop = { 'type': 'content_block_stop', 'index': current_block_index, } yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode() thinking_block_open = False current_block_index += 1 if not text_block_open: block_start = { 'type': 'content_block_start', 'index': current_block_index, 'content_block': {'type': 'text', 'text': ''}, } yield f'event: content_block_start\ndata: {JSONCodec.dumps(block_start)}\n\n'.encode() text_block_open = True block_delta = { 'type': 'content_block_delta', 'index': current_block_index, 'delta': {'type': 'text_delta', 'text': content}, } yield f'event: content_block_delta\ndata: {JSONCodec.dumps(block_delta)}\n\n'.encode() # --- Handle tool calls --- # Some providers put tool_calls on the final message object # instead of the delta; fall back to that when needed. tool_calls = delta.get('tool_calls') or [] if not tool_calls and message.get('tool_calls'): tool_calls = message['tool_calls'] if tool_calls: # Close text block if one is open (text comes before tools) if thinking_block_open: block_stop = { 'type': 'content_block_stop', 'index': current_block_index, } yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode() thinking_block_open = False current_block_index += 1 if text_block_open: block_stop = { 'type': 'content_block_stop', 'index': current_block_index, } yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode() text_block_open = False current_block_index += 1 for tool_call in tool_calls: tool_call_index = tool_call.get('index', 0) tool_call_id = tool_call.get('id', '') tool_call_name = (tool_call.get('function') or {}).get('name', '') arguments_chunk = (tool_call.get('function') or {}).get('arguments', '') # Resolve which tracked tool call this delta belongs to. # A delta with an id starts or identifies a specific tool. # A delta without an id carries arguments for the most # recent tool at this OpenAI index. if tool_call_id: if tool_call_id not in tracked_tool_calls: tracked_tool_calls[tool_call_id] = { 'id': tool_call_id, 'name': tool_call_name, 'arguments': '', 'block_index': -1, 'started': False, 'stopped': False, } index_to_tool_id[tool_call_index] = tool_call_id tool = tracked_tool_calls[tool_call_id] elif tool_call_index in index_to_tool_id: tool = tracked_tool_calls[index_to_tool_id[tool_call_index]] else: # First delta for this index with no id; create a # provisional entry with a generated fallback id. fallback_id = f'toolu_{_uuid.uuid4().hex[:24]}' tracked_tool_calls[fallback_id] = { 'id': fallback_id, 'name': tool_call_name, 'arguments': '', 'block_index': -1, 'started': False, 'stopped': False, } index_to_tool_id[tool_call_index] = fallback_id tool = tracked_tool_calls[fallback_id] # Update name if provided on a later delta if tool_call_name and not tool['name']: tool['name'] = tool_call_name # Emit content_block_start once we have a name if not tool['started'] and tool['name']: tool['block_index'] = current_block_index tool['started'] = True has_tool_calls = True block_start = { 'type': 'content_block_start', 'index': current_block_index, 'content_block': { 'type': 'tool_use', 'id': tool['id'], 'name': tool['name'], 'input': {}, }, } yield f'event: content_block_start\ndata: {JSONCodec.dumps(block_start)}\n\n'.encode() current_block_index += 1 # Buffer arguments and emit as input_json_delta if arguments_chunk: tool['arguments'] += arguments_chunk if tool['started'] and not tool['stopped']: block_delta = { 'type': 'content_block_delta', 'index': tool['block_index'], 'delta': { 'type': 'input_json_delta', 'partial_json': arguments_chunk, }, } yield f'event: content_block_delta\ndata: {JSONCodec.dumps(block_delta)}\n\n'.encode() # Close the block once arguments form complete JSON if tool['started'] and not tool['stopped']: try: JSONCodec.loads(tool['arguments']) tool['stopped'] = True block_stop = { 'type': 'content_block_stop', 'index': tool['block_index'], } yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode() except (JSONCodec.JSONDecodeError, ValueError): pass # --- 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 thinking block if thinking_block_open: block_stop = {'type': 'content_block_stop', 'index': current_block_index} yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode() current_block_index += 1 # Flush any tools that buffered arguments but never emitted a block for tool in tracked_tool_calls.values(): if not tool['started'] and tool['name']: tool['block_index'] = current_block_index tool['started'] = True block_start = { 'type': 'content_block_start', 'index': current_block_index, 'content_block': { 'type': 'tool_use', 'id': tool['id'], 'name': tool['name'], 'input': {}, }, } yield f'event: content_block_start\ndata: {JSONCodec.dumps(block_start)}\n\n'.encode() current_block_index += 1 if tool['arguments']: block_delta = { 'type': 'content_block_delta', 'index': tool['block_index'], 'delta': { 'type': 'input_json_delta', 'partial_json': tool['arguments'], }, } yield f'event: content_block_delta\ndata: {JSONCodec.dumps(block_delta)}\n\n'.encode() # 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: {JSONCodec.dumps(block_stop)}\n\n'.encode() # Close any tool call blocks that are still open for tool in tracked_tool_calls.values(): if tool['started'] and not tool['stopped']: block_stop = {'type': 'content_block_stop', 'index': tool['block_index']} yield f'event: content_block_stop\ndata: {JSONCodec.dumps(block_stop)}\n\n'.encode() # Emit message_delta with stop reason usage = {'output_tokens': output_tokens} if input_tokens is not None: usage['input_tokens'] = input_tokens if cache_creation_input_tokens is not None: usage['cache_creation_input_tokens'] = cache_creation_input_tokens if cache_read_input_tokens is not None: usage['cache_read_input_tokens'] = cache_read_input_tokens if output_tokens_details is not None: usage['output_tokens_details'] = output_tokens_details if server_tool_use is not None: usage['server_tool_use'] = server_tool_use if service_tier is not None: usage['service_tier'] = service_tier message_delta = { 'type': 'message_delta', 'delta': { 'stop_reason': stop_reason, 'stop_sequence': None, }, 'usage': usage, } yield f'event: message_delta\ndata: {JSONCodec.dumps(message_delta)}\n\n'.encode() # Emit message_stop yield f'event: message_stop\ndata: {JSONCodec.dumps({"type": "message_stop"})}\n\n'.encode()