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-07-14 00:10:28 -04:00
parent 0c23466a3e
commit 7088d245bb
18 changed files with 1329 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/open_webui/utils/middleware.py
+113 -89
View File
@@ -2592,14 +2592,16 @@ async def process_chat_payload(request, form_data, user, metadata, model):
# 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,
'features': features,
}
metadata.update(
{
'model_id': form_data.get('model'),
'tool_ids': tool_ids,
'skill_ids': list(skill_ids),
'terminal_id': terminal_id,
'files': files,
'features': features,
}
)
form_data['metadata'] = metadata
# When the caller provides an explicit `tools` key in the request body,
@@ -2717,6 +2719,7 @@ async def process_chat_payload(request, form_data, user, metadata, model):
)
except Exception as e:
log.exception(e)
raise HTTPException(status_code=503, detail=f'Terminal unavailable: {e}') from e
if direct_tool_servers:
for tool_server in direct_tool_servers:
@@ -3556,7 +3559,11 @@ async def non_streaming_chat_response_handler(response, ctx):
)
# Send a webhook notification if the user is not active
if await Config.get('ui.enable_user_webhooks') and not await Users.is_user_active(user.id):
if (
getattr(request.state, 'internal', False) is not True
and await Config.get('ui.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:
webui_url = await Config.get('webui.url')
@@ -4624,6 +4631,11 @@ async def streaming_chat_response_handler(response, ctx):
await response.background()
tool_call_iterations = 0
max_tool_call_iterations = getattr(
request.state,
'max_tool_call_iterations',
CHAT_RESPONSE_MAX_TOOL_CALL_ITERATIONS,
)
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'])
@@ -4644,8 +4656,7 @@ async def streaming_chat_response_handler(response, ctx):
)
while tool_calls and (
CHAT_RESPONSE_MAX_TOOL_CALL_ITERATIONS is None
or tool_call_iterations < CHAT_RESPONSE_MAX_TOOL_CALL_ITERATIONS
max_tool_call_iterations is None or tool_call_iterations < max_tool_call_iterations
):
tool_call_iterations += 1
@@ -4682,83 +4693,92 @@ async def streaming_chat_response_handler(response, ctx):
results = []
def parse_tool_params(tool_call):
tool_args = tool_call.get('function', {}).get('arguments', '{}')
params = {}
if tool_args and tool_args.strip():
try:
params = ast.literal_eval(tool_args)
except Exception as e:
log.debug(e)
try:
params = json.loads(tool_args)
except Exception:
return None
tool_call.setdefault('function', {})['arguments'] = json.dumps(params)
return params
async def execute_tool_call(tool_call):
name = tool_call.get('function', {}).get('name', '')
params = parse_tool_params(tool_call)
if params is None:
return {}, None, None, None, False
tool = tools.get(name)
if not tool:
return params, f'Error: Tool "{name}" not found.', None, None, False
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:
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)
return params, result, tool, tool_type, direct_tool
delegate_calls = [
tool_call
for tool_call in response_tool_calls
if tool_call.get('function', {}).get('name') == 'delegate_task'
]
tool_results = {}
for tool_call in response_tool_calls:
if tool_call.get('function', {}).get('name') != 'delegate_task':
tool_results[id(tool_call)] = await execute_tool_call(tool_call)
tool_results.update(
zip(
[id(tool_call) for tool_call in delegate_calls],
await asyncio.gather(*(execute_tool_call(tool_call) for tool_call in delegate_calls)),
)
)
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
tool_function_params, tool_result, tool, tool_type, direct_tool = tool_results[id(tool_call)]
if tool_result is None:
results.append(
{
'tool_call_id': tool_call_id,
'content': (
'Error: Tool call arguments could not be parsed. The model generated '
f'malformed or incomplete JSON for `{tool_function_name}`. Please try again.'
),
}
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)
else:
tool_result = f'Error: Tool "{tool_function_name}" not found.'
)
continue
tool_result, tool_result_files, tool_result_embeds = await process_tool_result(
request,
@@ -5030,12 +5050,12 @@ async def streaming_chat_response_handler(response, ctx):
break
if (
CHAT_RESPONSE_MAX_TOOL_CALL_ITERATIONS is not None
max_tool_call_iterations is not None
and tool_calls
and tool_call_iterations >= CHAT_RESPONSE_MAX_TOOL_CALL_ITERATIONS
and tool_call_iterations >= max_tool_call_iterations
):
log.warning('Tool-call iteration limit reached (%s)', CHAT_RESPONSE_MAX_TOOL_CALL_ITERATIONS)
error_content = f'Tool-call limit reached ({CHAT_RESPONSE_MAX_TOOL_CALL_ITERATIONS} iterations).'
log.warning('Tool-call iteration limit reached (%s)', max_tool_call_iterations)
error_content = f'Tool-call limit reached ({max_tool_call_iterations} iterations).'
if not metadata.get('chat_id', '').startswith('channel:'):
await Chats.upsert_message_to_chat_by_id_and_message_id(
metadata['chat_id'],
@@ -5261,7 +5281,11 @@ async def streaming_chat_response_handler(response, ctx):
)
# Send a webhook notification if the user is not active
if await Config.get('ui.enable_user_webhooks') and not await Users.is_user_active(user.id):
if (
getattr(request.state, 'internal', False) is not True
and await Config.get('ui.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:
webui_url = await Config.get('webui.url')
backend/open_webui/utils/subagents.py
+627
View File
@@ -0,0 +1,627 @@
from __future__ import annotations
import asyncio
import copy
import json
import time
from datetime import timedelta
from uuid import uuid4
from fastapi import Request
from fastapi.security import HTTPAuthorizationCredentials
from open_webui.models.chat_messages import ChatMessages
from open_webui.models.chats import ChatForm, Chats
from open_webui.models.config import Config
from open_webui.models.users import UserModel, Users
from open_webui.tasks import create_task, has_active_tasks
from open_webui.utils.auth import create_token
from open_webui.utils.misc import get_message_list
from starlette.datastructures import Headers
DEFAULT_SUBAGENT_SYSTEM_PROMPT = """You are a sub-agent working on a specific task assigned by the lead agent.
You have full access to the workspace — you can read, write, edit files, and run commands.
Focus exclusively on your assigned task. Do NOT work on anything outside your scope.
When done, end with a clear summary:
- What you did
- What files you changed (if any)
- Any issues or open questions
"""
MUTATING_MEMORY_TOOLS = {
'add_memory',
'delete_memory',
'replace_memory_content',
'update_memory',
}
_background_active: set[str] = set()
_background_lock = asyncio.Lock()
_foreground_semaphore: asyncio.Semaphore | None = None
_parent_locks: dict[str, asyncio.Lock] = {}
def _build_request(source: Request, user_id: str, *, internal: bool) -> Request:
scope = {
'type': 'http',
'asgi': {'version': '3.0', 'spec_version': '2.0'},
'method': 'POST',
'path': '/api/v1/subagents/internal',
'query_string': b'',
'headers': Headers({}).raw,
'client': ('127.0.0.1', 0),
'server': ('127.0.0.1', 80),
'scheme': 'http',
'app': source.app,
}
request = Request(scope)
token = create_token(
data={'id': user_id, 'typ': 'subagent'},
expires_delta=timedelta(hours=1),
)
request.state.token = HTTPAuthorizationCredentials(scheme='Bearer', credentials=token)
request.state.enable_api_keys = False
if internal:
request.state.internal = True
return request
async def process_pending_subagent_results(
source_request: Request,
parent_chat_id: str,
user_id: str,
run: dict,
) -> None:
lock = _parent_locks.setdefault(parent_chat_id, asyncio.Lock())
while await has_active_tasks(source_request.app.state.redis, parent_chat_id):
await asyncio.sleep(0.25)
async with lock:
if await has_active_tasks(source_request.app.state.redis, parent_chat_id):
return
chat = await Chats.get_chat_by_id_and_user_id(parent_chat_id, user_id)
user = await Users.get_user_by_id(user_id)
if not chat or not user:
return
history = copy.deepcopy(chat.chat.get('history') or {})
messages = history.get('messages') or {}
pending = [
message
for message in messages.values()
if message.get('role') == 'user'
and (message.get('meta') or {}).get('async_subagent_result') is True
and not message.get('childrenIds')
]
if not pending:
return
first = pending[0]
parent_id = first.get('parentId')
model_id = first.get('model') or run['model_id']
batch = [
message
for message in pending
if message.get('parentId') == parent_id and (message.get('model') or model_id) == model_id
]
combined_content = '\n\n'.join(message.get('content', '') for message in batch if message.get('content'))
delegation_ids = [
message['meta']['delegation_id'] for message in batch if (message.get('meta') or {}).get('delegation_id')
]
subagent_chat_ids = [
message['meta']['subagent_chat_id']
for message in batch
if (message.get('meta') or {}).get('subagent_chat_id')
]
combined_meta = {'async_subagent_result': True}
if len(delegation_ids) == 1:
combined_meta['delegation_id'] = delegation_ids[0]
elif delegation_ids:
combined_meta['delegation_ids'] = delegation_ids
if len(subagent_chat_ids) == 1:
combined_meta['subagent_chat_id'] = subagent_chat_ids[0]
elif subagent_chat_ids:
combined_meta['subagent_chat_ids'] = subagent_chat_ids
reuse_message = len(batch) == 1 and not (first.get('meta') or {}).get('async_subagent_pending')
user_message_id = first['id'] if reuse_message else str(uuid4())
if not reuse_message:
removed_ids = {message['id'] for message in batch}
for message_id in removed_ids:
messages.pop(message_id, None)
if parent_id and parent_id in messages:
messages[parent_id]['childrenIds'] = [
child_id for child_id in messages[parent_id].get('childrenIds', []) if child_id not in removed_ids
]
history['messages'] = messages
history['currentId'] = parent_id
updated_chat = copy.deepcopy(chat.chat)
updated_chat['history'] = history
await Chats.update_chat_by_id(parent_chat_id, updated_chat)
await ChatMessages.delete_message_ids_by_chat_id(parent_chat_id, removed_ids)
assistant_message_id = str(uuid4())
message_list = get_message_list(messages, parent_id)
system_prompt = run.get('system_prompt')
user_message = {
'id': user_message_id,
'parentId': parent_id,
'childrenIds': [assistant_message_id],
'role': 'user',
'content': combined_content,
'model': model_id,
'meta': combined_meta,
'timestamp': int(time.time()),
}
assistant_message = {
'id': assistant_message_id,
'parentId': user_message_id,
'childrenIds': [],
'role': 'assistant',
'content': '',
'done': False,
'model': model_id,
'timestamp': int(time.time()),
}
if parent_id and parent_id in messages:
parent_children = [
child_id for child_id in messages[parent_id].get('childrenIds', []) if child_id != user_message_id
]
parent_children.append(user_message_id)
await Chats.upsert_message_to_chat_by_id_and_message_id(
parent_chat_id,
parent_id,
{'childrenIds': parent_children},
)
await Chats.upsert_message_to_chat_by_id_and_message_id(
parent_chat_id,
user_message_id,
user_message,
)
await Chats.upsert_message_to_chat_by_id_and_message_id(
parent_chat_id,
assistant_message_id,
assistant_message,
)
from open_webui.socket.main import sio
await sio.emit(
'events',
{
'chat_id': parent_chat_id,
'message_id': assistant_message_id,
'data': {'type': 'chat:reload'},
},
room=f'user:{user.id}',
)
form_data = {
'model': model_id,
'messages': [
*([{'role': 'system', 'content': system_prompt}] if system_prompt else []),
*message_list,
{'role': 'user', 'content': combined_content},
],
'stream': True,
'chat_id': parent_chat_id,
'id': assistant_message_id,
'parent_id': parent_id,
'user_message': user_message,
'session_id': run.get('session_id') or f'subagent-result:{parent_chat_id}',
'background_tasks': {},
'tool_ids': run.get('tool_ids') or [],
'skill_ids': run.get('skill_ids') or [],
'filter_ids': run.get('filter_ids') or [],
'features': run.get('features') or {},
'variables': run.get('variables') or {},
}
if run.get('terminal_id'):
form_data['terminal_id'] = run['terminal_id']
request = _build_request(source_request, user.id, internal=False)
await source_request.app.state.CHAT_COMPLETION_HANDLER(request, form_data, user=user)
async def delegate(
task: str,
context: str,
background: bool,
*,
request: Request,
user_data: dict,
metadata: dict,
parent_chat_id: str,
parent_message_id: str | None,
) -> str:
global _foreground_semaphore
task = task.strip()
if not task:
return 'Error: task must not be empty.'
if not parent_chat_id or not user_data.get('id'):
return 'Error: chat and user context are required.'
config = await Config.get_many(
'subagents.background_enabled',
'subagents.max_concurrent',
'subagents.max_async',
'subagents.max_iterations',
'subagents.max_output',
'subagents.system_prompt',
)
max_concurrent = int(config.get('subagents.max_concurrent') or 20)
max_async = int(config.get('subagents.max_async') or 20)
max_iterations = int(config.get('subagents.max_iterations') or 30)
max_output = int(config.get('subagents.max_output') or 30_000)
if max_concurrent != -1:
max_concurrent = max(1, max_concurrent)
if max_async != -1:
max_async = max(1, max_async)
if background and not config.get('subagents.background_enabled'):
return 'Error: background sub-agents are disabled in settings.'
features = copy.deepcopy(metadata.get('features') or {})
if (
background
and features.get('code_interpreter')
and await Config.get('code_interpreter.engine', 'pyodide') != 'jupyter'
):
features.pop('code_interpreter')
run = {
'model_id': metadata.get('model_id') or (metadata.get('model') or {}).get('id'),
'session_id': metadata.get('session_id'),
'tool_ids': copy.deepcopy(metadata.get('tool_ids') or []),
'skill_ids': copy.deepcopy(metadata.get('skill_ids') or []),
'system_prompt': metadata.get('system_prompt'),
'tool_servers': [] if background else copy.deepcopy(metadata.get('tool_servers') or []),
'filter_ids': copy.deepcopy(metadata.get('filter_ids') or []),
'terminal_id': metadata.get('terminal_id'),
'features': features,
'files': copy.deepcopy(metadata.get('files') or []),
'variables': copy.deepcopy(metadata.get('variables') or {}),
'direct': bool(metadata.get('direct')),
}
if not run.get('model_id'):
return 'Error: model context is required.'
if run.get('direct'):
return 'Error: sub-agents are unavailable for direct connections.'
delegation_id = f'deleg_{uuid4().hex[:8]}'
foreground_semaphore = None
if background:
async with _background_lock:
if max_async != -1 and len(_background_active) >= max_async:
return (
f'Error: Async subagent capacity reached ({max_async} running). '
'Wait for one to finish or increase subagents.max_async.'
)
_background_active.add(delegation_id)
elif max_concurrent != -1:
if _foreground_semaphore is None:
_foreground_semaphore = asyncio.Semaphore(max_concurrent)
foreground_semaphore = _foreground_semaphore
await foreground_semaphore.acquire()
mode = 'background' if background else 'foreground'
try:
user = UserModel(**user_data)
chat_id = str(uuid4())
user_message_id = str(uuid4())
assistant_message_id = str(uuid4())
prompt = f'{task}\n\n## Context\n{context}' if context else task
chat = await Chats.insert_new_chat(
chat_id,
user.id,
ChatForm(
chat={
'id': chat_id,
'title': f'Sub-agent: {task[:60]}',
'models': [run['model_id']],
'history': {
'currentId': assistant_message_id,
'messages': {
user_message_id: {
'id': user_message_id,
'parentId': None,
'childrenIds': [assistant_message_id],
'role': 'user',
'content': prompt,
'timestamp': int(time.time()),
'models': [run['model_id']],
},
assistant_message_id: {
'id': assistant_message_id,
'parentId': user_message_id,
'childrenIds': [],
'role': 'assistant',
'content': '',
'done': False,
'model': run['model_id'],
'timestamp': int(time.time()),
},
},
},
'messages': [{'role': 'user', 'content': prompt}],
}
),
internal_meta={
'internal': True,
'type': 'subagent',
'parent_chat_id': parent_chat_id,
'parent_message_id': parent_message_id,
'delegation_id': delegation_id,
'mode': mode,
},
)
if not chat:
raise RuntimeError('Failed to create sub-agent chat')
except Exception as exc:
if background:
async with _background_lock:
_background_active.discard(delegation_id)
elif foreground_semaphore:
foreground_semaphore.release()
prefix = 'background ' if background else ''
return f'Error: failed to create {prefix}sub-agent: {exc}'
async def run_reserved() -> dict:
try:
child_request = _build_request(request, user.id, internal=True)
child_request.state.max_tool_call_iterations = max_iterations
parent_system_prompt = run.get('system_prompt') or ''
subagent_system_prompt = (
str(config.get('subagents.system_prompt') or '').strip() or DEFAULT_SUBAGENT_SYSTEM_PROMPT
)
form_data = {
'model': run['model_id'],
'messages': [
{
'role': 'system',
'content': (
f'{parent_system_prompt}\n\n{subagent_system_prompt}'
if parent_system_prompt
else subagent_system_prompt
),
},
{'role': 'user', 'content': prompt},
],
'stream': True,
'chat_id': chat_id,
'id': assistant_message_id,
'parent_id': None,
'user_message': {
'id': user_message_id,
'parentId': None,
'role': 'user',
'content': prompt,
},
'session_id': run.get('session_id') or f'subagent:{chat_id}',
'background_tasks': {},
'tool_ids': run.get('tool_ids') or [],
'skill_ids': run.get('skill_ids') or [],
'filter_ids': run.get('filter_ids') or [],
'features': run.get('features') or {},
'files': run.get('files') or [],
'variables': run.get('variables') or {},
}
if run.get('terminal_id'):
form_data['terminal_id'] = run['terminal_id']
if run.get('tool_servers'):
form_data['tool_servers'] = run['tool_servers']
await request.app.state.CHAT_COMPLETION_HANDLER(child_request, form_data, user=user)
message = await Chats.get_message_by_id_and_message_id(chat_id, assistant_message_id)
if not message:
return {
'status': 'error',
'summary': '',
'error': 'Sub-agent chat or completion message no longer exists.',
}
summary = message.get('content') or ''
if isinstance(summary, list):
summary = ''.join(
str(item.get('text', ''))
for item in summary
if isinstance(item, dict) and item.get('type') == 'text'
)
if not summary:
summary = ''.join(
str(part.get('text', ''))
for item in message.get('output') or []
if item.get('type') == 'message'
for part in item.get('content') or []
if part.get('type') == 'output_text'
)
if len(summary) > max_output:
summary = f'{summary[:max_output]}\n\n[output truncated]'
error = message.get('error')
return {
'status': 'error' if error else 'completed',
'summary': summary or ('Sub-agent produced no output.' if not error else ''),
'error': error,
}
except asyncio.CancelledError:
await Chats.upsert_message_to_chat_by_id_and_message_id(
chat_id,
assistant_message_id,
{'done': True, 'error': {'content': 'Sub-agent cancelled.'}},
)
raise
except Exception as exc:
await Chats.upsert_message_to_chat_by_id_and_message_id(
chat_id,
assistant_message_id,
{'done': True, 'error': {'content': str(exc)}},
)
raise
finally:
if background:
async with _background_lock:
_background_active.discard(delegation_id)
elif foreground_semaphore:
foreground_semaphore.release()
async def run_background() -> dict:
started_at = time.time()
cancelled = False
try:
result = await run_reserved()
except asyncio.CancelledError:
result = {'status': 'interrupted', 'summary': '', 'error': 'cancelled'}
cancelled = True
except Exception as exc:
result = {'status': 'error', 'summary': '', 'error': str(exc)}
parent = await Chats.get_chat_by_id_and_user_id(parent_chat_id, user.id)
if not parent:
if cancelled:
raise asyncio.CancelledError
return result
history = copy.deepcopy(parent.chat.get('history') or {})
messages = history.setdefault('messages', {})
done_assistants = [
message
for message in messages.values()
if message.get('role') == 'assistant' and message.get('done') is not False
]
result_parent_id = (
max(done_assistants, key=lambda message: message.get('timestamp', 0)).get('id')
if done_assistants
else parent_message_id
)
duration = f'{time.time() - started_at:.1f}s'
lines = [
f'[ASYNC SUBAGENT COMPLETE - {delegation_id}]',
(
'A background subagent you dispatched earlier has finished. '
'The original task source is included so you can decide whether '
'to use the result or continue without it.'
),
'',
f'Original task: {task}',
]
if context:
lines.append(f'Context provided: {context}')
lines.extend(
[
f'Subagent chat: {chat_id}',
f'Status: {result.get("status", "completed")} Duration: {duration}',
'--- RESULT ---',
]
)
if result.get('status') == 'completed':
lines.append(result.get('summary') or 'Subagent completed without a final summary.')
elif result.get('status') == 'interrupted':
lines.append('The subagent was interrupted before completing.')
if result.get('summary'):
lines.extend(['Partial output:', result['summary']])
else:
detail = f' {result.get("error")}' if result.get('error') else ''
lines.append(f'The subagent did not complete successfully.{detail}')
if result.get('summary'):
lines.extend(['Partial output:', result['summary']])
pending_message_id = str(uuid4())
pending_meta = {
'async_subagent_result': True,
'delegation_id': delegation_id,
'subagent_chat_id': chat_id,
}
pending_message = {
'id': pending_message_id,
'parentId': result_parent_id,
'childrenIds': [],
'role': 'user',
'content': '\n'.join(lines),
'model': run['model_id'],
'meta': pending_meta,
'timestamp': int(time.time()),
}
lock = _parent_locks.setdefault(parent_chat_id, asyncio.Lock())
async with lock:
parent = await Chats.get_chat_by_id_and_user_id(parent_chat_id, user.id)
if not parent:
if cancelled:
raise asyncio.CancelledError
return result
if await has_active_tasks(request.app.state.redis, parent_chat_id):
pending_message['meta']['async_subagent_pending'] = True
updated_chat = copy.deepcopy(parent.chat)
updated_history = updated_chat.setdefault('history', {})
updated_messages = updated_history.setdefault('messages', {})
updated_messages[pending_message_id] = pending_message
if result_parent_id and result_parent_id in updated_messages:
children = updated_messages[result_parent_id].setdefault('childrenIds', [])
if pending_message_id not in children:
children.append(pending_message_id)
await Chats.update_chat_by_id(parent_chat_id, updated_chat)
await ChatMessages.upsert_message(
message_id=pending_message_id,
chat_id=parent_chat_id,
user_id=user.id,
data=pending_message,
)
if pending_message['meta'].get('async_subagent_pending') is True:
from open_webui.socket.main import sio
await sio.emit(
'events',
{
'chat_id': parent_chat_id,
'message_id': pending_message_id,
'data': {'type': 'chat:reload'},
},
room=f'user:{user.id}',
)
if not await has_active_tasks(request.app.state.redis, parent_chat_id):
await process_pending_subagent_results(request, parent_chat_id, user.id, run)
if cancelled:
raise asyncio.CancelledError
return result
try:
_, child_task = await create_task(
request.app.state.redis,
run_background() if background else run_reserved(),
id=chat_id,
)
except Exception as exc:
if background:
async with _background_lock:
_background_active.discard(delegation_id)
elif foreground_semaphore:
foreground_semaphore.release()
return f'Error: {exc}'
if background:
return json.dumps(
{
'status': 'dispatched',
'delegation_id': delegation_id,
'subagent_chat_id': chat_id,
'mode': 'background',
'task': task,
},
ensure_ascii=False,
)
try:
result = await child_task
except asyncio.CancelledError:
if asyncio.current_task() and asyncio.current_task().cancelling():
raise
return 'Error: sub-agent was cancelled.'
except Exception as exc:
return f'Error: {exc}'
if result.get('status') != 'completed':
return f'Error: {result.get("error") or "sub-agent failed."}'
return result.get('summary') or 'Sub-agent produced no output.'
backend/open_webui/utils/tools.py
+33 -16
View File
@@ -51,6 +51,7 @@ from open_webui.tools.builtin import (
create_automation,
create_calendar_event,
create_tasks,
delegate_task,
delete_automation,
delete_calendar_event,
delete_memory,
@@ -98,6 +99,7 @@ from open_webui.utils.access_control import has_access, has_connection_access, h
from open_webui.utils.headers import get_custom_headers, include_user_info_headers
from open_webui.utils.misc import is_string_allowed
from open_webui.utils.plugin import get_tool_contents_cache, get_tools_cache, load_tool_module_by_id
from open_webui.utils.terminals import get_terminal_server_url
from pydantic import BaseModel, Field, create_model
from pydantic.fields import FieldInfo
@@ -499,6 +501,8 @@ async def get_builtin_tools(
'channels.enable',
'automations.enable',
'calendar.enable',
'subagents.enable',
'subagents.background_enabled',
)
async def has_user_permission(feature_key: str) -> bool:
@@ -563,6 +567,14 @@ async def get_builtin_tools(
if is_builtin_tool_enabled('chats'):
builtin_functions.extend([search_chats, view_chat])
if (
is_builtin_tool_enabled('subagents')
and config.get('subagents.enable')
and getattr(request.state, 'internal', False) is not True
and getattr(request.state, 'direct', False) is not True
):
builtin_functions.append(delegate_task)
# Add memory tools when memory is enabled and the model allows this builtin category.
if (
is_builtin_tool_enabled('memory')
@@ -660,6 +672,11 @@ async def get_builtin_tools(
[search_calendar_events, create_calendar_event, update_calendar_event, delete_calendar_event]
)
if getattr(request.state, 'internal', False) is True:
from open_webui.utils.subagents import MUTATING_MEMORY_TOOLS
builtin_functions = [func for func in builtin_functions if func.__name__ not in MUTATING_MEMORY_TOOLS]
for func in builtin_functions:
callable = await get_async_tool_function_and_apply_extra_params(
func,
@@ -679,6 +696,11 @@ async def get_builtin_tools(
pydantic_model = convert_function_to_pydantic_model(func)
spec = convert_pydantic_model_to_openai_function_spec(pydantic_model)
spec = clean_openai_tool_schema(spec)
if func.__name__ == 'delegate_task' and not config.get('subagents.background_enabled'):
parameters = spec.get('parameters', {})
parameters.get('properties', {}).pop('background', None)
if isinstance(parameters.get('required'), list):
parameters['required'] = [name for name in parameters['required'] if name != 'background']
tools_dict[func.__name__] = {
'tool_id': f'builtin:{func.__name__}',
@@ -1075,7 +1097,9 @@ async def get_terminal_system_prompt(
trust_env=True,
) as session:
# 1. Check feature flag
async with session.get(f'{base}/api/config', ssl=AIOHTTP_CLIENT_SESSION_SSL) as resp:
async with session.get(
f'{base}/api/config', headers=headers, cookies=cookies or {}, ssl=AIOHTTP_CLIENT_SESSION_SSL
) as resp:
if resp.status != 200:
return None
config = await resp.json()
@@ -1107,13 +1131,7 @@ async def set_terminal_servers(request: Request):
enabled = connection.get('enabled', True)
base_url = connection.get('url', '').rstrip('/')
policy_id = connection.get('policy_id', '')
# Orchestrator connections route through /p/{policy_id}/ — the
# OpenAPI spec lives on the proxied terminal, not the orchestrator.
if connection.get('server_type') == 'orchestrator' and policy_id:
base_url = f'{base_url}/p/{policy_id}'
base_url = get_terminal_server_url(connection)
server_configs.append(
{
@@ -1143,6 +1161,8 @@ async def set_terminal_servers(request: Request):
headers = {}
if connection.get('auth_type', 'bearer') == 'bearer':
headers.update(bearer_auth_header(connection.get('key', '')))
if connection.get('policy_id'):
headers['X-User-Id'] = 'system'
prompt = await get_terminal_system_prompt(server['url'], headers)
if prompt:
server['system_prompt'] = prompt
@@ -1192,24 +1212,21 @@ async def get_terminal_tools(
connections = await Config.get('terminal_server.connections', []) or []
connection = next((c for c in connections if c.get('id') == terminal_id), None)
if connection is None:
log.warning(f'Terminal server not found: {terminal_id}')
return {}
raise RuntimeError(f"Terminal server '{terminal_id}' not found")
user_group_ids = {group.id for group in await Groups.get_groups_by_member_id(user.id)}
if not await has_connection_access(user, connection, user_group_ids):
log.warning(f'Access denied to terminal {terminal_id} for user {user.id}')
return {}
raise RuntimeError(f'Access denied to terminal {terminal_id}')
# Find the cached spec data for this terminal
terminal_servers = await get_terminal_servers(request)
server_data = next((s for s in terminal_servers if s.get('id') == terminal_id), None)
if server_data is None:
log.warning(f'Terminal server spec not found for {terminal_id}')
return {}
raise RuntimeError(f"Terminal server '{terminal_id}' is unavailable")
specs = server_data.get('specs', [])
if not specs:
return {}
raise RuntimeError(f"Terminal server '{terminal_id}' has no available tools")
# Build auth headers
auth_type = connection.get('auth_type', 'bearer')
@@ -1236,7 +1253,7 @@ async def get_terminal_tools(
if session_id:
headers['X-Session-Id'] = session_id
terminal_cwd = await get_terminal_cwd(connection.get('url', ''), headers, cookies)
terminal_cwd = await get_terminal_cwd(server_data['url'], headers, cookies)
tools_dict = {}
for spec in specs: