bhetherman/open-webui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refac

Browse Source
This commit is contained in:
Timothy Jaeryang Baek
2026-08-13 19:59:11 -06:00
parent b018feb741
commit 7d99b2716a
27 changed files with 1069 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/open_webui/utils/middleware.py
+301 -10
View File
@@ -2982,6 +2982,261 @@ async def build_chat_response_context(request, form_data, user, model, metadata,
}
async def execute_tool_call_for_output(request, form_data, user, metadata, event_caller, event_emitter, tool_call):
tools = metadata.get('tools', {})
name = tool_call.get('function', {}).get('name', '')
tool_args = tool_call.get('function', {}).get('arguments', '{}')
params = {}
if tool_args and tool_args.strip():
try:
params = JSONCodec.loads(tool_args)
except Exception:
try:
params = ast.literal_eval(tool_args)
except Exception as e:
log.debug(e)
return {
'tool_call_id': tool_call.get('id', ''),
'content': (
'Error: Tool call arguments could not be parsed. '
'The model generated malformed or incomplete JSON.'
),
}
tool_call.setdefault('function', {})['arguments'] = JSONCodec.dumps(params)
tool = tools.get(name)
if not tool:
return {'tool_call_id': tool_call.get('id', ''), 'content': f'Error: Tool "{name}" not found.'}
spec = tool.get('spec', {})
tool_type = tool.get('type', '')
direct_tool = tool.get('direct', False)
allowed_params = spec.get('parameters', {}).get('properties', {}).keys()
params = {key: value for key, value in params.items() if key in allowed_params}
try:
if direct_tool:
if not event_caller:
result = 'Error: Browser session is not connected for this direct tool.'
else:
result = await event_caller(
{
'type': 'execute:tool',
'data': {
'id': str(uuid4()),
'name': name,
'params': params,
'server': tool.get('server', {}),
'session_id': metadata.get('session_id'),
},
}
)
else:
function = await get_updated_tool_function(
function=tool['callable'],
extra_params={
'__messages__': form_data.get('messages', []),
'__files__': metadata.get('files', []),
},
)
result = await function(**params)
except Exception as e:
result = str(e)
result, files, embeds = await process_tool_result(
request,
name,
result,
tool_type,
direct_tool,
metadata,
user,
)
await terminal_event_handler(name, params, result, event_emitter)
return {
'tool_call_id': tool_call.get('id', ''),
'content': str(result) if result else '',
**({'files': files} if files else {}),
**({'embeds': embeds} if embeds else {}),
}
def append_tool_result_output(output: list[dict], result: dict) -> None:
output_parts = [{'type': 'input_text', 'text': result.get('content', '')}]
display_files = []
for file_item in result.get('files', []):
if file_item.get('type') == 'image' and file_item.get('url', '').startswith('data:'):
output_parts.append({'type': 'input_image', 'image_url': file_item['url']})
else:
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 {}),
}
)
async def drain_approved_tool_calls(request, form_data, user, model, metadata) -> bool:
chat_id = metadata.get('chat_id')
message_id = metadata.get('message_id') or metadata.get('assistant_message_id')
if not is_saved_chat_id(chat_id) or not message_id:
return False
message = await Chats.get_message_by_id_and_message_id(chat_id, message_id)
output = message.get('output') if message else None
if not isinstance(output, list):
return False
result_call_ids = {
item.get('call_id') for item in output if item.get('type') == 'function_call_output' and item.get('call_id')
}
approved_calls = [
item
for item in output
if item.get('type') == 'function_call'
and item.get('call_id')
and item.get('status') == 'queued'
and item.get('approved') is True
and item.get('call_id') not in result_call_ids
]
if not approved_calls:
return False
event_emitter, event_caller = await get_event_emitter_and_caller(metadata)
changed = False
for item in approved_calls:
tool_call = {
'id': item.get('call_id', ''),
'type': 'function',
'function': {
'name': item.get('name', ''),
'arguments': item.get('arguments', '{}'),
},
}
result = await execute_tool_call_for_output(
request,
form_data,
user,
metadata,
event_caller,
event_emitter,
tool_call,
)
item['status'] = 'completed'
item['arguments'] = tool_call.get('function', {}).get('arguments', '{}')
append_tool_result_output(output, result)
changed = True
if changed:
result_call_ids = {
item.get('call_id') for item in output if item.get('type') == 'function_call_output' and item.get('call_id')
}
paused = any(
item.get('type') == 'function_call'
and item.get('call_id')
and item.get('status') in {'pending', 'queued', 'requires_approval'}
and item.get('call_id') not in result_call_ids
for item in output
)
if not paused:
output.append(
{
'type': 'message',
'id': output_id('msg'),
'status': 'in_progress',
'role': 'assistant',
'content': [{'type': 'output_text', 'text': ''}],
}
)
await Chats.upsert_message_to_chat_by_id_and_message_id(
chat_id,
message_id,
{'done': False, 'output': output},
touch=False,
)
if event_emitter:
await event_emitter(
{
'type': 'chat:completion',
'data': {
'done': False,
'output': output,
},
}
)
db_messages = await load_messages_from_db(chat_id, metadata.get('user_message_id'))
if db_messages:
assistant_message = await Chats.get_message_by_id_and_message_id(chat_id, message_id)
if assistant_message:
db_messages.append(
{
k: v
for k, v in assistant_message.items()
if k in ('id', 'role', 'content', 'output', 'files', 'contextSummary', 'usage')
}
)
form_data['messages'] = process_messages_with_output(
db_messages,
reasoning_format=get_reasoning_format(model),
)
form_data['messages'] = sanitize_tool_pairs(form_data['messages'])
return paused
return False
async def pause_for_tool_approval(chat_id: str, message_id: str, output: list[dict], form_data: dict, metadata: dict):
result_call_ids = {
item.get('call_id') for item in output if item.get('type') == 'function_call_output' and item.get('call_id')
}
for item in output:
if item.get('type') == 'function_call' and not item.get('call_id') and item.get('id'):
item['call_id'] = item['id']
if (
item.get('type') == 'function_call'
and item.get('call_id')
and item.get('call_id') not in result_call_ids
and item.get('status') != 'rejected'
):
item['status'] = 'pending'
await Chats.upsert_message_to_chat_by_id_and_message_id(
chat_id,
message_id,
{
'done': False,
'output': output,
'meta': {
**(metadata.get('tool_approval') or {}),
'session_id': metadata.get('session_id'),
'tool_ids': metadata.get('tool_ids') or [],
'skill_ids': metadata.get('skill_ids') or [],
'terminal_id': metadata.get('terminal_id'),
'tool_servers': metadata.get('tool_servers'),
'filter_ids': metadata.get('filter_ids') or [],
'features': metadata.get('features') or {},
'variables': metadata.get('variables') or {},
'files': metadata.get('files') or [],
'params': metadata.get('params') or {},
},
},
touch=False,
)
def get_response_data(response):
if isinstance(response, list) and len(response) == 1:
# If the response is a single-item list, unwrap it #17213
@@ -4032,7 +4287,20 @@ async def streaming_chat_response_handler(response, ctx):
# Initialize output: use existing from message if continuing, else create new
existing_output = message.get('output') if message else None
if existing_output:
prior_output = []
if existing_output and metadata.get('assistant_message_id'):
prior_output = list(existing_output)
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 = []
content_parts = []
elif existing_output:
output = existing_output
else:
# Only create an initial message item if there is content to initialize with
@@ -4050,7 +4318,6 @@ async def streaming_chat_response_handler(response, ctx):
output = []
usage = None
prior_output = []
last_response_id = None
def full_output():
@@ -4900,6 +5167,29 @@ async def streaming_chat_response_handler(response, ctx):
}
)
tool_approval_mode = metadata.get('params', {}).get('tool_approval_mode', 'full')
if (
tool_approval_mode == 'ask'
and is_saved_chat_id(metadata.get('chat_id'))
and metadata.get('message_id')
):
await pause_for_tool_approval(
metadata['chat_id'],
metadata['message_id'],
output,
form_data,
metadata,
)
await event_emitter(
{
'type': 'chat:completion',
'data': {
'output': full_output(),
},
}
)
return
await event_emitter(
{
'type': 'chat:completion',
@@ -5293,7 +5583,7 @@ async def streaming_chat_response_handler(response, ctx):
{
'type': 'chat:completion',
'data': {
'output': output,
'output': full_output(),
},
}
)
@@ -5420,7 +5710,7 @@ async def streaming_chat_response_handler(response, ctx):
{
'type': 'chat:completion',
'data': {
'output': output,
'output': full_output(),
},
}
)
@@ -5464,10 +5754,11 @@ async def streaming_chat_response_handler(response, ctx):
if item.get('status') == 'in_progress':
item['status'] = 'completed'
current_output = full_output()
title = await Chats.get_chat_title_by_id(metadata['chat_id']) if save_to_chat else ''
data = {
'done': True,
'output': output,
'output': current_output,
'title': title,
**({'usage': usage} if usage else {}),
}
@@ -5480,7 +5771,7 @@ async def streaming_chat_response_handler(response, ctx):
metadata['message_id'],
{
'done': True,
'output': output,
'output': current_output,
**({'usage': usage} if usage else {}),
},
)
@@ -5497,7 +5788,7 @@ async def streaming_chat_response_handler(response, ctx):
{'done': True},
)
await publish_chat_finished_event(request, user, metadata, title, ''.join(content_parts), output)
await publish_chat_finished_event(request, user, metadata, title, ''.join(content_parts), current_output)
await event_emitter(
{
@@ -5507,8 +5798,8 @@ async def streaming_chat_response_handler(response, ctx):
)
ctx['assistant_message'] = {
'content': ''.join(content_parts) or get_output_text(output),
'output': output,
'content': ''.join(content_parts) or get_output_text(current_output),
'output': current_output,
**({'usage': usage} if usage else {}),
}
await outlet_filter_handler(ctx)
@@ -5535,7 +5826,7 @@ async def streaming_chat_response_handler(response, ctx):
metadata['message_id'],
{
'done': True,
'output': output,
'output': full_output(),
},
)
else:
backend/open_webui/utils/misc.py
+16 -2
View File
@@ -293,9 +293,17 @@ def convert_output_to_messages(
pending_reasoning = [] # Only populated when reasoning_format == 'reasoning_content'
pending_reasoning_details = []
pending_tool_image_urls = []
function_call_ids = {
item.get('call_id') for item in output if item.get('type') == 'function_call' and item.get('call_id')
completed_call_ids = {
item.get('call_id')
for item in output
if item.get('type') == 'function_call'
and item.get('call_id')
and item.get('status') in {'completed', 'rejected'}
}
result_call_ids = {
item.get('call_id') for item in output if item.get('type') == 'function_call_output' and item.get('call_id')
}
function_call_ids = completed_call_ids & result_call_ids
def flush_pending():
nonlocal pending_content, pending_tool_calls, pending_reasoning, pending_reasoning_details
@@ -355,6 +363,9 @@ def convert_output_to_messages(
pending_content.append(text)
elif item_type == 'function_call':
if item.get('call_id') not in function_call_ids:
continue
# Collect tool calls to batch into assistant message
arguments = item.get('arguments', '{}')
# Ensure arguments is always a JSON string
@@ -372,6 +383,9 @@ def convert_output_to_messages(
)
elif item_type == 'function_call_output':
if item.get('call_id') not in function_call_ids:
continue
# Flush any pending content/tool_calls before adding tool result
flush_pending()
backend/open_webui/utils/payload.py
+2
View File
@@ -95,6 +95,7 @@ def apply_params_to_form_data(form_data: dict, model: dict, params: dict | None
'compact_token_threshold': int,
'system': str,
'note_id': str,
'tool_approval_mode': str,
}
for key in list(params.keys()):
@@ -149,6 +150,7 @@ def remove_open_webui_params(params: dict) -> dict:
'compact_token_threshold': int,
'system': str,
'note_id': str,
'tool_approval_mode': str,
}
for key in list(params.keys()):
backend/open_webui/utils/tool_approval.py
+172
View File
@@ -0,0 +1,172 @@
from typing import Any, Literal
from fastapi import HTTPException, status
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession
from open_webui.constants import ERROR_MESSAGES
from open_webui.models.chats import Chats
from open_webui.socket.main import get_event_emitter
from open_webui.utils.json_codec import JSONCodec
class ResolveToolCallForm(BaseModel):
call_id: str
action: Literal['approve', 'reject', 'answer']
answers: Any | None = None
timed_out: bool = False
async def resolve_tool_call_output(
chat_id: str,
message_id: str,
form_data: ResolveToolCallForm,
user,
db: AsyncSession | None = None,
) -> dict:
chat = await Chats.get_chat_by_id(chat_id, db=db)
if not chat or (chat.user_id != user.id and user.role != 'admin'):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
)
message = await Chats.get_message_by_id_and_message_id(chat_id, message_id)
if not message:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)
output = message.get('output') or []
if not isinstance(output, list):
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Message has no resolvable output.')
function_call = next(
(
item
for item in output
if item.get('type') == 'function_call'
and (item.get('call_id') or item.get('id')) == form_data.call_id
),
None,
)
if not function_call:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Tool call not found.')
function_call.setdefault('call_id', form_data.call_id)
if any(
item.get('type') == 'function_call_output' and item.get('call_id') == form_data.call_id for item in output
) or function_call.get('status') not in {'pending', 'queued', 'requires_approval'}:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail='Tool call has already been resolved.')
if form_data.action == 'approve':
function_call['status'] = 'queued'
function_call['approved'] = True
elif form_data.action == 'reject':
function_call['status'] = 'rejected'
output.append(
{
'type': 'function_call_output',
'id': f'fco_{form_data.call_id}',
'call_id': form_data.call_id,
'output': [{'type': 'input_text', 'text': 'Tool call was denied by the user.'}],
'status': 'rejected',
}
)
else:
function_call['status'] = 'completed'
answer_payload = {'answers': form_data.answers, 'timed_out': form_data.timed_out}
output.append(
{
'type': 'function_call_output',
'id': f'fco_{form_data.call_id}',
'call_id': form_data.call_id,
'output': [{'type': 'input_text', 'text': JSONCodec.dumps(answer_payload)}],
'status': 'completed',
}
)
await Chats.upsert_message_to_chat_by_id_and_message_id(
chat_id,
message_id,
{
'done': False,
'output': output,
},
touch=False,
)
event_emitter = await get_event_emitter(
{
'user_id': chat.user_id,
'chat_id': chat_id,
'message_id': message_id,
},
update_db=False,
)
if event_emitter:
await event_emitter({'type': 'chat:completion', 'data': {'output': output}})
result_call_ids = {
item.get('call_id') for item in output if item.get('type') == 'function_call_output' and item.get('call_id')
}
paused = any(
item.get('type') == 'function_call'
and item.get('call_id')
and item.get('status') in {'pending', 'queued', 'requires_approval'}
and item.get('call_id') not in result_call_ids
for item in output
)
return {'chat': chat, 'message': message, 'output': output, 'paused': paused}
async def build_tool_approval_resume_payload(chat_id: str, message_id: str, chat=None) -> dict:
chat = chat or await Chats.get_chat_by_id(chat_id)
if not chat:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)
assistant_message = await Chats.get_message_by_id_and_message_id(chat_id, message_id)
if not assistant_message:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)
user_message_id = assistant_message.get('parentId')
user_message = await Chats.get_message_by_id_and_message_id(chat_id, user_message_id) if user_message_id else None
if not user_message:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail='Tool call parent message is missing.')
chat_data = chat.chat or {}
message_meta = assistant_message.get('meta') if isinstance(assistant_message.get('meta'), dict) else {}
params = {
**(chat_data.get('params') or {}),
**(message_meta.get('params') if isinstance(message_meta.get('params'), dict) else {}),
}
if 'tool_approval_mode' not in params:
params['tool_approval_mode'] = 'ask'
model_id = assistant_message.get('model') or next(iter(chat_data.get('models') or []), None)
if not model_id:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail='Tool call message model is missing.')
messages = []
if params.get('system'):
messages.append({'role': 'system', 'content': params.get('system')})
return {
'stream': params.get('stream_response', True),
'model': model_id,
'messages': messages,
'params': params,
'files': message_meta.get('files') or chat_data.get('files') or None,
'filter_ids': message_meta.get('filter_ids') or None,
'tool_ids': message_meta.get('tool_ids') or None,
'skill_ids': message_meta.get('skill_ids') or None,
'terminal_id': message_meta.get('terminal_id') or None,
'tool_servers': message_meta.get('tool_servers') or None,
'features': message_meta.get('features') or {},
'variables': message_meta.get('variables') or {},
'chat_variables': chat.variables,
'session_id': message_meta.get('session_id'),
'chat_id': chat_id,
'id': message_id,
'parent_id': user_message.get('parentId'),
'user_message': user_message,
'assistant_message_id': message_id,
}