import copy
import time
import logging
import sys
import os
import base64
import textwrap
import asyncio
from aiocache import cached
from typing import Any, Optional
import random
import json
import html
import inspect
import re
import ast
from uuid import uuid4
from concurrent.futures import ThreadPoolExecutor
from fastapi import Request, HTTPException
from fastapi.responses import HTMLResponse
from starlette.responses import Response, StreamingResponse, JSONResponse
from open_webui.utils.misc import is_string_allowed
from open_webui.models.oauth_sessions import OAuthSessions
from open_webui.models.chats import Chats
from open_webui.models.folders import Folders
from open_webui.models.users import Users
from open_webui.socket.main import (
get_event_call,
get_event_emitter,
)
from open_webui.routers.tasks import (
generate_queries,
generate_title,
generate_follow_ups,
generate_image_prompt,
generate_chat_tags,
)
from open_webui.routers.retrieval import (
process_web_search,
SearchForm,
)
from open_webui.utils.tools import get_builtin_tools
from open_webui.routers.images import (
image_generations,
CreateImageForm,
image_edits,
EditImageForm,
)
from open_webui.routers.pipelines import (
process_pipeline_inlet_filter,
process_pipeline_outlet_filter,
)
from open_webui.routers.memories import query_memory, QueryMemoryForm
from open_webui.utils.webhook import post_webhook
from open_webui.utils.files import (
convert_markdown_base64_images,
get_file_url_from_base64,
get_image_base64_from_url,
get_image_url_from_base64,
)
from open_webui.models.users import UserModel
from open_webui.models.functions import Functions
from open_webui.models.models import Models
from open_webui.retrieval.utils import get_sources_from_items
from open_webui.utils.sanitize import sanitize_code
from open_webui.utils.chat import generate_chat_completion
from open_webui.utils.task import (
get_task_model_id,
rag_template,
tools_function_calling_generation_template,
)
from open_webui.utils.misc import (
deep_update,
extract_urls,
get_message_list,
add_or_update_system_message,
add_or_update_user_message,
set_last_user_message_content,
get_last_user_message,
get_last_user_message_item,
get_last_assistant_message,
get_system_message,
merge_system_messages,
replace_system_message_content,
prepend_to_first_user_message_content,
convert_logit_bias_input_to_json,
get_content_from_message,
convert_output_to_messages,
strip_empty_content_blocks,
)
from open_webui.utils.tools import (
get_tools,
get_updated_tool_function,
get_terminal_tools,
)
from open_webui.utils.access_control import has_connection_access
from open_webui.utils.access_control.files import get_accessible_folder_files
from open_webui.utils.plugin import load_function_module_by_id
from open_webui.utils.filter import (
get_sorted_filter_ids,
process_filter_functions,
)
from open_webui.utils.code_interpreter import execute_code_jupyter
from open_webui.utils.payload import apply_system_prompt_to_body
from open_webui.utils.response import normalize_usage
from open_webui.utils.mcp.client import MCPClient
from open_webui.config import (
CACHE_DIR,
DEFAULT_VOICE_MODE_PROMPT_TEMPLATE,
DEFAULT_TOOLS_FUNCTION_CALLING_PROMPT_TEMPLATE,
DEFAULT_CODE_INTERPRETER_PROMPT,
CODE_INTERPRETER_PYODIDE_PROMPT,
CODE_INTERPRETER_BLOCKED_MODULES,
)
from open_webui.env import (
GLOBAL_LOG_LEVEL,
ENABLE_CHAT_RESPONSE_BASE64_IMAGE_URL_CONVERSION,
CHAT_RESPONSE_STREAM_DELTA_CHUNK_SIZE,
CHAT_RESPONSE_MAX_TOOL_CALL_RETRIES,
BYPASS_MODEL_ACCESS_CONTROL,
ENABLE_REALTIME_CHAT_SAVE,
ENABLE_QUERIES_CACHE,
RAG_SYSTEM_CONTEXT,
ENABLE_FORWARD_USER_INFO_HEADERS,
FORWARD_SESSION_INFO_HEADER_CHAT_ID,
FORWARD_SESSION_INFO_HEADER_MESSAGE_ID,
ENABLE_RESPONSES_API_STATEFUL,
)
from open_webui.utils.headers import include_user_info_headers
from open_webui.constants import TASKS
logging.basicConfig(stream=sys.stdout, level=GLOBAL_LOG_LEVEL)
log = logging.getLogger(__name__)
# We believe in one maker of all models, seen and unseen,
# and in the reasoning which proceeds from the architect.
# We look for the resurrection of dead processes and the
# inference of the world to come.
DEFAULT_REASONING_TAGS = [
('', ''),
('', ''),
('', ''),
('', ''),
('', ''),
('', ''),
('<|begin_of_thought|>', '<|end_of_thought|>'),
('◁think▷', '◁/think▷'),
]
DEFAULT_SOLUTION_TAGS = [('<|begin_of_solution|>', '<|end_of_solution|>')]
DEFAULT_CODE_INTERPRETER_TAGS = [('', '')]
def output_id(prefix: str) -> str:
"""Generate OR-style ID: prefix + 24-char hex UUID."""
return f'{prefix}_{uuid4().hex[:24]}'
def _split_tool_calls(
tool_calls: list[dict],
) -> list[dict]:
"""Expand tool calls whose arguments contain multiple back-to-back JSON objects.
Some models (e.g. GPT-5.4) send multiple complete JSON argument objects
under the same tool call index, producing concatenated invalid JSON like:
'{"query":"A","count":5}{"query":"B","count":5}'
Each such tool call is split into separate entries so each gets executed
independently. Single-object arguments pass through unchanged.
"""
def split_json_objects(raw: str) -> list[str]:
decoder = json.JSONDecoder()
results = []
position = 0
while position < len(raw):
while position < len(raw) and raw[position].isspace():
position += 1
if position >= len(raw):
break
try:
_, end = decoder.raw_decode(raw, position)
results.append(raw[position:end].strip())
position = end
except json.JSONDecodeError:
return [raw]
return results or [raw]
expanded = []
for tool_call in tool_calls:
arguments = tool_call.get('function', {}).get('arguments', '')
split_arguments = split_json_objects(arguments)
if len(split_arguments) <= 1:
expanded.append(tool_call)
else:
for argument in split_arguments:
cloned = copy.deepcopy(tool_call)
cloned['id'] = f'call_{uuid4().hex[:24]}'
cloned['function']['arguments'] = argument
expanded.append(cloned)
return expanded
def get_citation_source_from_tool_result(
tool_name: str, tool_params: dict, tool_result: str, tool_id: str = ''
) -> list[dict]:
"""
Parse a tool's result and convert it to source dicts for citation display.
Follows the source format conventions from get_sources_from_items:
- source: file/item info object with id, name, type
- document: list of document contents
- metadata: list of metadata objects with source, file_id, name fields
Returns a list of sources (usually one, but query_knowledge_files may return multiple).
"""
_EXPECTS_LIST = {'search_web', 'query_knowledge_files'}
_EXPECTS_DICT = {'view_knowledge_file', 'view_file'}
try:
try:
tool_result = json.loads(tool_result)
except (json.JSONDecodeError, TypeError):
pass # keep tool_result as-is (e.g. fetch_url returns plain text)
if isinstance(tool_result, dict) and 'error' in tool_result:
return []
# Validate tool_result type based on what the branch expects
if tool_name in _EXPECTS_LIST and not isinstance(tool_result, list):
return []
elif tool_name in _EXPECTS_DICT and not isinstance(tool_result, dict):
return []
if tool_name == 'search_web':
# Parse JSON array: [{"title": "...", "link": "...", "snippet": "..."}]
results = tool_result
documents = []
metadata = []
for result in results:
title = result.get('title', '')
link = result.get('link', '')
snippet = result.get('snippet', '')
documents.append(f'{title}\n{snippet}')
metadata.append(
{
'source': link,
'name': title,
'url': link,
}
)
return [
{
'source': {'name': 'search_web', 'id': 'search_web'},
'document': documents,
'metadata': metadata,
}
]
elif tool_name in ('view_knowledge_file', 'view_file'):
file_data = tool_result
filename = file_data.get('filename', 'Unknown File')
file_id = file_data.get('id', '')
knowledge_name = file_data.get('knowledge_name', '')
return [
{
'source': {
'id': file_id,
'name': filename,
'type': 'file',
},
'document': [file_data.get('content', '')],
'metadata': [
{
'file_id': file_id,
'name': filename,
'source': filename,
**({'knowledge_name': knowledge_name} if knowledge_name else {}),
}
],
}
]
elif tool_name == 'fetch_url':
url = tool_params.get('url', '')
content = tool_result if isinstance(tool_result, str) else str(tool_result)
snippet = content[:500] + ('...' if len(content) > 500 else '')
return [
{
'source': {'name': url or 'fetch_url', 'id': url or 'fetch_url'},
'document': [snippet],
'metadata': [
{
'source': url,
'name': url,
'url': url,
}
],
}
]
elif tool_name == 'query_knowledge_files':
chunks = tool_result
# Group chunks by source for better citation display
# Each unique source becomes a separate source entry
sources_by_file = {}
for chunk in chunks:
source_name = chunk.get('source', 'Unknown')
file_id = chunk.get('file_id', '')
note_id = chunk.get('note_id', '')
chunk_type = chunk.get('type', 'file')
content = chunk.get('content', '')
# Use file_id or note_id as the key
key = file_id or note_id or source_name
if key not in sources_by_file:
sources_by_file[key] = {
'source': {
'id': file_id or note_id,
'name': source_name,
'type': chunk_type,
},
'document': [],
'metadata': [],
}
sources_by_file[key]['document'].append(content)
sources_by_file[key]['metadata'].append(
{
'file_id': file_id,
'name': source_name,
'source': source_name,
**({'note_id': note_id} if note_id else {}),
}
)
# Return all grouped sources as a list
if sources_by_file:
return list(sources_by_file.values())
# Empty result fallback
return []
else:
# Fallback for other tools
return [
{
'source': {
'name': tool_name,
'type': 'tool',
'id': tool_id or tool_name,
},
'document': [str(tool_result)],
'metadata': [{'source': tool_name, 'name': tool_name}],
}
]
except Exception as e:
log.exception(f'Error parsing tool result for {tool_name}: {e}')
return [
{
'source': {'name': tool_name, 'type': 'tool'},
'document': [str(tool_result)],
'metadata': [{'source': tool_name}],
}
]
def split_content_and_whitespace(content):
content_stripped = content.rstrip()
original_whitespace = content[len(content_stripped) :] if len(content) > len(content_stripped) else ''
return content_stripped, original_whitespace
def is_opening_code_block(content):
backtick_segments = content.split('```')
# Even number of segments means the last backticks are opening a new block
return len(backtick_segments) > 1 and len(backtick_segments) % 2 == 0
_OPENAI_TOOL_DISPLAY_NAMES = {
'web_search_call': 'Web Search',
'file_search_call': 'File Search',
'computer_call': 'Computer Use',
}
def _render_openai_tool_call_handler(item: dict, done: bool) -> str:
"""Render an OpenAI Responses API server-side tool item as a block.
Handles web_search_call, file_search_call, and computer_call items whose
schemas are defined in the openai-python SDK (generated from OpenAPI spec).
"""
item_type = item.get('type', '')
call_id = item.get('id', '')
display_name = _OPENAI_TOOL_DISPLAY_NAMES.get(item_type, item_type)
# Build a short summary of what the tool did
summary = ''
if item_type == 'web_search_call':
action = item.get('action', {})
if isinstance(action, dict):
atype = action.get('type', '')
if atype == 'search':
queries = action.get('queries') or []
query = action.get('query', '')
summary = (
f'Search: {", ".join(str(q) for q in queries)}'
if queries
else (f'Search: {query}' if query else '')
)
elif atype == 'open_page':
summary = f'Open page: {action.get("url", "")}' if action.get('url') else ''
elif atype == 'find_in_page':
summary = f'Find in page: {action.get("pattern", "")}' if action.get('pattern') else ''
elif item_type == 'file_search_call':
queries = item.get('queries', [])
if queries:
summary = f'Queries: {", ".join(str(q) for q in queries)}'
elif item_type == 'computer_call':
action = item.get('action')
actions = item.get('actions')
if isinstance(action, dict):
summary = f'Action: {action.get("type", "unknown")}'
elif isinstance(actions, list) and actions:
summary = f'Actions: {", ".join(a.get("type", "?") for a in actions if isinstance(a, dict))}'
escaped_name = html.escape(display_name)
if done:
return f'\nTool Executed
\n{html.escape(summary)}\n \n'
return f'\nExecuting...
\n \n'
def serialize_output(output: list) -> str:
"""
Convert OR-aligned output items to HTML for display.
For LLM consumption, use convert_output_to_messages() instead.
"""
parts: list[str] = []
# First pass: collect function_call_output items by call_id for lookup
tool_outputs = {}
for item in output:
if item.get('type') == 'function_call_output':
tool_outputs[item.get('call_id')] = item
# Second pass: render items in order
for idx, item in enumerate(output):
item_type = item.get('type', '')
if item_type == 'message':
for content_part in item.get('content', []):
if 'text' in content_part:
text = content_part.get('text', '').strip()
if text:
parts.append(text)
elif item_type == 'function_call':
call_id = item.get('call_id', '')
name = item.get('name', '')
arguments = item.get('arguments', '')
result_item = tool_outputs.get(call_id)
if result_item:
result_parts: list[str] = []
for result_output in result_item.get('output', []):
if 'text' in result_output:
output_text = result_output.get('text', '')
result_parts.append(str(output_text) if not isinstance(output_text, str) else output_text)
result_text = ''.join(result_parts)
files = result_item.get('files')
embeds = result_item.get('embeds', '')
parts.append(
f'\nTool Executed
\n{html.escape(json.dumps(result_text, ensure_ascii=False))}\n '
)
else:
parts.append(
f'\nExecuting...
\n '
)
elif item_type == 'function_call_output':
# Already handled inline with function_call above
pass
elif item_type in _OPENAI_TOOL_DISPLAY_NAMES:
status = item.get('status', 'in_progress')
done = status in ('completed', 'failed', 'incomplete') or idx != len(output) - 1
parts.append(_render_openai_tool_call_handler(item, done).rstrip('\n'))
elif item_type == 'reasoning':
reasoning_parts: list[str] = []
# Check for 'summary' (new structure) or 'content' (legacy/fallback)
source_list = item.get('summary', []) or item.get('content', [])
for content_part in source_list:
if 'text' in content_part:
reasoning_parts.append(content_part.get('text', ''))
elif 'summary' in content_part: # Handle potential nested logic if any
pass
reasoning_content = ''.join(reasoning_parts).strip()
duration = item.get('duration')
status = item.get('status', 'in_progress')
# Infer completion: if this reasoning item is NOT the last item,
# render as done (a subsequent item means reasoning is complete)
is_last_item = idx == len(output) - 1
display = html.escape(
'\n'.join(
(f'> {line}' if not line.startswith('>') else line) for line in reasoning_content.splitlines()
)
)
if status == 'completed' or duration is not None or not is_last_item:
parts.append(
f'\nThought for {duration or 0} seconds
\n{display}\n '
)
else:
parts.append(
f'\nThinking…
\n{display}\n '
)
elif item_type == 'open_webui:code_interpreter':
# Code interpreter needs to inspect/mutate prior accumulated content
# to strip trailing unclosed code fences — materialize only here.
content = '\n'.join(parts)
content_stripped, original_whitespace = split_content_and_whitespace(content)
if is_opening_code_block(content_stripped):
content = content_stripped.rstrip('`').rstrip() + original_whitespace
else:
content = content_stripped + original_whitespace
# Re-split back into parts list after mutation
parts = [content] if content else []
# Render the code_interpreter item as a block
# so the frontend Collapsible renders "Analyzing..."/"Analyzed".
code = item.get('code', '').strip()
lang = item.get('lang', 'python')
status = item.get('status', 'in_progress')
duration = item.get('duration')
is_last_item = idx == len(output) - 1
# Build inner content: code block
display = ''
if code:
display = f'```{lang}\n{code}\n```'
# Build output attribute as HTML-escaped JSON for CodeBlock.svelte
ci_output = item.get('output')
output_attr = ''
if ci_output:
if isinstance(ci_output, dict):
output_json = json.dumps(ci_output, ensure_ascii=False)
else:
output_json = json.dumps({'result': str(ci_output)}, ensure_ascii=False)
output_attr = f' output="{html.escape(output_json)}"'
if status == 'completed' or duration is not None or not is_last_item:
parts.append(
f'\nAnalyzed
\n{display}\n '
)
else:
parts.append(
f'\nAnalyzing…
\n{display}\n '
)
return '\n'.join(parts).strip()
def deep_merge(target, source):
"""
Merge source into target recursively (returning new structure).
- Dicts: Recursive merge.
- Strings: Concatenation.
- Others: Overwrite.
"""
if isinstance(target, dict) and isinstance(source, dict):
new_target = target.copy()
for k, v in source.items():
if k in new_target:
new_target[k] = deep_merge(new_target[k], v)
else:
new_target[k] = v
return new_target
elif isinstance(target, str) and isinstance(source, str):
return target + source
else:
return source
def handle_responses_streaming_event(
data: dict,
current_output: list,
) -> tuple[list, dict | None]:
"""
Handle Responses API streaming events in a pure functional way.
Args:
data: The event data
current_output: List of output items (treated as immutable)
Returns:
tuple[list, dict | None]: (new_output, metadata)
- new_output: The updated output list.
- metadata: Metadata to emit (e.g. usage), {} if update occurred, None if skip.
"""
# Default: no change
# Note: treating current_output as immutable, but avoiding full deepcopy for perf.
# We will shallow copy only if we need to modify the list structure or items.
event_type = data.get('type', '')
if event_type == 'response.output_item.added':
item = data.get('item', {})
if item:
new_output = list(current_output)
new_output.append(item)
return new_output, None
return current_output, None
elif event_type == 'response.content_part.added':
part = data.get('part', {})
output_index = data.get('output_index', len(current_output) - 1)
if current_output and 0 <= output_index < len(current_output):
new_output = list(current_output)
# Copy the item to mutate it
item = new_output[output_index].copy()
new_output[output_index] = item
if 'content' not in item:
item['content'] = []
else:
# Copy content list
item['content'] = list(item['content'])
if item.get('type') == 'reasoning':
# Reasoning items should not have content parts
pass
else:
item['content'].append(part)
return new_output, None
return current_output, None
elif event_type == 'response.reasoning_summary_part.added':
part = data.get('part', {})
output_index = data.get('output_index', len(current_output) - 1)
if current_output and 0 <= output_index < len(current_output):
new_output = list(current_output)
item = new_output[output_index].copy()
new_output[output_index] = item
if 'summary' not in item:
item['summary'] = []
else:
item['summary'] = list(item['summary'])
item['summary'].append(part)
return new_output, None
return current_output, None
elif event_type.startswith('response.') and event_type.endswith('.delta'):
# Generic Delta Handling
parts = event_type.split('.')
if len(parts) >= 3:
delta_type = parts[1]
delta = data.get('delta', '')
output_index = data.get('output_index', len(current_output) - 1)
if current_output and 0 <= output_index < len(current_output):
new_output = list(current_output)
item = new_output[output_index].copy()
new_output[output_index] = item
item_type = item.get('type', '')
# Determine target field and object based on delta_type and item_type
if delta_type == 'function_call_arguments':
key = 'arguments'
if item_type == 'function_call':
# Function call args are usually strings
item[key] = item.get(key, '') + str(delta)
else:
# Generic handling, refined by item type below
pass
if item_type == 'message':
# Message items: "text"/"output_text" -> "text"
# "reasoning_text" -> Skipped (should use reasoning item)
if delta_type in ['text', 'output_text']:
key = 'text'
elif delta_type in ['reasoning_text', 'reasoning_summary_text']:
# Skip reasoning updates for message items
return new_output, None
else:
key = delta_type
content_index = data.get('content_index', 0)
if 'content' not in item:
item['content'] = []
else:
item['content'] = list(item['content'])
content_list = item['content']
while len(content_list) <= content_index:
content_list.append({'type': 'text', 'text': ''})
# Copy the part to mutate it
part = content_list[content_index].copy()
content_list[content_index] = part
current_val = part.get(key)
if current_val is None:
# Initialize based on delta type
current_val = {} if isinstance(delta, dict) else ''
part[key] = deep_merge(current_val, delta)
elif item_type == 'reasoning':
# Reasoning items: "reasoning_text"/"reasoning_summary_text" -> "text"
# "text"/"output_text" -> Skipped (should use message item)
if delta_type == 'reasoning_summary_text':
# Summary updates -> item['summary']
key = 'text'
summary_index = data.get('summary_index', 0)
if 'summary' not in item:
item['summary'] = []
else:
item['summary'] = list(item['summary'])
summary_list = item['summary']
while len(summary_list) <= summary_index:
summary_list.append({'type': 'summary_text', 'text': ''})
part = summary_list[summary_index].copy()
summary_list[summary_index] = part
target_val = part.get(key, '')
part[key] = deep_merge(target_val, delta)
elif delta_type == 'reasoning_text':
# Reasoning body updates -> item['content']
key = 'text'
content_index = data.get('content_index', 0)
if 'content' not in item:
item['content'] = []
else:
item['content'] = list(item['content'])
content_list = item['content']
while len(content_list) <= content_index:
# Reasoning content parts default to text
content_list.append({'type': 'text', 'text': ''})
part = content_list[content_index].copy()
content_list[content_index] = part
target_val = part.get(key, '')
part[key] = deep_merge(target_val, delta)
elif delta_type in ['text', 'output_text']:
return new_output, None
else:
# Fallback just in case other deltas target reasoning?
pass
else:
# Fallback for other item types
if delta_type in ['text', 'output_text']:
key = 'text'
else:
key = delta_type
current_val = item.get(key)
if current_val is None:
current_val = {} if isinstance(delta, dict) else ''
item[key] = deep_merge(current_val, delta)
return new_output, None
elif event_type.startswith('response.') and event_type.endswith('.done'):
# Delta Events: response.content_part.done, response.text.done, etc.
parts = event_type.split('.')
if len(parts) >= 3:
type_name = parts[1]
# 1. Handle specific Delta "done" signals
if type_name == 'content_part':
# "Signaling that no further changes will occur to a content part"
# If payloads contains the full part, we could update it.
# Usually purely signaling in standard implementation, but we check payload.
part = data.get('part')
output_index = data.get('output_index', len(current_output) - 1)
if part and current_output and 0 <= output_index < len(current_output):
new_output = list(current_output)
item = new_output[output_index].copy()
new_output[output_index] = item
if 'content' in item:
item['content'] = list(item['content'])
content_index = data.get('content_index', len(item['content']) - 1)
if 0 <= content_index < len(item['content']):
item['content'][content_index] = part
return new_output, {}
return current_output, None
elif type_name == 'reasoning_summary_part':
part = data.get('part')
output_index = data.get('output_index', len(current_output) - 1)
if part and current_output and 0 <= output_index < len(current_output):
new_output = list(current_output)
item = new_output[output_index].copy()
new_output[output_index] = item
if 'summary' in item:
item['summary'] = list(item['summary'])
summary_index = data.get('summary_index', len(item['summary']) - 1)
if 0 <= summary_index < len(item['summary']):
item['summary'][summary_index] = part
return new_output, {}
return current_output, None
# 2. Skip Output Item done (handled specifically below)
if type_name == 'output_item':
pass
# 3. Generic Field Done (text.done, audio.done)
elif type_name not in ['completed', 'failed']:
output_index = data.get('output_index', len(current_output) - 1)
if current_output and 0 <= output_index < len(current_output):
key = (
'text'
if type_name
in [
'text',
'output_text',
'reasoning_text',
'reasoning_summary_text',
]
else type_name
)
if type_name == 'function_call_arguments':
key = 'arguments'
if key in data:
final_value = data[key]
new_output = list(current_output)
item = new_output[output_index].copy()
new_output[output_index] = item
item_type = item.get('type', '')
if type_name == 'function_call_arguments':
if item_type == 'function_call':
item['arguments'] = final_value
elif item_type == 'message':
content_index = data.get('content_index', 0)
if 'content' in item:
item['content'] = list(item['content'])
if len(item['content']) > content_index:
part = item['content'][content_index].copy()
item['content'][content_index] = part
part[key] = final_value
elif item_type == 'reasoning':
item['status'] = 'completed'
else:
item[key] = final_value
return new_output, {}
return current_output, None
elif event_type == 'response.output_item.done':
# Delta Event: Output item complete
item = data.get('item')
output_index = data.get('output_index', len(current_output) - 1)
new_output = list(current_output)
if item and 0 <= output_index < len(current_output):
new_output[output_index] = item
elif item:
new_output.append(item)
return new_output, {}
elif event_type == 'response.completed':
# State Machine Event: Completed
response_data = data.get('response', {})
final_output = response_data.get('output')
new_output = final_output if final_output is not None else current_output
# Ensure reasoning items are marked as completed in the final output
if new_output:
for item in new_output:
if item.get('type') == 'reasoning' and item.get('status') != 'completed':
item['status'] = 'completed'
return new_output, {
'usage': response_data.get('usage'),
'done': True,
'response_id': response_data.get('id'),
}
elif event_type == 'response.in_progress':
# State Machine Event: In Progress
# We could extract metadata if needed, but for now just acknowledge iteration
return current_output, None
elif event_type == 'response.failed':
# State Machine Event: Failed
error = data.get('response', {}).get('error', {})
return current_output, {'error': error}
else:
return current_output, None
def get_source_context(sources: list, source_ids: dict = None, include_content: bool = True) -> str:
"""
Build tag context string from citation sources.
"""
context_string = ''
if source_ids is None:
source_ids = {}
for source in sources:
for doc, meta in zip(source.get('document', []), source.get('metadata', [])):
source_id = meta.get('source') or source.get('source', {}).get('id') or 'N/A'
if source_id not in source_ids:
source_ids[source_id] = len(source_ids) + 1
src_name = source.get('source', {}).get('name')
src_type = source.get('source', {}).get('type')
src_rid = source.get('source', {}).get('id')
body = doc if include_content else ''
context_string += (
f'{body}\n'
)
return context_string
async def apply_source_context_to_messages(
request: Request,
messages: list,
sources: list,
user_message: str,
include_content: bool = True,
) -> list:
"""
Build source context from citation sources and apply to messages.
Uses RAG template to format context for model consumption.
When include_content is False, emit tags with id/name but no
document body — useful when the content is already present elsewhere
(e.g. in a tool result message) and only citation markers are needed.
"""
if not sources or not user_message:
return messages
context = get_source_context(sources, include_content=include_content)
context = context.strip()
if not context:
return messages
if RAG_SYSTEM_CONTEXT:
return add_or_update_system_message(
await rag_template(request.app.state.config.RAG_TEMPLATE, context, user_message),
messages,
append=True,
)
else:
return add_or_update_user_message(
await rag_template(request.app.state.config.RAG_TEMPLATE, context, user_message),
messages,
append=False,
)
async def process_tool_result(
request,
tool_function_name,
tool_result,
tool_type,
direct_tool=False,
metadata=None,
user=None,
):
tool_result_embeds = []
EXTERNAL_TOOL_TYPES = ('external', 'action', 'terminal')
# Support (HTMLResponse, result_context) tuples: the optional second
# element lets tool authors provide the LLM with actionable context
# about the generated embed instead of the generic fallback message.
result_context = None
if isinstance(tool_result, tuple) and len(tool_result) == 2 and isinstance(tool_result[0], HTMLResponse):
tool_result, result_context = tool_result
if isinstance(tool_result, HTMLResponse):
content_disposition = tool_result.headers.get('Content-Disposition', '')
if 'inline' in content_disposition:
content = tool_result.body.decode('utf-8', 'replace')
tool_result_embeds.append(content)
if 200 <= tool_result.status_code < 300:
if result_context is not None and isinstance(result_context, (str, dict, list)):
tool_result = result_context
else:
tool_result = {
'status': 'success',
'code': 'ui_component',
'message': f'{tool_function_name}: Embedded UI result is active and visible to the user.',
}
elif 400 <= tool_result.status_code < 500:
tool_result = {
'status': 'error',
'code': 'ui_component',
'message': f'{tool_function_name}: Client error {tool_result.status_code} from embedded UI result.',
}
elif 500 <= tool_result.status_code < 600:
tool_result = {
'status': 'error',
'code': 'ui_component',
'message': f'{tool_function_name}: Server error {tool_result.status_code} from embedded UI result.',
}
else:
tool_result = {
'status': 'error',
'code': 'ui_component',
'message': f'{tool_function_name}: Unexpected status code {tool_result.status_code} from embedded UI result.',
}
else:
tool_result = tool_result.body.decode('utf-8', 'replace')
elif (tool_type in EXTERNAL_TOOL_TYPES and isinstance(tool_result, tuple)) or (
direct_tool and isinstance(tool_result, list) and len(tool_result) == 2
):
tool_result, tool_response_headers = tool_result
try:
if not isinstance(tool_response_headers, dict):
tool_response_headers = dict(tool_response_headers)
except Exception as e:
tool_response_headers = {}
log.debug(e)
if tool_response_headers and isinstance(tool_response_headers, dict):
content_disposition = tool_response_headers.get(
'Content-Disposition',
tool_response_headers.get('content-disposition', ''),
)
if 'inline' in content_disposition:
content_type = tool_response_headers.get(
'Content-Type',
tool_response_headers.get('content-type', ''),
)
location = tool_response_headers.get(
'Location',
tool_response_headers.get('location', ''),
)
if 'text/html' in content_type:
# Support (html_content, result_context) nested tuple
result_context = None
html_content = tool_result
if isinstance(tool_result, (tuple, list)) and len(tool_result) == 2:
html_content, result_context = tool_result
# Display as iframe embed
tool_result_embeds.append(html_content)
if result_context is not None and isinstance(result_context, (str, dict, list)):
tool_result = result_context
else:
tool_result = {
'status': 'success',
'code': 'ui_component',
'message': f'{tool_function_name}: Embedded UI result is active and visible to the user.',
}
elif location:
# Support (html_content, result_context) nested tuple for location embeds
result_context = None
if isinstance(tool_result, (tuple, list)) and len(tool_result) == 2:
_, result_context = tool_result
tool_result_embeds.append(location)
if result_context is not None and isinstance(result_context, (str, dict, list)):
tool_result = result_context
else:
tool_result = {
'status': 'success',
'code': 'ui_component',
'message': f'{tool_function_name}: Embedded UI result is active and visible to the user.',
}
tool_result_files = []
# Detect base64 image data URIs from tool results (e.g. binary image
# responses from execute_tool_server). Move the data URI to
# tool_result_files and replace tool_result with a text summary.
if isinstance(tool_result, str) and tool_result.startswith('data:image/'):
tool_result_files.append({'type': 'image', 'url': tool_result})
tool_result = f'{tool_function_name}: Image file read successfully.'
if isinstance(tool_result, list):
if tool_type == 'mcp': # MCP
tool_response = []
for item in tool_result:
if isinstance(item, dict):
if item.get('type') == 'text':
text = item.get('text', '')
if isinstance(text, str):
try:
text = json.loads(text)
except json.JSONDecodeError:
pass
tool_response.append(text)
elif item.get('type') in ['image', 'audio']:
file_url = await get_file_url_from_base64(
request,
f'data:{item.get("mimeType")};base64,{item.get("data", item.get("blob", ""))}',
{
'chat_id': metadata.get('chat_id', None),
'message_id': metadata.get('message_id', None),
'session_id': metadata.get('session_id', None),
'result': item,
},
user,
)
tool_result_files.append(
{
'type': item.get('type', 'data'),
'url': file_url,
}
)
elif item.get('type') == 'resource':
resource = item.get('resource', {})
text = resource.get('text', '')
if isinstance(text, str) and text:
try:
text = json.loads(text)
except json.JSONDecodeError:
pass
tool_response.append(text)
tool_result = tool_response[0] if len(tool_response) == 1 else tool_response
else: # OpenAPI
for item in tool_result:
if isinstance(item, str) and item.startswith('data:'):
tool_result_files.append(
{
'type': 'data',
'content': item,
}
)
tool_result.remove(item)
if isinstance(tool_result, list):
tool_result = {'results': tool_result}
if isinstance(tool_result, dict) or isinstance(tool_result, list):
tool_result = json.dumps(tool_result, indent=2, ensure_ascii=False)
# Safety: ensure tool_result is always a string (or None) to prevent
# downstream TypeError when concatenating (e.g. if an upstream callable
# returned a tuple that was not unpacked by the branches above).
if tool_result is not None and not isinstance(tool_result, str):
if isinstance(tool_result, tuple):
# execute_tool_server returns (data, headers); unpack the data part
tool_result = json.dumps(tool_result[0], indent=2, ensure_ascii=False) if len(tool_result) > 0 else ''
else:
tool_result = str(tool_result)
return tool_result, tool_result_files, tool_result_embeds
async def terminal_event_handler(
tool_function_name: str,
tool_function_params: dict,
tool_result,
event_emitter,
):
"""Emit terminal:* events for Open Terminal tools.
- display_file → emits 'terminal:display_file' to open the file preview.
- write_file / replace_file_content → emits 'terminal:write_file' to refresh.
- run_command → emits 'terminal:run_command' with cwd to refresh if relevant.
"""
if not event_emitter:
return
if tool_function_name == 'display_file':
path = tool_function_params.get('path', '')
if not path:
return
# Only emit if the file actually exists
parsed = tool_result
if isinstance(parsed, str):
try:
parsed = json.loads(parsed)
except (json.JSONDecodeError, TypeError):
pass
if isinstance(parsed, dict) and parsed.get('exists') is False:
return
await event_emitter(
{
'type': f'terminal:{tool_function_name}',
'data': {'path': path},
}
)
elif tool_function_name in ('write_file', 'replace_file_content'):
path = tool_function_params.get('path', '')
if not path:
return
await event_emitter(
{
'type': f'terminal:{tool_function_name}',
'data': {'path': path},
}
)
elif tool_function_name == 'run_command':
await event_emitter(
{
'type': 'terminal:run_command',
'data': {},
}
)
async def chat_completion_tools_handler(
request: Request, body: dict, extra_params: dict, user: UserModel, models, tools
) -> tuple[dict, dict]:
async def get_content_from_response(response) -> Optional[str]:
content = None
if hasattr(response, 'body_iterator'):
async for chunk in response.body_iterator:
data = json.loads(chunk.decode('utf-8', 'replace'))
content = data['choices'][0]['message']['content']
# Cleanup any remaining background tasks if necessary
if response.background is not None:
await response.background()
else:
content = response['choices'][0]['message']['content']
return content
def get_tools_function_calling_payload(messages, task_model_id, content):
user_message = get_last_user_message(messages)
if user_message and messages and messages[-1]['role'] == 'user':
# Remove the last user message to avoid duplication
messages = messages[:-1]
recent_messages = messages[-4:] if len(messages) > 4 else messages
chat_history = '\n'.join(
f'{message["role"].upper()}: """{get_content_from_message(message)}"""' for message in recent_messages
)
prompt = f'History:\n{chat_history}\nQuery: {user_message}' if chat_history else f'Query: {user_message}'
return {
'model': task_model_id,
'messages': [
{'role': 'system', 'content': content},
{'role': 'user', 'content': prompt},
],
'stream': False,
'metadata': {'task': str(TASKS.FUNCTION_CALLING)},
}
event_caller = extra_params['__event_call__']
event_emitter = extra_params['__event_emitter__']
metadata = extra_params['__metadata__']
task_model_id = get_task_model_id(
body['model'],
request.app.state.config.TASK_MODEL,
request.app.state.config.TASK_MODEL_EXTERNAL,
models,
)
skip_files = False
sources = []
specs = [tool['spec'] for tool in tools.values()]
tools_specs = json.dumps(specs, ensure_ascii=False)
if request.app.state.config.TOOLS_FUNCTION_CALLING_PROMPT_TEMPLATE != '':
template = request.app.state.config.TOOLS_FUNCTION_CALLING_PROMPT_TEMPLATE
else:
template = DEFAULT_TOOLS_FUNCTION_CALLING_PROMPT_TEMPLATE
tools_function_calling_prompt = tools_function_calling_generation_template(template, tools_specs)
payload = get_tools_function_calling_payload(body['messages'], task_model_id, tools_function_calling_prompt)
try:
response = await generate_chat_completion(request, form_data=payload, user=user)
log.debug(f'{response=}')
content = await get_content_from_response(response)
log.debug(f'{content=}')
if not content:
return body, {}
try:
content = content[content.find('{') : content.rfind('}') + 1]
if not content:
raise Exception('No JSON object found in the response')
result = json.loads(content)
async def tool_call_handler(tool_call):
nonlocal skip_files
log.debug(f'{tool_call=}')
tool_function_name = tool_call.get('name', None)
if tool_function_name not in tools:
return body, {}
tool_function_params = tool_call.get('parameters', {})
tool = None
tool_type = ''
direct_tool = False
try:
tool = tools[tool_function_name]
tool_type = tool.get('type', '')
direct_tool = tool.get('direct', False)
spec = tool.get('spec', {})
allowed_params = spec.get('parameters', {}).get('properties', {}).keys()
tool_function_params = {k: v for k, v in tool_function_params.items() if k in allowed_params}
if tool.get('direct', False):
tool_result = await event_caller(
{
'type': 'execute:tool',
'data': {
'id': str(uuid4()),
'name': tool_function_name,
'params': tool_function_params,
'server': tool.get('server', {}),
'session_id': metadata.get('session_id', None),
},
}
)
else:
tool_function = tool['callable']
tool_result = await tool_function(**tool_function_params)
except Exception as e:
tool_result = str(e)
tool_result, tool_result_files, tool_result_embeds = await process_tool_result(
request,
tool_function_name,
tool_result,
tool_type,
direct_tool,
metadata,
user,
)
if event_emitter:
await terminal_event_handler(
tool_function_name,
tool_function_params,
tool_result,
event_emitter,
)
if tool_result_files:
await event_emitter(
{
'type': 'files',
'data': {
'files': tool_result_files,
},
}
)
if tool_result_embeds:
await event_emitter(
{
'type': 'embeds',
'data': {
'embeds': tool_result_embeds,
},
}
)
if tool_result:
tool = tools[tool_function_name]
tool_id = tool.get('tool_id', '')
tool_name = f'{tool_id}/{tool_function_name}' if tool_id else f'{tool_function_name}'
# Citation is enabled for this tool
sources.append(
{
'source': {
'name': (f'{tool_name}'),
},
'document': [str(tool_result)],
'metadata': [
{
'source': (f'{tool_name}'),
'parameters': tool_function_params,
}
],
'tool_result': True,
}
)
if tools[tool_function_name].get('metadata', {}).get('file_handler', False):
skip_files = True
# check if "tool_calls" in result
if result.get('tool_calls'):
for tool_call in result.get('tool_calls'):
await tool_call_handler(tool_call)
else:
await tool_call_handler(result)
except Exception as e:
log.debug(f'Error: {e}')
content = None
except Exception as e:
log.debug(f'Error: {e}')
content = None
log.debug(f'tool_contexts: {sources}')
if skip_files and 'files' in body.get('metadata', {}):
del body['metadata']['files']
return body, {'sources': sources}
async def chat_memory_handler(request: Request, form_data: dict, extra_params: dict, user):
try:
results = await query_memory(
request,
QueryMemoryForm(
**{
'content': get_last_user_message(form_data['messages']) or '',
'k': 3,
}
),
user,
)
except Exception as e:
log.debug(e)
results = None
user_context = ''
if results and hasattr(results, 'documents'):
if results.documents and len(results.documents) > 0:
for doc_idx, doc in enumerate(results.documents[0]):
created_at_date = 'Unknown Date'
if results.metadatas[0][doc_idx].get('created_at'):
created_at_timestamp = results.metadatas[0][doc_idx]['created_at']
created_at_date = time.strftime('%Y-%m-%d', time.localtime(created_at_timestamp))
user_context += f'{doc_idx + 1}. [{created_at_date}] {doc}\n'
form_data['messages'] = add_or_update_system_message(
f'User Context:\n{user_context}\n', form_data['messages'], append=True
)
return form_data
async def chat_web_search_handler(request: Request, form_data: dict, extra_params: dict, user):
event_emitter = extra_params['__event_emitter__']
await event_emitter(
{
'type': 'status',
'data': {
'action': 'web_search',
'description': 'Searching the web',
'done': False,
},
}
)
messages = form_data['messages']
user_message = get_last_user_message(messages)
queries = []
try:
res = await generate_queries(
request,
{
'model': form_data['model'],
'messages': messages,
'prompt': user_message,
'type': 'web_search',
'chat_id': extra_params.get('__chat_id__'),
},
user,
)
response = res['choices'][0]['message']['content']
try:
bracket_start = response.rfind('{')
bracket_end = response.rfind('}') + 1
if bracket_start == -1 or bracket_end == -1:
raise Exception('No JSON object found in the response')
response = response[bracket_start:bracket_end]
queries = json.loads(response)
queries = queries.get('queries', [])
except Exception as e:
queries = [response]
if ENABLE_QUERIES_CACHE:
request.state.cached_queries = queries
except Exception as e:
log.exception(e)
queries = [user_message or '']
# Check if generated queries are empty
if len(queries) == 1 and queries[0].strip() == '':
queries = [user_message or '']
# Check if queries are not found
if len(queries) == 0:
await event_emitter(
{
'type': 'status',
'data': {
'action': 'web_search',
'description': 'No search query generated',
'done': True,
},
}
)
return form_data
await event_emitter(
{
'type': 'status',
'data': {
'action': 'web_search_queries_generated',
'queries': queries,
'done': False,
},
}
)
try:
results = await process_web_search(
request,
SearchForm(queries=queries),
user=user,
)
if results:
files = form_data.get('files', [])
if results.get('collection_names'):
for col_idx, collection_name in enumerate(results.get('collection_names')):
files.append(
{
'collection_name': collection_name,
'name': ', '.join(queries),
'type': 'web_search',
'urls': results['filenames'],
'queries': queries,
}
)
elif results.get('docs'):
# Invoked when bypass embedding and retrieval is set to True
docs = results['docs']
files.append(
{
'docs': docs,
'name': ', '.join(queries),
'type': 'web_search',
'urls': results['filenames'],
'queries': queries,
}
)
form_data['files'] = files
await event_emitter(
{
'type': 'status',
'data': {
'action': 'web_search',
'description': 'Searched {{count}} sites',
'urls': results['filenames'],
'items': results.get('items', []),
'done': True,
},
}
)
else:
await event_emitter(
{
'type': 'status',
'data': {
'action': 'web_search',
'description': 'No search results found',
'done': True,
'error': True,
},
}
)
except Exception as e:
log.exception(e)
await event_emitter(
{
'type': 'status',
'data': {
'action': 'web_search',
'description': 'An error occurred while searching the web',
'queries': queries,
'done': True,
'error': True,
},
}
)
return form_data
def get_images_from_messages(message_list):
images = []
for message in reversed(message_list):
message_images = []
for file in message.get('files', []):
if file.get('type') == 'image':
message_images.append(file.get('url'))
elif file.get('content_type', '').startswith('image/'):
message_images.append(file.get('url'))
if message_images:
images.append(message_images)
return images
async def get_image_urls(delta_images, request, metadata, user) -> list[str]:
if not isinstance(delta_images, list):
return []
image_urls = []
for img in delta_images:
if not isinstance(img, dict) or img.get('type') != 'image_url':
continue
url = img.get('image_url', {}).get('url')
if not url:
continue
if url.startswith('data:image/png;base64'):
url = await get_image_url_from_base64(request, url, metadata, user)
image_urls.append(url)
return image_urls
async def add_file_context(messages: list, chat_id: str, user) -> list:
"""
Add file URLs to messages for native function calling.
"""
if not chat_id or chat_id.startswith('local:') or chat_id.startswith('channel:'):
return messages
chat = await Chats.get_chat_by_id_and_user_id(chat_id, user.id)
if not chat:
return messages
history = chat.chat.get('history', {})
stored_messages = get_message_list(history.get('messages', {}), history.get('currentId'))
def format_file_tag(file):
attrs = f'type="{file.get("type", "file")}" url="{file["url"]}"'
if file.get('content_type'):
attrs += f' content_type="{file["content_type"]}"'
if file.get('name'):
attrs += f' name="{file["name"]}"'
return f''
# Pair only user-role messages from both lists to avoid misalignment.
# After process_messages_with_output(), assistant messages with tool calls
# are expanded into multiple messages (assistant + tool results), making
# the payload message list longer than the stored message list. A naive
# positional zip() would pair user messages with wrong stored messages,
# causing later images to lose their file context (see #21878).
user_messages = [m for m in messages if m.get('role') == 'user']
stored_user_messages = [m for m in stored_messages if m.get('role') == 'user']
for message, stored_message in zip(user_messages, stored_user_messages):
files_with_urls = [
file
for file in stored_message.get('files', [])
if file.get('url') and not file.get('url').startswith('data:')
]
if not files_with_urls:
continue
file_tags = [format_file_tag(file) for file in files_with_urls]
file_context = '\n' + '\n'.join(file_tags) + '\n\n\n'
content = message.get('content', '')
if isinstance(content, list):
message['content'] = [{'type': 'text', 'text': file_context}] + content
else:
message['content'] = file_context + content
return messages
async def chat_image_generation_handler(request: Request, form_data: dict, extra_params: dict, user):
metadata = extra_params.get('__metadata__', {})
chat_id = metadata.get('chat_id', None)
__event_emitter__ = extra_params.get('__event_emitter__', None)
if not chat_id or not isinstance(chat_id, str) or not __event_emitter__:
return form_data
if chat_id.startswith('local:') or chat_id.startswith('channel:'):
message_list = form_data.get('messages', [])
else:
chat = await Chats.get_chat_by_id_and_user_id(chat_id, user.id)
await __event_emitter__(
{
'type': 'status',
'data': {'description': 'Creating image', 'done': False},
}
)
messages_map = chat.chat.get('history', {}).get('messages', {})
message_id = chat.chat.get('history', {}).get('currentId')
message_list = get_message_list(messages_map, message_id)
user_message = get_last_user_message(message_list)
prompt = user_message
message_images = get_images_from_messages(message_list)
# Limit to first 2 sets of images
# We may want to change this in the future to allow more images
input_images = []
for idx, images in enumerate(message_images):
if idx >= 2:
break
for image in images:
input_images.append(image)
system_message_content = ''
if len(input_images) > 0 and request.app.state.config.ENABLE_IMAGE_EDIT:
# Edit image(s)
try:
images = await image_edits(
request=request,
form_data=EditImageForm(**{'prompt': prompt, 'image': input_images}),
metadata={
'chat_id': metadata.get('chat_id', None),
'message_id': metadata.get('message_id', None),
},
user=user,
)
await __event_emitter__(
{
'type': 'status',
'data': {'description': 'Image created', 'done': True},
}
)
await __event_emitter__(
{
'type': 'files',
'data': {
'files': [
{
'type': 'image',
'url': image['url'],
}
for image in images
]
},
}
)
system_message_content = 'The requested image has been edited and created and is now being shown to the user. Let them know that it has been generated.'
except Exception as e:
log.debug(e)
error_message = ''
if isinstance(e, HTTPException):
if e.detail and isinstance(e.detail, dict):
error_message = e.detail.get('message', str(e.detail))
else:
error_message = str(e.detail)
await __event_emitter__(
{
'type': 'status',
'data': {
'description': f'An error occurred while generating an image',
'done': True,
},
}
)
system_message_content = f'Image generation was attempted but failed. The system is currently unable to generate the image. Tell the user that the following error occurred: {error_message}'
else:
# Create image(s)
if request.app.state.config.ENABLE_IMAGE_PROMPT_GENERATION:
try:
res = await generate_image_prompt(
request,
{
'model': form_data['model'],
'messages': form_data['messages'],
'chat_id': metadata.get('chat_id'),
},
user,
)
response = res['choices'][0]['message']['content']
try:
bracket_start = response.rfind('{')
bracket_end = response.rfind('}') + 1
if bracket_start == -1 or bracket_end == -1:
raise Exception('No JSON object found in the response')
response = response[bracket_start:bracket_end]
response = json.loads(response)
prompt = response.get('prompt', [])
except Exception as e:
prompt = user_message
except Exception as e:
log.exception(e)
prompt = user_message
try:
images = await image_generations(
request=request,
form_data=CreateImageForm(**{'prompt': prompt}),
metadata={
'chat_id': metadata.get('chat_id', None),
'message_id': metadata.get('message_id', None),
},
user=user,
)
await __event_emitter__(
{
'type': 'status',
'data': {'description': 'Image created', 'done': True},
}
)
await __event_emitter__(
{
'type': 'files',
'data': {
'files': [
{
'type': 'image',
'url': image['url'],
}
for image in images
]
},
}
)
system_message_content = 'The requested image has been created by the system successfully and is now being shown to the user. Let the user know that the image they requested has been generated and is now shown in the chat.'
except Exception as e:
log.debug(e)
error_message = ''
if isinstance(e, HTTPException):
if e.detail and isinstance(e.detail, dict):
error_message = e.detail.get('message', str(e.detail))
else:
error_message = str(e.detail)
await __event_emitter__(
{
'type': 'status',
'data': {
'description': f'An error occurred while generating an image',
'done': True,
},
}
)
system_message_content = f'Image generation was attempted but failed because of an error. The system is currently unable to generate the image. Tell the user that the following error occurred: {error_message}'
if system_message_content:
form_data['messages'] = add_or_update_system_message(system_message_content, form_data['messages'])
return form_data
async def chat_completion_files_handler(
request: Request, body: dict, extra_params: dict, user: UserModel
) -> tuple[dict, dict[str, list]]:
__event_emitter__ = extra_params['__event_emitter__']
sources = []
if files := body.get('metadata', {}).get('files', None):
# Check if all files are in full context mode
all_full_context = all(item.get('context') == 'full' for item in files)
queries = []
if not all_full_context:
try:
queries_response = await generate_queries(
request,
{
'model': body['model'],
'messages': body['messages'],
'type': 'retrieval',
'chat_id': body.get('metadata', {}).get('chat_id'),
},
user,
)
queries_response = queries_response['choices'][0]['message']['content']
try:
bracket_start = queries_response.rfind('{')
bracket_end = queries_response.rfind('}') + 1
if bracket_start == -1 or bracket_end == -1:
raise Exception('No JSON object found in the response')
queries_response = queries_response[bracket_start:bracket_end]
queries_response = json.loads(queries_response)
except Exception as e:
queries_response = {'queries': [queries_response]}
queries = queries_response.get('queries', [])
except Exception:
pass
await __event_emitter__(
{
'type': 'status',
'data': {
'action': 'queries_generated',
'queries': queries,
'done': False,
},
}
)
if len(queries) == 0:
queries = [get_last_user_message(body['messages']) or '']
try:
# Directly await async get_sources_from_items (no thread needed - fully async now)
sources = await get_sources_from_items(
request=request,
items=files,
queries=queries,
embedding_function=lambda query, prefix: request.app.state.EMBEDDING_FUNCTION(
query, prefix=prefix, user=user
),
k=request.app.state.config.TOP_K,
reranking_function=(
(lambda query, documents: request.app.state.RERANKING_FUNCTION(query, documents, user=user))
if request.app.state.RERANKING_FUNCTION
else None
),
k_reranker=request.app.state.config.TOP_K_RERANKER,
r=request.app.state.config.RELEVANCE_THRESHOLD,
hybrid_bm25_weight=request.app.state.config.HYBRID_BM25_WEIGHT,
hybrid_search=request.app.state.config.ENABLE_RAG_HYBRID_SEARCH,
full_context=all_full_context or request.app.state.config.RAG_FULL_CONTEXT,
user=user,
)
except Exception as e:
log.exception(e)
log.debug(f'rag_contexts:sources: {sources}')
unique_ids = set()
for source in sources or []:
if not source or len(source.keys()) == 0:
continue
documents = source.get('document') or []
metadatas = source.get('metadata') or []
src_info = source.get('source') or {}
for index, _ in enumerate(documents):
metadata = metadatas[index] if index < len(metadatas) else None
_id = (metadata or {}).get('source') or (src_info or {}).get('id') or 'N/A'
unique_ids.add(_id)
sources_count = len(unique_ids)
await __event_emitter__(
{
'type': 'status',
'data': {
'action': 'sources_retrieved',
'count': sources_count,
'done': True,
},
}
)
return body, {'sources': sources}
def apply_params_to_form_data(form_data, model):
params = form_data.pop('params', {})
custom_params = params.pop('custom_params', {})
open_webui_params = {
'stream_response': bool,
'stream_delta_chunk_size': int,
'function_calling': str,
'reasoning_tags': list,
'system': str,
}
for key in list(params.keys()):
if key in open_webui_params:
del params[key]
if custom_params:
# Attempt to parse custom_params if they are strings
for key, value in custom_params.items():
if isinstance(value, str):
try:
# Attempt to parse the string as JSON
custom_params[key] = json.loads(value)
except json.JSONDecodeError:
# If it fails, keep the original string
pass
# If custom_params are provided, merge them into params
params = deep_update(params, custom_params)
if model.get('owned_by') == 'ollama':
# Ollama specific parameters
form_data['options'] = params
else:
if isinstance(params, dict):
for key, value in params.items():
if value is not None:
form_data[key] = value
if 'logit_bias' in params and params['logit_bias'] is not None:
try:
logit_bias = convert_logit_bias_input_to_json(params['logit_bias'])
if logit_bias:
form_data['logit_bias'] = json.loads(logit_bias)
except Exception as e:
log.exception(f'Error parsing logit_bias: {e}')
return form_data
async def convert_url_images_to_base64(form_data):
messages = form_data.get('messages', [])
for message in messages:
content = message.get('content')
if not isinstance(content, list):
continue
new_content = []
for item in content:
if not isinstance(item, dict) or item.get('type') != 'image_url':
new_content.append(item)
continue
image_url = item.get('image_url', {}).get('url', '')
if image_url.startswith('data:image/'):
new_content.append(item)
continue
try:
base64_data = await get_image_base64_from_url(image_url)
if base64_data:
new_content.append(
{
'type': 'image_url',
'image_url': {'url': base64_data},
}
)
else:
new_content.append(item)
except Exception as e:
log.debug(f'Error converting image URL to base64: {e}')
new_content.append(item)
message['content'] = new_content
return form_data
async def load_messages_from_db(chat_id: str, message_id: str) -> Optional[list[dict]]:
"""
Load the message chain from DB up to message_id,
keeping only LLM-relevant fields (role, content, output).
"""
messages_map = await Chats.get_messages_map_by_chat_id(chat_id)
if not messages_map:
return None
db_messages = get_message_list(messages_map, message_id)
if not db_messages:
return None
return [{k: v for k, v in msg.items() if k in ('role', 'content', 'output', 'files')} for msg in db_messages]
def get_reasoning_format(model: dict) -> str | None:
"""
Determine how reasoning should be included in reconstructed messages.
Returns:
'think_tags': Ollama expects tags in content.
'reasoning_content': llama.cpp supports reasoning_content as a top-level field.
None: skip reasoning (safe default for strict providers).
"""
provider = model.get('provider', '')
if provider == 'ollama':
return 'think_tags'
if provider == 'llama.cpp':
return 'reasoning_content'
return None
def process_messages_with_output(
messages: list[dict],
reasoning_format: str | None = None,
) -> list[dict]:
"""
Process messages with OR-aligned output items for LLM consumption.
For assistant messages with 'output' field, produces properly formatted
OpenAI-style messages (tool_calls + tool results). Strips 'output' before LLM.
"""
processed = []
for message in messages:
if message.get('role') == 'assistant' and message.get('output'):
# Use output items for clean OpenAI-format messages
output_messages = convert_output_to_messages(
message['output'],
raw=True,
reasoning_format=reasoning_format,
)
if output_messages:
processed.extend(output_messages)
continue
# Strip 'output' field before adding (LLM shouldn't see it)
clean_message = {k: v for k, v in message.items() if k != 'output'}
processed.append(clean_message)
return processed
SKILL_MENTION_RE = re.compile(r'<\$([^|>]+)\|?[^>]*>')
def _get_text_parts(message: dict) -> list[str]:
"""Return all text segments from a message's content."""
content = message.get('content')
if isinstance(content, str):
return [content]
if isinstance(content, list):
return [p.get('text', '') for p in content if isinstance(p, dict) and p.get('type') == 'text']
return []
def extract_skill_ids_from_messages(messages: list[dict]) -> set[str]:
"""Extract skill IDs from <$skillId|label> mention tags in messages."""
ids: set[str] = set()
for message in messages:
for text in _get_text_parts(message):
ids.update(m.group(1) for m in SKILL_MENTION_RE.finditer(text))
return ids
def strip_skill_mentions(messages: list[dict]) -> None:
"""Strip <$skillId|label> mention tags from message content in-place."""
strip_re = re.compile(r'<\$[^>]+>')
for message in messages:
content = message.get('content')
if isinstance(content, str) and strip_re.search(content):
message['content'] = strip_re.sub('', content).strip()
elif isinstance(content, list):
for part in content:
if isinstance(part, dict) and part.get('type') == 'text':
text = part.get('text', '')
if strip_re.search(text):
part['text'] = strip_re.sub('', text).strip()
async def process_chat_payload(request, form_data, user, metadata, model):
# Pipeline Inlet -> Filter Inlet -> Chat Memory -> Chat Web Search -> Chat Image Generation
# -> Chat Code Interpreter (Form Data Update) -> (Default) Chat Tools Function Calling
# -> Chat Files
# Arena model resolution — pick the sub-model now so all downstream
# processing (knowledge, capabilities, tools, params) uses its settings
# instead of the empty arena wrapper.
if model.get('owned_by') == 'arena':
arena_model_ids = model.get('info', {}).get('meta', {}).get('model_ids')
arena_filter_mode = model.get('info', {}).get('meta', {}).get('filter_mode')
if arena_model_ids and arena_filter_mode == 'exclude':
arena_model_ids = [
available_model['id']
for available_model in request.app.state.MODELS.values()
if available_model.get('owned_by') != 'arena' and available_model['id'] not in arena_model_ids
]
if isinstance(arena_model_ids, list) and arena_model_ids:
selected_model_id = random.choice(arena_model_ids)
else:
arena_model_ids = [
available_model['id']
for available_model in request.app.state.MODELS.values()
if available_model.get('owned_by') != 'arena'
]
selected_model_id = random.choice(arena_model_ids)
selected_model = request.app.state.MODELS.get(selected_model_id)
if selected_model:
model = selected_model
form_data['model'] = selected_model_id
metadata['selected_model_id'] = selected_model_id
form_data = apply_params_to_form_data(form_data, model)
log.debug(f'form_data: {form_data}')
# Guided regeneration: extract before it reaches the LLM provider
regeneration_prompt = form_data.pop('regeneration_prompt', None)
# Load messages from DB when available — DB preserves structured 'output' items
# which the frontend strips, causing tool calls to be merged into content.
chat_id = metadata.get('chat_id')
user_message_id = metadata.get('user_message_id')
if chat_id and user_message_id and not chat_id.startswith('local:') and not chat_id.startswith('channel:'):
db_messages = await load_messages_from_db(chat_id, user_message_id)
if db_messages:
# Continue: frontend sends assistant_message_id when continuing
# an existing response. Load its content so the LLM sees prior output.
assistant_message_id = metadata.get('assistant_message_id')
if assistant_message_id:
assistant_message = await Chats.get_message_by_id_and_message_id(chat_id, assistant_message_id)
if assistant_message and (assistant_message.get('content') or assistant_message.get('output')):
db_messages.append(
{k: v for k, v in assistant_message.items() if k in ('role', 'content', 'output', 'files')}
)
system_message = get_system_message(form_data.get('messages', []))
form_data['messages'] = [system_message, *db_messages] if system_message else db_messages
# Inject image files into content as image_url parts (mirrors frontend logic)
for message in form_data['messages']:
image_files = [
f
for f in message.get('files', [])
if f.get('type') == 'image' or (f.get('content_type') or '').startswith('image/')
]
if message.get('role') == 'user' and image_files:
text_content = message.get('content', '')
if isinstance(text_content, str):
message['content'] = [
{'type': 'text', 'text': text_content},
*[
{
'type': 'image_url',
'image_url': {'url': f['url']},
}
for f in image_files
if f.get('url')
],
]
# Strip files field — it's been incorporated into content
message.pop('files', None)
if regeneration_prompt:
form_data['messages'].append({'role': 'user', 'content': regeneration_prompt})
# Process messages with OR-aligned output items for clean LLM messages
form_data['messages'] = process_messages_with_output(
form_data.get('messages', []),
reasoning_format=get_reasoning_format(model),
)
system_message = get_system_message(form_data.get('messages', []))
if system_message: # Chat Controls/User Settings
try:
form_data = await apply_system_prompt_to_body(
system_message.get('content'), form_data, metadata, user, replace=True
) # Required to handle system prompt variables
except Exception:
pass
form_data = await convert_url_images_to_base64(form_data)
event_emitter = await get_event_emitter(metadata)
event_caller = await get_event_call(metadata)
extra_params = {
'__event_emitter__': event_emitter,
'__event_call__': event_caller,
'__user__': user.model_dump() if isinstance(user, UserModel) else {},
'__metadata__': metadata,
'__oauth_token__': await get_system_oauth_token(request, user),
'__request__': request,
'__model__': model,
'__chat_id__': metadata.get('chat_id'),
'__message_id__': metadata.get('message_id'),
}
# Initialize events to store additional event to be sent to the client
# Initialize contexts and citation
if getattr(request.state, 'direct', False) and hasattr(request.state, 'model'):
models = {
request.state.model['id']: request.state.model,
}
else:
models = request.app.state.MODELS
task_model_id = get_task_model_id(
form_data['model'],
request.app.state.config.TASK_MODEL,
request.app.state.config.TASK_MODEL_EXTERNAL,
models,
)
events = []
sources = []
# Folder "Project" handling
# Check if the request has chat_id and is inside of a folder
# Uses lightweight column query — only fetches folder_id, not the full chat JSON blob
chat_id = metadata.get('chat_id', None)
folder_id = None
if chat_id and user:
folder_id = await Chats.get_chat_folder_id(chat_id, user.id)
# Fallback: use folder_id from metadata (temporary chats have no DB record)
if not folder_id:
folder_id = metadata.get('folder_id', None)
if folder_id and user:
folder = await Folders.get_folder_by_id_and_user_id(folder_id, user.id)
if folder and folder.data:
if 'system_prompt' in folder.data:
form_data = await apply_system_prompt_to_body(folder.data['system_prompt'], form_data, metadata, user)
if 'files' in folder.data:
# Defensive: filter to entries the caller can still read.
allowed_files = await get_accessible_folder_files(folder.data['files'], user)
if metadata.get('params', {}).get('function_calling') != 'native':
form_data['files'] = [
*allowed_files,
*form_data.get('files', []),
]
else:
# Native FC: skip RAG injection, builtin tools
# will read folder knowledge from metadata.
metadata['folder_knowledge'] = allowed_files
# Model "Knowledge" handling
user_message = get_last_user_message(form_data['messages'])
model_knowledge = model.get('info', {}).get('meta', {}).get('knowledge', False)
if model_knowledge and metadata.get('params', {}).get('function_calling') != 'native':
await event_emitter(
{
'type': 'status',
'data': {
'action': 'knowledge_search',
'query': user_message,
'done': False,
},
}
)
knowledge_files = []
for item in model_knowledge:
if item.get('collection_name'):
knowledge_files.append(
{
'id': item.get('collection_name'),
'name': item.get('name'),
'legacy': True,
}
)
elif item.get('collection_names'):
knowledge_files.append(
{
'name': item.get('name'),
'type': 'collection',
'collection_names': item.get('collection_names'),
'legacy': True,
}
)
else:
knowledge_files.append(item)
files = form_data.get('files', [])
files.extend(knowledge_files)
form_data['files'] = files
variables = form_data.pop('variables', None)
payload_tools = form_data.get('tools', None) # snapshot before filters
# Process the form_data through the pipeline
try:
form_data = await process_pipeline_inlet_filter(request, form_data, user, models)
except Exception as e:
raise e
try:
filter_ids = await get_sorted_filter_ids(request, model, metadata.get('filter_ids', []))
filter_functions = await Functions.get_functions_by_ids(filter_ids)
form_data, flags = await process_filter_functions(
request=request,
filter_functions=filter_functions,
filter_type='inlet',
form_data=form_data,
extra_params=extra_params,
)
except Exception as e:
raise Exception(f'{e}')
features = form_data.pop('features', None) or {}
extra_params['__features__'] = features
if features:
if 'voice' in features and features['voice']:
if getattr(request.app.state.config, 'ENABLE_VOICE_MODE_PROMPT', True):
if request.app.state.config.VOICE_MODE_PROMPT_TEMPLATE:
template = request.app.state.config.VOICE_MODE_PROMPT_TEMPLATE
else:
template = DEFAULT_VOICE_MODE_PROMPT_TEMPLATE
form_data['messages'] = add_or_update_system_message(
template,
form_data['messages'],
)
if 'memory' in features and features['memory']:
# Skip forced memory injection when native FC is enabled - model can use memory tools
if metadata.get('params', {}).get('function_calling') != 'native':
form_data = await chat_memory_handler(request, form_data, extra_params, user)
if 'web_search' in features and features['web_search']:
# Skip forced RAG web search when native FC is enabled - model can use web_search tool
if metadata.get('params', {}).get('function_calling') != 'native':
form_data = await chat_web_search_handler(request, form_data, extra_params, user)
if 'image_generation' in features and features['image_generation']:
# Skip forced image generation when native FC is enabled - model can use generate_image tool
if metadata.get('params', {}).get('function_calling') != 'native':
form_data = await chat_image_generation_handler(request, form_data, extra_params, user)
if 'code_interpreter' in features and features['code_interpreter']:
engine = getattr(request.app.state.config, 'CODE_INTERPRETER_ENGINE', 'pyodide')
# Skip XML-tag prompt injection when native FC is enabled —
# execute_code will be injected as a builtin tool instead
if metadata.get('params', {}).get('function_calling') != 'native':
prompt = (
request.app.state.config.CODE_INTERPRETER_PROMPT_TEMPLATE
if request.app.state.config.CODE_INTERPRETER_PROMPT_TEMPLATE != ''
else DEFAULT_CODE_INTERPRETER_PROMPT
)
# Append filesystem awareness only for pyodide engine
if engine != 'jupyter':
prompt += CODE_INTERPRETER_PYODIDE_PROMPT
form_data['messages'] = add_or_update_user_message(
prompt,
form_data['messages'],
)
else:
# Native FC: tool docstring can't be dynamic, so inject
# filesystem context into the system message for pyodide
# engine. Appending to the system prompt (instead of the
# user message) keeps it in the stable cached prefix so
# providers with prefix caching don't re-bill the full
# conversation on every turn.
if engine != 'jupyter':
form_data['messages'] = add_or_update_system_message(
CODE_INTERPRETER_PYODIDE_PROMPT,
form_data['messages'],
append=True,
)
tool_ids = form_data.pop('tool_ids', None)
terminal_id = form_data.pop('terminal_id', None)
files = form_data.pop('files', None)
form_data.pop('folder_id', None)
# If the original caller provided tools, use them as-is (skip resolution).
# Otherwise, save any tools that filter inlets added for merging later.
inlet_filter_tools = None if payload_tools else form_data.get('tools', None)
# Skills — extract IDs from message content (<$skillId|label> tags) so
# persisted chats work without relying on the frontend to send skill_ids.
user_skill_ids = set(form_data.pop('skill_ids', None) or [])
user_skill_ids |= extract_skill_ids_from_messages(form_data.get('messages', []))
model_skill_ids = set(model.get('info', {}).get('meta', {}).get('skillIds', []))
all_skill_ids = user_skill_ids | model_skill_ids
available_skills = []
if all_skill_ids:
from open_webui.models.skills import Skills as SkillsModel
accessible_skill_ids = {s.id for s in await SkillsModel.get_skills_by_user_id(user.id, 'read')}
available_skills = []
for sid in all_skill_ids:
if sid in accessible_skill_ids:
s = await SkillsModel.get_skill_by_id(sid)
if s and s.is_active:
available_skills.append(s)
skill_descriptions = ''
for skill in available_skills:
if skill.id in user_skill_ids:
# User-selected: inject full content
form_data['messages'] = add_or_update_system_message(
f'\n{skill.content}\n',
form_data['messages'],
append=True,
)
else:
# Model-attached: name+description only
skill_descriptions += f'\n{skill.id}\n{skill.name}\n{skill.description or ""}\n\n'
if skill_descriptions:
form_data['messages'] = add_or_update_system_message(
f'\n{skill_descriptions}',
form_data['messages'],
append=True,
)
# Strip <$skillId|label> mention tags so the model doesn't see raw markup.
strip_skill_mentions(form_data.get('messages', []))
prompt = get_last_user_message(form_data['messages'])
# TODO: re-enable URL extraction from prompt
# urls = []
# if prompt and len(prompt or "") < 500 and (not files or len(files) == 0):
# urls = extract_urls(prompt)
if files:
if not files:
files = []
for file_item in files:
if file_item.get('type', 'file') == 'folder':
# Get folder files
folder_id = file_item.get('id', None)
if folder_id:
folder = await Folders.get_folder_by_id_and_user_id(folder_id, user.id)
if folder and folder.data and 'files' in folder.data:
files = [f for f in files if f.get('id', None) != folder_id]
files = [*files, *await get_accessible_folder_files(folder.data['files'], user)]
# files = [*files, *[{"type": "url", "url": url, "name": url} for url in urls]]
# Remove duplicate files based on their content
files = list({json.dumps(f, sort_keys=True): f for f in files}.values())
metadata = {
**metadata,
'model_id': form_data.get('model'),
'tool_ids': tool_ids,
'terminal_id': terminal_id,
'files': files,
}
form_data['metadata'] = metadata
# When the caller provides an explicit OpenAI-style `tools` array in the
# request body, skip all server-side tool resolution and pass the caller's
# tools through to the model unchanged.
if not payload_tools:
# Server side tools
tool_ids = metadata.get('tool_ids', None)
# Client side tools
direct_tool_servers = metadata.get('tool_servers', None)
log.debug(f'{tool_ids=}')
log.debug(f'{direct_tool_servers=}')
tools_dict = {}
mcp_clients = {}
mcp_tools_dict = {}
if tool_ids:
for tool_id in tool_ids:
if tool_id.startswith('server:mcp:'):
try:
server_id = tool_id[len('server:mcp:') :]
mcp_server_connection = None
for server_connection in request.app.state.config.TOOL_SERVER_CONNECTIONS:
if (
server_connection.get('type', '') == 'mcp'
and server_connection.get('info', {}).get('id') == server_id
):
mcp_server_connection = server_connection
break
if not mcp_server_connection:
log.error(f'MCP server with id {server_id} not found')
continue
# Check access control for MCP server
if not await has_connection_access(user, mcp_server_connection):
log.warning(f'Access denied to MCP server {server_id} for user {user.id}')
continue
auth_type = mcp_server_connection.get('auth_type', '')
headers = {}
if auth_type == 'bearer':
headers['Authorization'] = f'Bearer {mcp_server_connection.get("key", "")}'
elif auth_type == 'none':
# No authentication
pass
elif auth_type == 'session':
headers['Authorization'] = f'Bearer {request.state.token.credentials}'
elif auth_type == 'system_oauth':
oauth_token = extra_params.get('__oauth_token__', None)
if oauth_token:
headers['Authorization'] = f'Bearer {oauth_token.get("access_token", "")}'
elif auth_type in ('oauth_2.1', 'oauth_2.1_static'):
try:
splits = server_id.split(':')
server_id = splits[-1] if len(splits) > 1 else server_id
oauth_token = await request.app.state.oauth_client_manager.get_oauth_token(
user.id, f'mcp:{server_id}'
)
if oauth_token:
headers['Authorization'] = f'Bearer {oauth_token.get("access_token", "")}'
except Exception as e:
log.error(f'Error getting OAuth token: {e}')
oauth_token = None
connection_headers = mcp_server_connection.get('headers', None)
if connection_headers and isinstance(connection_headers, dict):
for key, value in connection_headers.items():
headers[key] = value
# Add user info headers if enabled
if ENABLE_FORWARD_USER_INFO_HEADERS and user:
headers = include_user_info_headers(headers, user)
if metadata and metadata.get('chat_id'):
headers[FORWARD_SESSION_INFO_HEADER_CHAT_ID] = metadata.get('chat_id')
if metadata and metadata.get('message_id'):
headers[FORWARD_SESSION_INFO_HEADER_MESSAGE_ID] = metadata.get('message_id')
mcp_clients[server_id] = MCPClient()
await mcp_clients[server_id].connect(
url=mcp_server_connection.get('url', ''),
headers=headers if headers else None,
)
function_name_filter_list = mcp_server_connection.get('config', {}).get(
'function_name_filter_list', ''
)
if isinstance(function_name_filter_list, str):
function_name_filter_list = function_name_filter_list.split(',')
tool_specs = await mcp_clients[server_id].list_tool_specs()
for tool_spec in tool_specs:
async def make_tool_function(client, function_name):
async def tool_function(**kwargs):
return await client.call_tool(
function_name,
function_args=kwargs,
)
return tool_function
if function_name_filter_list:
if not is_string_allowed(tool_spec['name'], function_name_filter_list):
# Skip this function
continue
tool_function = await make_tool_function(mcp_clients[server_id], tool_spec['name'])
mcp_tools_dict[f'{server_id}_{tool_spec["name"]}'] = {
'spec': {
**tool_spec,
'name': f'{server_id}_{tool_spec["name"]}',
},
'callable': tool_function,
'type': 'mcp',
'client': mcp_clients[server_id],
'direct': False,
}
except Exception as e:
log.debug(e)
if event_emitter:
await event_emitter(
{
'type': 'chat:message:error',
'data': {'error': {'content': f"Failed to connect to MCP server '{server_id}'"}},
}
)
continue
tools_dict = await get_tools(
request,
tool_ids,
user,
{
**extra_params,
'__model__': models[task_model_id],
'__messages__': form_data['messages'],
'__files__': metadata.get('files', []),
},
)
if mcp_tools_dict:
tools_dict = {**tools_dict, **mcp_tools_dict}
# Resolve terminal tools if terminal_id is set (outside tool_ids check
# so system terminals work even when no other tools are selected)
terminal_capability = (model.get('info', {}).get('meta', {}).get('capabilities') or {}).get('terminal', True)
if terminal_id and terminal_capability:
try:
terminal_result = await get_terminal_tools(
request,
terminal_id,
user,
extra_params,
)
if isinstance(terminal_result, tuple):
terminal_tools, system_prompt = terminal_result
else:
terminal_tools = terminal_result
system_prompt = None
if terminal_tools:
tools_dict = {**tools_dict, **terminal_tools}
if system_prompt:
form_data['messages'] = add_or_update_system_message(
system_prompt,
form_data['messages'],
append=True,
)
except Exception as e:
log.exception(e)
if direct_tool_servers:
for tool_server in direct_tool_servers:
system_prompt = tool_server.pop('system_prompt', None)
if system_prompt:
form_data['messages'] = add_or_update_system_message(
system_prompt,
form_data['messages'],
append=True,
)
tool_specs = tool_server.pop('specs', [])
for tool in tool_specs:
tools_dict[tool['name']] = {
'spec': tool,
'direct': True,
'server': tool_server,
}
if mcp_clients:
metadata['mcp_clients'] = mcp_clients
# Inject builtin tools for native function calling based on enabled features and model capability
# Check if builtin_tools capability is enabled for this model (defaults to True if not specified)
builtin_tools_enabled = (model.get('info', {}).get('meta', {}).get('capabilities') or {}).get(
'builtin_tools', True
)
if metadata.get('params', {}).get('function_calling') == 'native' and builtin_tools_enabled:
# Add file context to user messages
chat_id = metadata.get('chat_id')
form_data['messages'] = await add_file_context(form_data.get('messages', []), chat_id, user)
builtin_tools = await get_builtin_tools(
request,
{
**extra_params,
'__event_emitter__': event_emitter,
'__skill_ids__': [s.id for s in available_skills if s.id not in user_skill_ids],
},
features,
model,
)
for name, tool_dict in builtin_tools.items():
if name not in tools_dict:
tools_dict[name] = tool_dict
if tools_dict:
# Always store resolved tools in metadata so downstream consumers
# (e.g. pipe functions) can access all tools including MCP and builtins.
metadata['tools'] = tools_dict
if metadata.get('params', {}).get('function_calling') == 'native':
# If the function calling is native, then call the tools function calling handler
form_data['tools'] = [
{'type': 'function', 'function': tool.get('spec', {})} for tool in tools_dict.values()
]
if inlet_filter_tools:
form_data['tools'].extend(inlet_filter_tools)
else:
# If the function calling is not native, then call the tools function calling handler
try:
form_data, flags = await chat_completion_tools_handler(
request, form_data, extra_params, user, models, tools_dict
)
sources.extend(flags.get('sources', []))
except Exception as e:
log.exception(e)
# Check if file context extraction is enabled for this model (default True)
file_context_enabled = (model.get('info', {}).get('meta', {}).get('capabilities') or {}).get('file_context', True)
if file_context_enabled:
try:
form_data, flags = await chat_completion_files_handler(request, form_data, extra_params, user)
sources.extend(flags.get('sources', []))
except Exception as e:
log.exception(e)
# Save the pre-RAG message state so the native tool call loop can
# restore to the true original (before file-source injection) rather
# than a snapshot that already has the RAG template baked in.
system_message = get_system_message(form_data['messages'])
metadata['system_prompt'] = get_content_from_message(system_message) if system_message else None
metadata['user_prompt'] = get_last_user_message(form_data['messages'])
metadata['sources'] = sources[:] if sources else []
# If context is not empty, insert it into the messages
if sources and prompt:
form_data['messages'] = await apply_source_context_to_messages(request, form_data['messages'], sources, prompt)
# If there are citations, add them to the data_items
sources = [
source
for source in sources
if source.get('source', {}).get('name', '') or source.get('source', {}).get('id', '')
]
if len(sources) > 0:
events.append({'sources': sources})
if model_knowledge:
await event_emitter(
{
'type': 'status',
'data': {
'action': 'knowledge_search',
'query': user_message,
'done': True,
'hidden': True,
},
}
)
# Strip empty text content blocks from multimodal messages
# to prevent errors from providers like Gemini and Claude
form_data['messages'] = strip_empty_content_blocks(form_data.get('messages', []))
# Merge any duplicate system messages into a single message at position 0
# to prevent template parsing errors with strict chat templates (e.g. Qwen)
form_data['messages'] = merge_system_messages(form_data.get('messages', []))
return form_data, metadata, events
async def get_event_emitter_and_caller(metadata):
event_emitter = None
event_caller = None
# event_emitter only needs user_id + chat_id + message_id.
# It broadcasts to user:{user_id} room AND persists to DB,
# so it works for backend-initiated calls (automations, API).
if metadata.get('chat_id') and metadata.get('message_id'):
event_emitter = await get_event_emitter(metadata)
# event_caller needs session_id — it calls back to a specific
# websocket session (used by direct tools, pyodide code interpreter).
if metadata.get('session_id') and metadata.get('chat_id') and metadata.get('message_id'):
event_caller = await get_event_call(metadata)
return event_emitter, event_caller
async def build_chat_response_context(request, form_data, user, model, metadata, tasks, events):
event_emitter, event_caller = await get_event_emitter_and_caller(metadata)
return {
'request': request,
'form_data': form_data,
'user': user,
'model': model,
'metadata': metadata,
'tasks': tasks,
'events': events,
'event_emitter': event_emitter,
'event_caller': event_caller,
}
def get_response_data(response):
if isinstance(response, list) and len(response) == 1:
# If the response is a single-item list, unwrap it #17213
response = response[0]
if isinstance(response, JSONResponse):
if isinstance(response.body, bytes):
try:
response_data = json.loads(response.body.decode('utf-8', 'replace'))
except json.JSONDecodeError:
response_data = {'error': {'detail': 'Invalid JSON response'}}
else:
response_data = response
elif isinstance(response, dict):
response_data = response
else:
response_data = None
return response, response_data
def merge_events_into_response(response_data, events):
if events and isinstance(events, list):
extra_response = {}
for event in events:
if isinstance(event, dict):
extra_response.update(event)
else:
extra_response[event] = True
return {
**extra_response,
**response_data,
}
return response_data
def build_response_object(response, response_data):
if isinstance(response, dict):
return response_data
if isinstance(response, JSONResponse):
return JSONResponse(
content=response_data,
headers=response.headers,
status_code=response.status_code,
)
return response
async def get_system_oauth_token(request, user):
"""Get the system OAuth token for a user.
Primary path: use the oauth_session_id cookie (browser requests).
Fallback: look up the user's most recent OAuth session from the DB
(covers automations, API calls, and other cookie-less contexts).
"""
oauth_token = None
try:
oauth_session_id = request.cookies.get('oauth_session_id', None)
if oauth_session_id:
oauth_token = await request.app.state.oauth_manager.get_oauth_token(
user.id,
oauth_session_id,
)
# Fallback: no cookie (automation, API key, etc.) — use most recent session
if oauth_token is None:
from open_webui.models.oauth_sessions import OAuthSessions
sessions = await OAuthSessions.get_sessions_by_user_id(user.id)
if sessions:
best = max(sessions, key=lambda s: s.updated_at)
oauth_token = await request.app.state.oauth_manager.get_oauth_token(
user.id,
best.id,
)
except Exception as e:
log.error(f'Error getting OAuth token: {e}')
return oauth_token
async def background_tasks_handler(ctx):
request = ctx['request']
form_data = ctx['form_data']
user = ctx['user']
metadata = ctx['metadata']
tasks = ctx['tasks']
event_emitter = ctx['event_emitter']
message = None
messages = []
if (
'chat_id' in metadata
and not metadata['chat_id'].startswith('local:')
and not metadata['chat_id'].startswith('channel:')
):
messages_map = await Chats.get_messages_map_by_chat_id(metadata['chat_id'])
message = messages_map.get(metadata['message_id']) if messages_map else None
message_list = get_message_list(messages_map, metadata['message_id'])
# Remove details tags and files from the messages.
# as get_message_list creates a new list, it does not affect
# the original messages outside of this handler
messages = []
for message in message_list:
content = message.get('content', '')
if isinstance(content, list):
for item in content:
if item.get('type') == 'text':
content = item['text']
break
if isinstance(content, str):
content = re.sub(
r']*>.*?<\/details>|!\[.*?\]\(.*?\)',
'',
content,
flags=re.S | re.I,
).strip()
messages.append(
{
**message,
'role': message.get('role', 'assistant'), # Safe fallback for missing role
'content': content,
}
)
else:
# Local temp chat, get the model and message from the form_data
message = get_last_user_message_item(form_data.get('messages', []))
messages = form_data.get('messages', [])
if message:
message['model'] = form_data.get('model')
if message and 'model' in message:
if tasks and messages:
if TASKS.FOLLOW_UP_GENERATION in tasks and tasks[TASKS.FOLLOW_UP_GENERATION]:
res = await generate_follow_ups(
request,
{
'model': message['model'],
'messages': messages,
'message_id': metadata['message_id'],
'chat_id': metadata['chat_id'],
},
user,
)
if res and isinstance(res, dict):
if len(res.get('choices', [])) == 1:
response_message = res.get('choices', [])[0].get('message', {})
follow_ups_string = response_message.get('content') or response_message.get(
'reasoning_content', ''
)
else:
follow_ups_string = ''
follow_ups_string = follow_ups_string[
follow_ups_string.find('{') : follow_ups_string.rfind('}') + 1
]
try:
follow_ups = json.loads(follow_ups_string).get('follow_ups', [])
await event_emitter(
{
'type': 'chat:message:follow_ups',
'data': {
'follow_ups': follow_ups,
},
}
)
if not metadata.get('chat_id', '').startswith('local:') and not metadata.get(
'chat_id', ''
).startswith('channel:'):
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{
'followUps': follow_ups,
},
)
except Exception as e:
pass
if not metadata.get('chat_id', '').startswith('local:') and not metadata.get('chat_id', '').startswith(
'channel:'
): # Only update titles and tags for non-temp chats
if TASKS.TITLE_GENERATION in tasks:
user_message = get_last_user_message(messages)
if user_message and len(user_message) > 100:
user_message = user_message[:100] + '...'
title = None
if tasks[TASKS.TITLE_GENERATION]:
res = await generate_title(
request,
{
'model': message['model'],
'messages': messages,
'chat_id': metadata['chat_id'],
},
user,
)
if res and isinstance(res, dict):
if len(res.get('choices', [])) == 1:
response_message = res.get('choices', [])[0].get('message', {})
title_string = (
response_message.get('content')
or response_message.get(
'reasoning_content',
)
or message.get('content', user_message)
)
else:
title_string = ''
title_string = title_string[title_string.find('{') : title_string.rfind('}') + 1]
try:
title = json.loads(title_string).get('title', user_message)
except Exception as e:
title = ''
if not title:
title = messages[0].get('content', user_message)
await Chats.update_chat_title_by_id(metadata['chat_id'], title)
await event_emitter(
{
'type': 'chat:title',
'data': title,
}
)
if title == None and len(messages) == 2 and (not messages_map or len(messages_map) <= 2):
title = messages[0].get('content', user_message)
await Chats.update_chat_title_by_id(metadata['chat_id'], title)
await event_emitter(
{
'type': 'chat:title',
'data': message.get('content', user_message),
}
)
if TASKS.TAGS_GENERATION in tasks and tasks[TASKS.TAGS_GENERATION]:
res = await generate_chat_tags(
request,
{
'model': message['model'],
'messages': messages,
'chat_id': metadata['chat_id'],
},
user,
)
if res and isinstance(res, dict):
if len(res.get('choices', [])) == 1:
response_message = res.get('choices', [])[0].get('message', {})
tags_string = response_message.get('content') or response_message.get(
'reasoning_content', ''
)
else:
tags_string = ''
tags_string = tags_string[tags_string.find('{') : tags_string.rfind('}') + 1]
try:
tags = json.loads(tags_string).get('tags', [])
await Chats.update_chat_tags_by_id(metadata['chat_id'], tags, user)
await event_emitter(
{
'type': 'chat:tags',
'data': tags,
}
)
except Exception as e:
pass
async def outlet_filter_handler(ctx):
"""Run outlet filters inline after chat completion.
Replaces the separate POST /api/chat/completed round-trip.
Persists outlet-modified content to DB and emits a chat:outlet event
so the frontend can sync its in-memory state.
For temp chats (local: prefix), messages are built from form_data
plus the assistant response message stored in ctx['assistant_message'],
since temp chats have no DB-persisted history.
"""
request = ctx['request']
user = ctx['user']
model = ctx['model']
metadata = ctx['metadata']
event_emitter = ctx.get('event_emitter')
event_caller = ctx.get('event_caller')
chat_id = metadata.get('chat_id', '')
message_id = metadata.get('message_id')
if not chat_id or not message_id:
return
is_temp_chat = chat_id.startswith('local:') or chat_id.startswith('channel:')
try:
messages_map = None
if is_temp_chat:
# Temp chats have no DB record — build message list from
# the in-memory form_data plus the assistant response.
form_messages = ctx.get('form_data', {}).get('messages', [])
assistant_message = ctx.get('assistant_message', {})
message_list = [
{
'role': m.get('role'),
'content': m.get('content', ''),
}
for m in form_messages
]
# Append the full assistant message (content, output, usage, etc.)
if assistant_message:
message_list.append(
{
'id': message_id,
'role': 'assistant',
**assistant_message,
}
)
else:
messages_map = await Chats.get_messages_map_by_chat_id(chat_id)
if not messages_map:
return
message_list = get_message_list(messages_map, message_id)
if not message_list:
return
model_id = model.get('id') if isinstance(model, dict) else model
outlet_data = {
'model': model_id,
'messages': [
{
'id': m.get('id'),
'role': m.get('role'),
'content': m.get('content', ''),
'info': m.get('info'),
'timestamp': m.get('timestamp'),
**({'output': m['output']} if m.get('output') else {}),
**({'usage': m['usage']} if m.get('usage') else {}),
**({'sources': m['sources']} if m.get('sources') else {}),
}
for m in message_list
],
'filter_ids': metadata.get('filter_ids', []),
'chat_id': chat_id,
'session_id': metadata.get('session_id'),
'id': message_id,
}
# Pipeline outlet filters
models = request.app.state.MODELS
try:
outlet_data = await process_pipeline_outlet_filter(request, outlet_data, user, models)
except Exception as e:
log.debug(f'Pipeline outlet filter error: {e}')
# Function outlet filters
extra_params = {
'__event_emitter__': event_emitter,
'__event_call__': event_caller,
'__user__': user.model_dump() if isinstance(user, UserModel) else {},
'__metadata__': metadata,
'__request__': request,
'__model__': model,
}
filter_ids = await get_sorted_filter_ids(request, model, metadata.get('filter_ids', []))
filter_functions = await Functions.get_functions_by_ids(filter_ids)
outlet_result, _ = await process_filter_functions(
request=request,
filter_functions=filter_functions,
filter_type='outlet',
form_data=outlet_data,
extra_params=extra_params,
)
# Persist outlet-modified content and notify frontend
# (skip DB persistence for temp chats — they have no DB record)
if outlet_result and outlet_result.get('messages'):
if not is_temp_chat and messages_map:
for message in outlet_result['messages']:
outlet_message_id = message.get('id')
if outlet_message_id and outlet_message_id in messages_map:
original_message = messages_map[outlet_message_id]
if original_message.get('content') != message.get('content'):
await Chats.upsert_message_to_chat_by_id_and_message_id(
chat_id,
outlet_message_id,
{
'content': message['content'],
'originalContent': original_message.get('content'),
},
)
if event_emitter:
await event_emitter(
{
'type': 'chat:outlet',
'data': {'messages': outlet_result['messages']},
}
)
except Exception as e:
log.debug(f'Error running outlet filters: {e}')
async def non_streaming_chat_response_handler(response, ctx):
request = ctx['request']
user = ctx['user']
metadata = ctx['metadata']
events = ctx['events']
event_emitter = ctx['event_emitter']
response, response_data = get_response_data(response)
if response_data is None:
return response
if event_emitter:
try:
if 'error' in response_data:
error = response_data.get('error')
if isinstance(error, dict):
error = error.get('detail', error)
else:
error = str(error)
log.error('Provider returned error (non-streaming): %s', error)
if not metadata['chat_id'].startswith('channel:'):
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{
'error': {'content': error},
},
)
if isinstance(error, str) or isinstance(error, dict):
await event_emitter(
{
'type': 'chat:message:error',
'data': {'error': {'content': error}},
}
)
if 'selected_model_id' in response_data and not metadata['chat_id'].startswith('channel:'):
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{
'selectedModelId': response_data['selected_model_id'],
},
)
choices = response_data.get('choices', [])
if choices and choices[0].get('message', {}).get('content'):
content = response_data['choices'][0]['message']['content']
if content:
await event_emitter(
{
'type': 'chat:completion',
'data': response_data,
}
)
title = (
await Chats.get_chat_title_by_id(metadata['chat_id'])
if not metadata['chat_id'].startswith('channel:')
else ''
)
# Use output from backend if provided (OR-compliant backends),
# otherwise generate from response content
response_output = response_data.get('output')
if not response_output:
response_output = [
{
'type': 'message',
'id': output_id('msg'),
'status': 'completed',
'role': 'assistant',
'content': [{'type': 'output_text', 'text': content}],
}
]
await event_emitter(
{
'type': 'chat:completion',
'data': {
'done': True,
'content': content,
'output': response_output,
'title': title,
},
}
)
# Save message in the database
usage = normalize_usage(response_data.get('usage', {}) or {})
if not metadata['chat_id'].startswith('channel:'):
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{
'done': True,
'role': 'assistant',
'content': content,
'output': response_output,
**({'usage': usage} if usage else {}),
},
)
# Send a webhook notification if the user is not active
if request.app.state.config.ENABLE_USER_WEBHOOKS and not await Users.is_user_active(user.id):
webhook_url = await Users.get_user_webhook_url_by_id(user.id)
if webhook_url:
await post_webhook(
request.app.state.WEBUI_NAME,
webhook_url,
f'{content}\n\n{title} - {request.app.state.config.WEBUI_URL}/c/{metadata["chat_id"]}',
{
'action': 'chat',
'message': content,
'title': title,
'url': f'{request.app.state.config.WEBUI_URL}/c/{metadata["chat_id"]}',
},
)
await background_tasks_handler(ctx)
ctx['assistant_message'] = {
'content': content,
'output': response_output,
**({'usage': usage} if usage else {}),
}
await outlet_filter_handler(ctx)
response = build_response_object(response, merge_events_into_response(response_data, events))
except Exception as e:
log.debug(f'Error occurred while processing request: {e}')
pass
return response
if isinstance(response, dict):
response = merge_events_into_response(response_data, events)
return response
async def streaming_chat_response_handler(response, ctx):
request = ctx['request']
form_data = ctx['form_data']
user = ctx['user']
model = ctx['model']
metadata = ctx['metadata']
events = ctx['events']
event_emitter = ctx['event_emitter']
event_caller = ctx['event_caller']
extra_params = {
'__event_emitter__': event_emitter,
'__event_call__': event_caller,
'__user__': user.model_dump() if isinstance(user, UserModel) else {},
'__metadata__': metadata,
'__oauth_token__': await get_system_oauth_token(request, user),
'__request__': request,
'__model__': model,
}
filter_functions = [
await Functions.get_function_by_id(filter_id)
for filter_id in await get_sorted_filter_ids(request, model, metadata.get('filter_ids', []))
]
# Standard streaming response handler
# event_caller is optional — only needed for direct (client-side) tools
# and pyodide code interpreter. Server-side tools work without it.
if event_emitter:
task_id = str(uuid4()) # Create a unique task ID.
model_id = form_data.get('model', '')
# Handle as a background task
async def response_handler(response, events):
def tag_output_handler(content_type, tags, output):
"""
Detect special tags (reasoning, solution, code_interpreter) in streaming
content and create corresponding OR-aligned output items directly.
Operates on output items instead of content_blocks.
Uses the text from the output items themselves for tag detection,
eliminating state divergence between accumulated content and items.
"""
end_flag = False
def extract_attributes(tag_content):
"""Extract attributes from a tag if they exist."""
attributes = {}
if not tag_content:
return attributes
matches = re.findall(r'(\w+)\s*=\s*"([^"]+)"', tag_content)
for key, value in matches:
attributes[key] = value
return attributes
def get_last_text(out):
"""Get text from last message item, or empty string."""
if out and out[-1].get('type') == 'message':
parts = out[-1].get('content', [])
if parts and parts[-1].get('type') == 'output_text':
return parts[-1].get('text', '')
return ''
def set_last_text(out, text):
"""Set text on last message item's output_text."""
if out and out[-1].get('type') == 'message':
parts = out[-1].get('content', [])
if parts and parts[-1].get('type') == 'output_text':
parts[-1]['text'] = text
# Map content_type to output item type
output_type_map = {
'reasoning': 'reasoning',
'solution': 'message', # solution tags just produce text
'code_interpreter': 'open_webui:code_interpreter',
}
output_item_type = output_type_map.get(content_type, content_type)
last_type = output[-1].get('type', '') if output else ''
if last_type == 'message':
# Use the output item's own text for tag detection
item_text = get_last_text(output)
for start_tag, end_tag in tags:
start_tag_pattern = rf'{re.escape(start_tag)}'
if start_tag.startswith('<') and start_tag.endswith('>'):
start_tag_pattern = rf'<{re.escape(start_tag[1:-1])}(\s.*?)?>'
match = re.search(start_tag_pattern, item_text)
if match:
try:
attr_content = match.group(1) if match.group(1) else ''
except Exception:
attr_content = ''
attributes = extract_attributes(attr_content)
before_tag = item_text[: match.start()]
after_tag = item_text[match.end() :]
# Keep only text before the tag in the message
set_last_text(output, before_tag)
if not before_tag.strip():
# Remove empty message item
if output and output[-1].get('type') == 'message':
output.pop()
# Append the new output item
if output_item_type == 'reasoning':
output.append(
{
'type': 'reasoning',
'id': output_id('r'),
'status': 'in_progress',
'start_tag': start_tag,
'end_tag': end_tag,
'attributes': attributes,
'content': [],
'summary': None,
'started_at': time.time(),
}
)
elif output_item_type == 'open_webui:code_interpreter':
output.append(
{
'type': 'open_webui:code_interpreter',
'id': output_id('ci'),
'status': 'in_progress',
'start_tag': start_tag,
'end_tag': end_tag,
'attributes': attributes,
'lang': attributes.get('lang', 'python'),
'code': '',
'output': None,
'started_at': time.time(),
}
)
else:
# solution or other text-producing tag
output.append(
{
'type': 'message',
'id': output_id('msg'),
'status': 'in_progress',
'role': 'assistant',
'content': [{'type': 'output_text', 'text': ''}],
'_tag_type': content_type,
'start_tag': start_tag,
'end_tag': end_tag,
'attributes': attributes,
'started_at': time.time(),
}
)
if after_tag:
# Set the after_tag content on the new item
if output_item_type == 'reasoning':
output[-1]['content'] = [{'type': 'output_text', 'text': after_tag}]
elif output_item_type == 'open_webui:code_interpreter':
output[-1]['code'] = after_tag
else:
set_last_text(output, after_tag)
_, recursive_end = tag_output_handler(content_type, tags, output)
if recursive_end:
end_flag = True
break
elif (
(last_type == 'reasoning' and content_type == 'reasoning')
or (last_type == 'open_webui:code_interpreter' and content_type == 'code_interpreter')
or (last_type == 'message' and output[-1].get('_tag_type') == content_type)
):
item = output[-1]
start_tag = item.get('start_tag', '')
end_tag = item.get('end_tag', '')
end_tag_pattern = rf'{re.escape(end_tag)}'
# Get the block content from the item itself
if last_type == 'reasoning':
parts = item.get('content', [])
block_content = ''
if parts and parts[-1].get('type') == 'output_text':
block_content = parts[-1].get('text', '')
elif last_type == 'open_webui:code_interpreter':
block_content = item.get('code', '')
else:
block_content = get_last_text(output)
if re.search(end_tag_pattern, block_content):
end_flag = True
# Strip start and end tags from content
start_tag_pattern = rf'{re.escape(start_tag)}'
if start_tag.startswith('<') and start_tag.endswith('>'):
start_tag_pattern = rf'<{re.escape(start_tag[1:-1])}(\s.*?)?>'
block_content = re.sub(start_tag_pattern, '', block_content).strip()
end_tag_regex = re.compile(end_tag_pattern, re.DOTALL)
split_content = end_tag_regex.split(block_content, maxsplit=1)
block_content = split_content[0].strip() if split_content else ''
leftover_content = split_content[1].strip() if len(split_content) > 1 else ''
if block_content:
# Update the item with final content
if last_type == 'reasoning':
item['content'] = [{'type': 'output_text', 'text': block_content}]
item['ended_at'] = time.time()
item['duration'] = int(item['ended_at'] - item['started_at'])
item['status'] = 'completed'
elif last_type == 'open_webui:code_interpreter':
item['code'] = block_content
item['ended_at'] = time.time()
item['duration'] = int(item['ended_at'] - item['started_at'])
else:
set_last_text(output, block_content)
item['ended_at'] = time.time()
# Reset by appending a new message item for leftover
output.append(
{
'type': 'message',
'id': output_id('msg'),
'status': 'in_progress',
'role': 'assistant',
'content': [
{
'type': 'output_text',
'text': leftover_content,
}
],
}
)
else:
# Remove the block if content is empty
output.pop()
output.append(
{
'type': 'message',
'id': output_id('msg'),
'status': 'in_progress',
'role': 'assistant',
'content': [
{
'type': 'output_text',
'text': leftover_content,
}
],
}
)
return output, end_flag
message = await Chats.get_message_by_id_and_message_id(metadata['chat_id'], metadata['message_id'])
tool_calls = []
last_assistant_message = None
try:
if form_data['messages'][-1]['role'] == 'assistant':
last_assistant_message = get_last_assistant_message(form_data['messages'])
except Exception as e:
pass
content = (
message.get('content', '') if message else last_assistant_message if last_assistant_message else ''
)
# Initialize output: use existing from message if continuing, else create new
existing_output = message.get('output') if message else None
if existing_output:
output = existing_output
else:
# Only create an initial message item if there is content to initialize with
if content:
output = [
{
'type': 'message',
'id': output_id('msg'),
'status': 'in_progress',
'role': 'assistant',
'content': [{'type': 'output_text', 'text': content}],
}
]
else:
output = []
usage = None
prior_output = []
last_response_id = None
def full_output():
return prior_output + output if prior_output else output
reasoning_tags_param = metadata.get('params', {}).get('reasoning_tags')
DETECT_REASONING_TAGS = reasoning_tags_param is not False
DETECT_CODE_INTERPRETER = metadata.get('features', {}).get('code_interpreter', False)
reasoning_tags = []
if DETECT_REASONING_TAGS:
if isinstance(reasoning_tags_param, list) and len(reasoning_tags_param) == 2:
reasoning_tags = [(reasoning_tags_param[0], reasoning_tags_param[1])]
else:
reasoning_tags = DEFAULT_REASONING_TAGS
try:
for event in events:
await event_emitter(
{
'type': 'chat:completion',
'data': event,
}
)
# Save message in the database
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{
**event,
},
)
async def stream_body_handler(response, form_data):
nonlocal content
nonlocal usage
nonlocal output
nonlocal prior_output
nonlocal last_response_id
response_tool_calls = []
delta_count = 0
delta_chunk_size = max(
CHAT_RESPONSE_STREAM_DELTA_CHUNK_SIZE,
int(metadata.get('params', {}).get('stream_delta_chunk_size') or 1),
)
last_delta_data = None
async def flush_pending_delta_data(threshold: int = 0):
nonlocal delta_count
nonlocal last_delta_data
if delta_count >= threshold and last_delta_data:
await event_emitter(
{
'type': 'chat:completion',
'data': last_delta_data,
}
)
delta_count = 0
last_delta_data = None
async for line in response.body_iterator:
line = line.decode('utf-8', 'replace') if isinstance(line, bytes) else line
data = line
# Skip empty lines
if not data.strip():
continue
# "data:" is the prefix for each event
if not data.startswith('data:'):
continue
# Remove the prefix
data = data[len('data:') :].strip()
try:
data = json.loads(data)
data, _ = await process_filter_functions(
request=request,
filter_functions=filter_functions,
filter_type='stream',
form_data=data,
extra_params={'__body__': form_data, **extra_params},
)
if data:
if 'event' in data and not getattr(request.state, 'direct', False):
await event_emitter(data.get('event', {}))
if 'selected_model_id' in data:
model_id = data['selected_model_id']
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{
'selectedModelId': model_id,
},
)
await event_emitter(
{
'type': 'chat:completion',
'data': data,
}
)
# Check for Responses API events (type field starts with "response.")
elif data.get('type', '').startswith('response.'):
output, response_metadata = handle_responses_streaming_event(data, output)
# Emit citation sources from finalized output items
# (mirrors Chat Completions annotation handling at delta level)
if data.get('type') == 'response.output_item.done':
item = data.get('item', {})
if item.get('type') == 'message':
for part in item.get('content', []):
for annotation in part.get('annotations', []):
if annotation.get('type') == 'url_citation':
# Handle both flat (Responses API) and nested (Chat Completions) formats
url_citation = annotation.get('url_citation', annotation)
url = url_citation.get('url', '')
title = url_citation.get('title', url)
if url:
await event_emitter(
{
'type': 'source',
'data': {
'source': {
'name': title,
'url': url,
},
'document': [title],
'metadata': [
{
'source': url,
'name': title,
}
],
},
}
)
processed_data = {
'output': full_output(),
'content': serialize_output(full_output()),
}
# print(data)
# print(processed_data)
# Merge any metadata (usage, etc.)
# Strip 'done' — response.completed emits
# it but we may still need to execute tool
# calls. The outer middleware manages the
# actual completion signal.
if response_metadata:
if ENABLE_RESPONSES_API_STATEFUL:
response_id = response_metadata.pop('response_id', None)
if response_id:
last_response_id = response_id
# Normalize and capture usage for DB persistence
if response_metadata.get('usage'):
response_metadata['usage'] = normalize_usage(response_metadata['usage'])
usage = response_metadata['usage']
processed_data.update(response_metadata)
processed_data.pop('done', None)
await event_emitter(
{
'type': 'chat:completion',
'data': processed_data,
}
)
continue
else:
choices = data.get('choices', [])
# Normalize usage data to standard format
raw_usage = data.get('usage', {}) or {}
raw_usage.update(data.get('timings', {})) # llama.cpp
if raw_usage:
usage = normalize_usage(raw_usage)
await event_emitter(
{
'type': 'chat:completion',
'data': {
'usage': usage,
},
}
)
if not choices:
error = data.get('error', {})
if error:
log.error('Provider returned error (streaming): %s', error)
try:
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{
'error': {'content': error},
},
)
except Exception:
pass
await event_emitter(
{
'type': 'chat:completion',
'data': {
'error': error,
},
}
)
continue
delta = choices[0].get('delta', {})
# Handle delta annotations
annotations = delta.get('annotations')
if annotations:
for annotation in annotations:
if (
annotation.get('type') == 'url_citation'
and 'url_citation' in annotation
):
url_citation = annotation['url_citation']
url = url_citation.get('url', '')
title = url_citation.get('title', url)
await event_emitter(
{
'type': 'source',
'data': {
'source': {
'name': title,
'url': url,
},
'document': [title],
'metadata': [
{
'source': url,
'name': title,
}
],
},
}
)
delta_tool_calls = delta.get('tool_calls', None)
if delta_tool_calls:
for delta_tool_call in delta_tool_calls:
tool_call_index = delta_tool_call.get('index')
if tool_call_index is not None:
# Check if the tool call already exists
current_response_tool_call = None
for response_tool_call in response_tool_calls:
if response_tool_call.get('index') == tool_call_index:
current_response_tool_call = response_tool_call
break
if current_response_tool_call is None:
# Add the new tool call
delta_tool_call.setdefault('function', {})
delta_tool_call['function'].setdefault('name', '')
delta_tool_call['function'].setdefault('arguments', '')
response_tool_calls.append(delta_tool_call)
else:
# Update the existing tool call
delta_name = delta_tool_call.get('function', {}).get('name')
delta_arguments = delta_tool_call.get('function', {}).get(
'arguments'
)
if delta_name:
current_response_tool_call['function']['name'] = delta_name
if delta_arguments:
current_response_tool_call['function']['arguments'] += (
delta_arguments
)
# Emit pending tool calls in real-time
if response_tool_calls:
# Flush any pending text first
await flush_pending_delta_data()
# Build pending function_call output items for display
pending_fc_items = []
for tc in response_tool_calls:
call_id = tc.get('id', '')
func = tc.get('function', {})
pending_fc_items.append(
{
'type': 'function_call',
'id': call_id or output_id('fc'),
'call_id': call_id,
'name': func.get('name', ''),
'arguments': func.get('arguments', '{}'),
'status': 'in_progress',
}
)
await event_emitter(
{
'type': 'chat:completion',
'data': {
'content': serialize_output(full_output() + pending_fc_items),
},
}
)
image_urls = await get_image_urls(delta.get('images', []), request, metadata, user)
if image_urls:
image_file_list = [{'type': 'image', 'url': url} for url in image_urls]
message_files = await Chats.add_message_files_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
image_file_list,
)
if message_files is None:
message_files = image_file_list
await event_emitter(
{
'type': 'files',
'data': {'files': message_files},
}
)
value = delta.get('content')
reasoning_content = (
delta.get('reasoning_content')
or delta.get('reasoning')
or delta.get('thinking')
)
if reasoning_content:
if not output or output[-1].get('type') != 'reasoning':
reasoning_item = {
'type': 'reasoning',
'id': output_id('r'),
'status': 'in_progress',
'start_tag': '',
'end_tag': '',
'attributes': {'type': 'reasoning_content'},
'content': [],
'summary': None,
'started_at': time.time(),
}
output.append(reasoning_item)
else:
reasoning_item = output[-1]
# Append to reasoning content
parts = reasoning_item.get('content', [])
if parts and parts[-1].get('type') == 'output_text':
parts[-1]['text'] += reasoning_content
else:
reasoning_item['content'] = [
{
'type': 'output_text',
'text': reasoning_content,
}
]
data = {'content': serialize_output(full_output())}
if value:
if (
output
and output[-1].get('type') == 'reasoning'
and output[-1].get('attributes', {}).get('type') == 'reasoning_content'
):
reasoning_item = output[-1]
reasoning_item['ended_at'] = time.time()
reasoning_item['duration'] = int(
reasoning_item['ended_at'] - reasoning_item['started_at']
)
reasoning_item['status'] = 'completed'
output.append(
{
'type': 'message',
'id': output_id('msg'),
'status': 'in_progress',
'role': 'assistant',
'content': [
{
'type': 'output_text',
'text': '',
}
],
}
)
if ENABLE_CHAT_RESPONSE_BASE64_IMAGE_URL_CONVERSION:
value = await convert_markdown_base64_images(
request,
value,
{
'chat_id': metadata.get('chat_id', None),
'message_id': metadata.get('message_id', None),
},
user,
)
content = f'{content}{value}'
# Check if we're inside a tag-based block
# (reasoning, code_interpreter, or solution).
# If so, append to the existing in-progress
# item instead of creating a new message —
# otherwise tag_output_handler re-detects the
# start tag on every chunk and fragments the
# output.
last_item = output[-1] if output else None
last_item_type = last_item.get('type', '') if last_item else ''
inside_tag_block = (
last_item is not None
and last_item.get('status') == 'in_progress'
and last_item.get('attributes', {}).get('type') != 'reasoning_content'
and (
last_item_type == 'reasoning'
or last_item_type == 'open_webui:code_interpreter'
or (
last_item_type == 'message'
and last_item.get('_tag_type') is not None
)
)
)
if inside_tag_block:
# Append to the existing tag-based item
if last_item_type == 'open_webui:code_interpreter':
last_item['code'] = last_item.get('code', '') + value
elif last_item_type == 'reasoning':
parts = last_item.get('content', [])
if parts and parts[-1].get('type') == 'output_text':
parts[-1]['text'] += value
else:
last_item['content'] = [
{
'type': 'output_text',
'text': value,
}
]
else:
# solution or other _tag_type message
msg_parts = last_item.get('content', [])
if msg_parts and msg_parts[-1].get('type') == 'output_text':
msg_parts[-1]['text'] += value
else:
last_item['content'] = [
{
'type': 'output_text',
'text': value,
}
]
else:
if not output or output[-1].get('type') != 'message':
output.append(
{
'type': 'message',
'id': output_id('msg'),
'status': 'in_progress',
'role': 'assistant',
'content': [
{
'type': 'output_text',
'text': '',
}
],
}
)
# Append value to last message item's text
msg_parts = output[-1].get('content', [])
if msg_parts and msg_parts[-1].get('type') == 'output_text':
msg_parts[-1]['text'] += value
else:
output[-1]['content'] = [
{
'type': 'output_text',
'text': value,
}
]
if DETECT_REASONING_TAGS:
output, _ = tag_output_handler(
'reasoning',
reasoning_tags,
output,
)
output, _ = tag_output_handler(
'solution',
DEFAULT_SOLUTION_TAGS,
output,
)
if DETECT_CODE_INTERPRETER:
output, end = tag_output_handler(
'code_interpreter',
DEFAULT_CODE_INTERPRETER_TAGS,
output,
)
if end:
break
if ENABLE_REALTIME_CHAT_SAVE and not metadata['chat_id'].startswith('channel:'):
# Save message in the database
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{
'content': serialize_output(full_output()),
'output': full_output(),
},
)
else:
data = {
'content': serialize_output(full_output()),
}
if delta:
delta_count += 1
last_delta_data = data
if delta_count >= delta_chunk_size:
await flush_pending_delta_data(delta_chunk_size)
else:
await event_emitter(
{
'type': 'chat:completion',
'data': data,
}
)
except (asyncio.CancelledError, KeyboardInterrupt):
raise
except Exception as e:
done = 'data: [DONE]' in line
if done:
pass
else:
log.debug(f'Error: {e}')
continue
await flush_pending_delta_data()
if output:
# Clean up the last message item
if output[-1].get('type') == 'message':
parts = output[-1].get('content', [])
if parts and parts[-1].get('type') == 'output_text':
parts[-1]['text'] = parts[-1]['text'].strip()
if not parts[-1]['text']:
output.pop()
if not output:
output.append(
{
'type': 'message',
'id': output_id('msg'),
'status': 'in_progress',
'role': 'assistant',
'content': [{'type': 'output_text', 'text': ''}],
}
)
if output[-1].get('type') == 'reasoning':
reasoning_item = output[-1]
if reasoning_item.get('ended_at') is None:
reasoning_item['ended_at'] = time.time()
reasoning_item['duration'] = int(
reasoning_item['ended_at'] - reasoning_item['started_at']
)
reasoning_item['status'] = 'completed'
if response_tool_calls:
tool_calls.append(_split_tool_calls(response_tool_calls))
# Responses API path: extract function_call items from output
if not response_tool_calls and output:
# Collect call_ids that already have results,
# including those from prior_output so we don't
# re-process tool calls from a previous turn.
handled_call_ids = {
item.get('call_id')
for item in (prior_output + output)
if item.get('type') == 'function_call_output'
}
responses_api_tool_calls = []
for item in output:
if item.get('type') == 'function_call' and item.get('call_id') not in handled_call_ids:
arguments = item.get('arguments', '{}')
responses_api_tool_calls.append(
{
'id': item.get('call_id', ''),
'index': len(responses_api_tool_calls),
'function': {
'name': item.get('name', ''),
'arguments': (
arguments if isinstance(arguments, str) else json.dumps(arguments)
),
},
}
)
if responses_api_tool_calls:
tool_calls.append(_split_tool_calls(responses_api_tool_calls))
try:
await stream_body_handler(response, form_data)
finally:
if response.background:
await response.background()
tool_call_retries = 0
tool_call_sources = [] # Track citation sources from tool results
all_tool_call_sources = [] # Accumulated sources across all iterations
user_message = get_last_user_message(form_data['messages'])
# Check if citations are enabled for this model
citations_enabled = (model.get('info', {}).get('meta', {}).get('capabilities') or {}).get(
'citations', True
)
# Use the pre-RAG system content captured before the
# initial file-source injection in process_chat_payload.
# This ensures restore truly undoes the RAG template.
original_system_content = metadata.get('system_prompt')
if original_system_content is None:
original_system_message = get_system_message(form_data['messages'])
original_system_content = (
get_content_from_message(original_system_message) if original_system_message else None
)
while len(tool_calls) > 0 and tool_call_retries < CHAT_RESPONSE_MAX_TOOL_CALL_RETRIES:
tool_call_retries += 1
response_tool_calls = tool_calls.pop(0)
# Append function_call items for each tool call
# (Responses API already has them from streaming, so skip duplicates)
existing_call_ids = {item.get('call_id') for item in output if item.get('type') == 'function_call'}
for tc in response_tool_calls:
call_id = tc.get('id', '')
if call_id not in existing_call_ids:
func = tc.get('function', {})
output.append(
{
'type': 'function_call',
'id': call_id or output_id('fc'),
'call_id': call_id,
'name': func.get('name', ''),
'arguments': func.get('arguments', '{}'),
'status': 'in_progress',
}
)
await event_emitter(
{
'type': 'chat:completion',
'data': {
'content': serialize_output(full_output()),
'output': full_output(),
},
}
)
tools = metadata.get('tools', {})
results = []
for tool_call in response_tool_calls:
tool_call_id = tool_call.get('id', '')
tool_function_name = tool_call.get('function', {}).get('name', '')
tool_args = tool_call.get('function', {}).get('arguments', '{}')
tool_function_params = {}
if tool_args and tool_args.strip():
try:
# json.loads cannot be used because some models do not produce valid JSON
tool_function_params = ast.literal_eval(tool_args)
except Exception as e:
log.debug(e)
# Fallback to JSON parsing
try:
tool_function_params = json.loads(tool_args)
except Exception as e:
log.error(f'Error parsing tool call arguments: {tool_args}')
results.append(
{
'tool_call_id': tool_call_id,
'content': f'Error: Tool call arguments could not be parsed. The model generated malformed or incomplete JSON for `{tool_function_name}`. Please try again.',
}
)
continue
# Ensure arguments are valid JSON for downstream LLM integrations
log.debug(f'Parsed args from {tool_args} to {tool_function_params}')
tool_call.setdefault('function', {})['arguments'] = json.dumps(tool_function_params)
tool_result = None
tool = None
tool_type = None
direct_tool = False
if tool_function_name in tools:
tool = tools[tool_function_name]
spec = tool.get('spec', {})
tool_type = tool.get('type', '')
direct_tool = tool.get('direct', False)
try:
allowed_params = spec.get('parameters', {}).get('properties', {}).keys()
tool_function_params = {
k: v for k, v in tool_function_params.items() if k in allowed_params
}
if direct_tool:
tool_result = await event_caller(
{
'type': 'execute:tool',
'data': {
'id': str(uuid4()),
'name': tool_function_name,
'params': tool_function_params,
'server': tool.get('server', {}),
'session_id': metadata.get('session_id', None),
},
}
)
else:
tool_function = await get_updated_tool_function(
function=tool['callable'],
extra_params={
'__messages__': form_data.get('messages', []),
'__files__': metadata.get('files', []),
},
)
tool_result = await tool_function(**tool_function_params)
except Exception as e:
tool_result = str(e)
tool_result, tool_result_files, tool_result_embeds = await process_tool_result(
request,
tool_function_name,
tool_result,
tool_type,
direct_tool,
metadata,
user,
)
await terminal_event_handler(
tool_function_name,
tool_function_params,
tool_result,
event_emitter,
)
# Extract citation sources from tool results
if (
citations_enabled
and tool_function_name
in [
'search_web',
'fetch_url',
'view_file',
'view_knowledge_file',
'query_knowledge_files',
]
and tool_result
):
try:
citation_sources = get_citation_source_from_tool_result(
tool_name=tool_function_name,
tool_params=tool_function_params,
tool_result=tool_result,
tool_id=tool.get('tool_id', '') if tool else '',
)
tool_call_sources.extend(citation_sources)
except Exception as e:
log.exception(f'Error extracting citation source: {e}')
results.append(
{
'tool_call_id': tool_call_id,
'content': str(tool_result) if tool_result else '',
**({'files': tool_result_files} if tool_result_files else {}),
**({'embeds': tool_result_embeds} if tool_result_embeds else {}),
}
)
# Update function_call statuses and append function_call_output items
for tc in response_tool_calls:
call_id = tc.get('id', '')
# Mark function_call as completed
for item in output:
if item.get('type') == 'function_call' and item.get('call_id') == call_id:
item['status'] = 'completed'
# Update arguments with parsed/sanitized version
item['arguments'] = tc.get('function', {}).get('arguments', '{}')
break
for result in results:
output_parts = [{'type': 'input_text', 'text': result.get('content', '')}]
# Separate image data URIs (for LLM via input_image) from
# other files (for frontend display via files attribute).
display_files = []
for file_item in result.get('files', []):
if file_item.get('type') == 'image' and file_item.get('url', '').startswith('data:'):
# LLM-only: add as input_image part (invisible to serialize_output)
output_parts.append({'type': 'input_image', 'image_url': file_item['url']})
else:
# Frontend display (MCP images, audio, etc.)
display_files.append(file_item)
output.append(
{
'type': 'function_call_output',
'id': output_id('fco'),
'call_id': result.get('tool_call_id', ''),
'output': output_parts,
'status': 'completed',
**({'files': display_files} if display_files else {}),
**({'embeds': result.get('embeds')} if result.get('embeds') else {}),
}
)
# Append a new empty message item for the next response
output.append(
{
'type': 'message',
'id': output_id('msg'),
'status': 'in_progress',
'role': 'assistant',
'content': [{'type': 'output_text', 'text': ''}],
}
)
# Emit citation sources to the frontend for display
if citations_enabled:
for source in tool_call_sources:
await event_emitter({'type': 'source', 'data': source})
# Apply tool source context to messages for the model.
# Restoring to pre-RAG original prevents duplicating
# the RAG template across file and tool sources.
all_tool_call_sources.extend(tool_call_sources)
if all_tool_call_sources and user_message:
# Restore pre-RAG message state before re-applying
# to prevent RAG template duplication.
original_user_message = metadata.get('user_prompt') or user_message
set_last_user_message_content(
original_user_message,
form_data['messages'],
)
replace_system_message_content(
original_system_content or '',
form_data['messages'],
)
# Build context: file sources with content,
# tool sources as citation markers only.
source_ids = {}
source_context = get_source_context(
metadata.get('sources', []), source_ids
) + get_source_context(
all_tool_call_sources,
source_ids,
include_content=False,
)
source_context = source_context.strip()
if source_context:
rag_content = await rag_template(
request.app.state.config.RAG_TEMPLATE,
source_context,
user_message,
)
if RAG_SYSTEM_CONTEXT:
form_data['messages'] = add_or_update_system_message(
rag_content,
form_data['messages'],
append=True,
)
else:
form_data['messages'] = add_or_update_user_message(
rag_content,
form_data['messages'],
append=False,
)
tool_call_sources.clear()
# Strip input_image parts (large base64 data URIs) from the
# output sent to the frontend — they're only for LLM consumption
# via convert_output_to_messages.
frontend_output = []
for item in output:
if item.get('type') == 'function_call_output':
parts = item.get('output', [])
if any(p.get('type') == 'input_image' for p in parts):
item = {**item, 'output': [p for p in parts if p.get('type') != 'input_image']}
frontend_output.append(item)
await event_emitter(
{
'type': 'chat:completion',
'data': {
'content': serialize_output(output),
'output': frontend_output,
},
}
)
try:
new_form_data = {
**form_data,
'model': model_id,
'stream': True,
'metadata': metadata,
}
if ENABLE_RESPONSES_API_STATEFUL and last_response_id:
system_message = get_system_message(form_data['messages'])
new_form_data['messages'] = (
[system_message] if system_message else []
) + convert_output_to_messages(
output, raw=True, reasoning_format=get_reasoning_format(model)
)
new_form_data['previous_response_id'] = last_response_id
else:
tool_messages = convert_output_to_messages(
output, raw=True, reasoning_format=get_reasoning_format(model)
)
# Chat Completions providers don't support multimodal
# tool messages. Extract images into a user message.
image_urls = []
for message in tool_messages:
if message.get('role') == 'tool' and isinstance(message.get('content'), list):
text_parts = []
for part in message['content']:
if part.get('type') == 'input_text':
text_parts.append(part.get('text', ''))
elif part.get('type') == 'input_image':
image_urls.append(part.get('image_url', ''))
message['content'] = ''.join(text_parts)
new_form_data['messages'] = [
*form_data['messages'],
*tool_messages,
]
if image_urls:
new_form_data['messages'].append(
{
'role': 'user',
'content': [
{
'type': 'text',
'text': 'Here are the images from the tool results above. Please analyze them.',
},
*[{'type': 'image_url', 'image_url': {'url': url}} for url in image_urls],
],
}
)
res = await generate_chat_completion(
request,
new_form_data,
user,
bypass_system_prompt=True,
)
if isinstance(res, StreamingResponse):
# Save accumulated output and start fresh.
# Responses API output_index values are relative
# to the current response — a clean output list
# keeps indices aligned. The display prefix
# ensures the UI shows tool history during
# streaming.
prior_output = list(output)
# Trim the trailing empty placeholder message
# so it doesn't persist as a ghost item once
# the new stream produces real content.
if (
prior_output
and prior_output[-1].get('type') == 'message'
and prior_output[-1].get('status') == 'in_progress'
):
msg_parts = prior_output[-1].get('content', [])
if not msg_parts or (len(msg_parts) == 1 and not msg_parts[0].get('text', '').strip()):
prior_output.pop()
output = []
await stream_body_handler(res, new_form_data)
output[:0] = prior_output
prior_output = []
else:
break
except Exception as e:
log.debug(e)
break
if DETECT_CODE_INTERPRETER:
MAX_RETRIES = 5
retries = 0
while output and output[-1].get('type') == 'open_webui:code_interpreter' and retries < MAX_RETRIES:
await event_emitter(
{
'type': 'chat:completion',
'data': {
'content': serialize_output(output),
'output': output,
},
}
)
retries += 1
log.debug(f'Attempt count: {retries}')
ci_item = output[-1]
ci_output = ''
try:
if ci_item.get('attributes', {}).get('type') == 'code':
code = ci_item.get('code', '')
# Sanitize code (strips ANSI codes and markdown fences)
code = sanitize_code(code)
if CODE_INTERPRETER_BLOCKED_MODULES:
blocking_code = textwrap.dedent(f"""
import builtins
BLOCKED_MODULES = {CODE_INTERPRETER_BLOCKED_MODULES}
_real_import = builtins.__import__
async def restricted_import(name, globals=None, locals=None, fromlist=(), level=0):
if name.split('.')[0] in BLOCKED_MODULES:
importer_name = globals.get('__name__') if globals else None
if importer_name == '__main__':
raise ImportError(
f"Direct import of module {{name}} is restricted."
)
return _real_import(name, globals, locals, fromlist, level)
builtins.__import__ = restricted_import
""")
code = blocking_code + '\n' + code
if request.app.state.config.CODE_INTERPRETER_ENGINE == 'pyodide':
ci_output = await event_caller(
{
'type': 'execute:python',
'data': {
'id': str(uuid4()),
'code': code,
'session_id': metadata.get('session_id', None),
'files': metadata.get('files', []),
},
}
)
elif request.app.state.config.CODE_INTERPRETER_ENGINE == 'jupyter':
ci_output = await execute_code_jupyter(
request.app.state.config.CODE_INTERPRETER_JUPYTER_URL,
code,
(
request.app.state.config.CODE_INTERPRETER_JUPYTER_AUTH_TOKEN
if request.app.state.config.CODE_INTERPRETER_JUPYTER_AUTH == 'token'
else None
),
(
request.app.state.config.CODE_INTERPRETER_JUPYTER_AUTH_PASSWORD
if request.app.state.config.CODE_INTERPRETER_JUPYTER_AUTH == 'password'
else None
),
request.app.state.config.CODE_INTERPRETER_JUPYTER_TIMEOUT,
)
else:
ci_output = {'stdout': 'Code interpreter engine not configured.'}
log.debug(f'Code interpreter output: {ci_output}')
# Handle error responses from event_caller
# (e.g. session disconnected, timeout)
if isinstance(ci_output, dict) and ci_output.get('error'):
ci_output = {'stderr': ci_output['error']}
if isinstance(ci_output, dict):
stdout = ci_output.get('stdout', '')
if isinstance(stdout, str):
stdoutLines = stdout.split('\n')
for idx, line in enumerate(stdoutLines):
if re.match(r'data:image/\w+;base64', line):
image_url = await get_image_url_from_base64(
request,
line,
metadata,
user,
)
if image_url:
stdoutLines[idx] = f''
ci_output['stdout'] = '\n'.join(stdoutLines)
result = ci_output.get('result', '')
if isinstance(result, str):
resultLines = result.split('\n')
for idx, line in enumerate(resultLines):
if re.match(r'data:image/\w+;base64', line):
image_url = await get_image_url_from_base64(
request,
line,
metadata,
user,
)
resultLines[idx] = f''
ci_output['result'] = '\n'.join(resultLines)
except Exception as e:
ci_output = str(e)
ci_item['output'] = ci_output
ci_item['status'] = 'completed'
output.append(
{
'type': 'message',
'id': output_id('msg'),
'status': 'in_progress',
'role': 'assistant',
'content': [{'type': 'output_text', 'text': ''}],
}
)
await event_emitter(
{
'type': 'chat:completion',
'data': {
'content': serialize_output(output),
'output': output,
},
}
)
try:
new_form_data = {
**form_data,
'model': model_id,
'stream': True,
'metadata': metadata,
'messages': [
*form_data['messages'],
*convert_output_to_messages(
output, raw=True, reasoning_format=get_reasoning_format(model)
),
],
}
res = await generate_chat_completion(
request,
new_form_data,
user,
bypass_system_prompt=True,
)
if isinstance(res, StreamingResponse):
await stream_body_handler(res, new_form_data)
else:
break
except Exception as e:
log.debug(e)
break
# Mark all in-progress items as completed
for item in output:
if item.get('status') == 'in_progress':
item['status'] = 'completed'
title = (
await Chats.get_chat_title_by_id(metadata['chat_id'])
if not metadata['chat_id'].startswith('channel:')
else ''
)
data = {
'done': True,
'content': serialize_output(output),
'output': output,
'title': title,
**({'usage': usage} if usage else {}),
}
if not metadata['chat_id'].startswith('channel:'):
if not ENABLE_REALTIME_CHAT_SAVE:
# Save message in the database
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{
'done': True,
'content': serialize_output(output),
'output': output,
**({'usage': usage} if usage else {}),
},
)
elif usage:
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{'done': True, 'usage': usage},
)
else:
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{'done': True},
)
# Send a webhook notification if the user is not active
if request.app.state.config.ENABLE_USER_WEBHOOKS and not await Users.is_user_active(user.id):
webhook_url = await Users.get_user_webhook_url_by_id(user.id)
if webhook_url:
await post_webhook(
request.app.state.WEBUI_NAME,
webhook_url,
f'{content}\n\n{title} - {request.app.state.config.WEBUI_URL}/c/{metadata["chat_id"]}',
{
'action': 'chat',
'message': content,
'title': title,
'url': f'{request.app.state.config.WEBUI_URL}/c/{metadata["chat_id"]}',
},
)
await event_emitter(
{
'type': 'chat:completion',
'data': data,
}
)
await background_tasks_handler(ctx)
ctx['assistant_message'] = {
'content': serialize_output(output),
'output': output,
**({'usage': usage} if usage else {}),
}
await outlet_filter_handler(ctx)
except asyncio.CancelledError:
log.warning('Task was cancelled!')
# Close the response body iterator to trigger cleanup
# in stream_wrapper's finally block and release the
# upstream connection. Without this, the async
# generator is orphaned and may spin in anyio internals.
if hasattr(response, 'body_iterator') and hasattr(response.body_iterator, 'aclose'):
try:
await asyncio.shield(response.body_iterator.aclose())
except (asyncio.CancelledError, Exception):
pass
async def save_cancelled_state():
await event_emitter({'type': 'chat:tasks:cancel'})
if not metadata['chat_id'].startswith('channel:'):
if not ENABLE_REALTIME_CHAT_SAVE:
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{
'done': True,
'content': serialize_output(output),
'output': output,
},
)
else:
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
metadata['message_id'],
{'done': True},
)
try:
await asyncio.shield(save_cancelled_state())
except (asyncio.CancelledError, Exception):
pass
raise # re-raise CancelledError for proper propagation
if response.background is not None:
await response.background()
return await response_handler(response, events)
else:
# Fallback to the original response
async def stream_wrapper(original_generator, events):
def wrap_item(item):
return f'data: {item}\n\n'
for event in events:
event, _ = await process_filter_functions(
request=request,
filter_functions=filter_functions,
filter_type='stream',
form_data=event,
extra_params=extra_params,
)
if event:
yield wrap_item(json.dumps(event))
async for data in original_generator:
data, _ = await process_filter_functions(
request=request,
filter_functions=filter_functions,
filter_type='stream',
form_data=data,
extra_params=extra_params,
)
if data:
yield data
return StreamingResponse(
stream_wrapper(response.body_iterator, events),
headers=dict(response.headers),
background=response.background,
)
async def process_chat_response(response, ctx):
# Non-streaming response
if not isinstance(response, StreamingResponse):
return await non_streaming_chat_response_handler(response, ctx)
# Non standard response
if not any(
content_type in response.headers['Content-Type']
for content_type in ['text/event-stream', 'application/x-ndjson']
):
return response
# Streaming response
return await streaming_chat_response_handler(response, ctx)