diff --git a/backend/open_webui/config.py b/backend/open_webui/config.py index edd989857a..1b842940d9 100644 --- a/backend/open_webui/config.py +++ b/backend/open_webui/config.py @@ -1991,6 +1991,8 @@ ENABLE_SUBAGENTS = os.getenv('ENABLE_SUBAGENTS', 'False').lower() == 'true' SUBAGENTS_BACKGROUND_ENABLED = os.getenv('SUBAGENTS_BACKGROUND_ENABLED', 'False').lower() == 'true' SUBAGENTS_MAX_CONCURRENT = int(os.getenv('SUBAGENTS_MAX_CONCURRENT', '20')) SUBAGENTS_MAX_ASYNC = int(os.getenv('SUBAGENTS_MAX_ASYNC', '20')) +SUBAGENTS_MAX_ITERATIONS = int(os.getenv('SUBAGENTS_MAX_ITERATIONS', '30')) +SUBAGENTS_MAX_OUTPUT = int(os.getenv('SUBAGENTS_MAX_OUTPUT', '30000')) SUBAGENTS_SYSTEM_PROMPT = os.getenv('SUBAGENTS_SYSTEM_PROMPT', '') AUTOMATION_MAX_COUNT = os.getenv('AUTOMATION_MAX_COUNT', '') @@ -3029,6 +3031,8 @@ DEFAULT_CONFIG = { 'subagents.background_enabled': SUBAGENTS_BACKGROUND_ENABLED, 'subagents.max_concurrent': SUBAGENTS_MAX_CONCURRENT, 'subagents.max_async': SUBAGENTS_MAX_ASYNC, + 'subagents.max_iterations': SUBAGENTS_MAX_ITERATIONS, + 'subagents.max_output': SUBAGENTS_MAX_OUTPUT, 'subagents.system_prompt': SUBAGENTS_SYSTEM_PROMPT, 'automations.max_count': AUTOMATION_MAX_COUNT, 'automations.min_interval': AUTOMATION_MIN_INTERVAL, diff --git a/backend/open_webui/main.py b/backend/open_webui/main.py index 0136cf9a8f..754bcc8a5d 100644 --- a/backend/open_webui/main.py +++ b/backend/open_webui/main.py @@ -1131,6 +1131,7 @@ async def chat_completion( metadata = { 'user_id': user.id, 'user_agent': request.headers.get('user-agent', '') or '', + 'internal': getattr(request.state, 'internal', False) is True, 'chat_id': form_data.pop('chat_id', None) or '', 'user_message': user_message, 'user_message_id': user_message.get('id') if user_message else None, @@ -1590,9 +1591,39 @@ async def chat_completion( except Exception: pass + try: + chat_id = metadata.get('chat_id') + if ( + chat_id + and getattr(request.state, 'internal', False) is not True + and not await has_active_tasks(request.app.state.redis, chat_id) + ): + from open_webui.utils.subagents import process_pending_subagent_results + + await process_pending_subagent_results( + request, + chat_id, + user.id, + { + 'model_id': metadata.get('model_id') or form_data.get('model'), + 'session_id': metadata.get('session_id'), + 'tool_ids': metadata.get('tool_ids') or [], + 'skill_ids': metadata.get('skill_ids') or [], + 'system_prompt': metadata.get('system_prompt'), + 'filter_ids': metadata.get('filter_ids') or [], + 'terminal_id': metadata.get('terminal_id'), + 'features': metadata.get('features') or {}, + 'variables': metadata.get('variables') or {}, + }, + ) + except Exception: + log.exception('Failed to process pending sub-agent results for chat %s', metadata.get('chat_id')) + # Fan out: one task per model if metadata.get('session_id') and metadata.get('chat_id'): task_ids = [] + subagent_results = [] + is_internal = getattr(request.state, 'internal', False) is True chat_id = metadata['chat_id'] for idx, entry in enumerate(message_ids): @@ -1619,28 +1650,39 @@ async def chat_completion( # Only the first model runs chat-level background tasks; # subsequent models only run follow-ups. + process = process_chat( + request, + model_form_data, + user, + per_model_metadata, + resolved_model, + tasks + if idx == 0 + else { + k: v for k, v in (tasks or {}).items() if k not in (TASKS.TITLE_GENERATION, TASKS.TAGS_GENERATION) + } + or None, + ) + if is_internal: + subagent_results.append(await process) + continue + task_id, _ = await create_task( request.app.state.redis, - process_chat( - request, - model_form_data, - user, - per_model_metadata, - resolved_model, - tasks - if idx == 0 - else { - k: v - for k, v in (tasks or {}).items() - if k not in (TASKS.TITLE_GENERATION, TASKS.TAGS_GENERATION) - } - or None, - ), + process, id=chat_id, ) per_model_metadata['task_id'] = task_id task_ids.append(task_id) + if is_internal: + return { + 'status': True, + 'task_ids': [], + 'chat_id': chat_id, + 'results': subagent_results, + } + # Emit chat:active=true if task_ids: event_emitter = await get_event_emitter( diff --git a/backend/open_webui/models/chats.py b/backend/open_webui/models/chats.py index 52981d7e3f..ce7480807f 100644 --- a/backend/open_webui/models/chats.py +++ b/backend/open_webui/models/chats.py @@ -74,6 +74,10 @@ class Chat(Base): # database table mapping for chat entity ) +def is_internal_chat(meta: dict | None) -> bool: + return bool(meta and meta.get('internal') is True) + + class ChatModel(BaseModel): model_config = ConfigDict(from_attributes=True) # allows ORM model binding id: str @@ -349,7 +353,13 @@ class ChatTable: return True async def insert_new_chat( - self, id: str, user_id: str, form_data: ChatForm, db: AsyncSession | None = None + self, + id: str, + user_id: str, + form_data: ChatForm, + db: AsyncSession | None = None, + *, + internal_meta: dict | None = None, ) -> ChatModel | None: async with get_async_db_context(db) as session: chat = ChatModel( @@ -361,6 +371,7 @@ class ChatTable: ), 'chat': self._clean_null_bytes(form_data.chat), 'folder_id': form_data.folder_id, + 'meta': internal_meta or {}, 'created_at': int(time.time()), 'updated_at': int(time.time()), 'last_read_at': int(time.time()), @@ -389,6 +400,17 @@ class ChatTable: return ChatModel.model_validate(chat_item) if chat_item else None + async def get_internal_chat_ids_by_parent_id(self, parent_chat_id: str, user_id: str) -> list[str]: + async with get_async_db_context() as session: + result = await session.execute( + select(Chat.id).where( + Chat.user_id == user_id, + Chat.meta['internal'].as_boolean().is_(True), + Chat.meta['parent_chat_id'].as_string() == parent_chat_id, + ) + ) + return list(result.scalars().all()) + def _chat_import_form_to_chat_model(self, user_id: str, form_data: ChatImportForm) -> ChatModel: id = str(uuid.uuid4()) chat = ChatModel( @@ -951,6 +973,7 @@ class ChatTable: stmt = select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at).filter_by( user_id=user_id, archived=True ) + stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True)) if filter: query_key = filter.get('query') @@ -998,7 +1021,8 @@ class ChatTable: db: AsyncSession | None = None, ) -> int: async with get_async_db_context(db) as session: - result = await session.execute(select(func.count(Chat.id)).filter_by(user_id=user_id, archived=True)) + stmt = select(func.count(Chat.id)).filter_by(user_id=user_id, archived=True) + result = await session.execute(stmt.where(Chat.meta['internal'].as_boolean().is_not(True))) return result.scalar() or 0 async def get_shared_chat_list_by_user_id( @@ -1027,6 +1051,7 @@ class ChatTable: stmt = select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at, Chat.last_read_at).filter_by( user_id=user_id ) + stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True)) if not include_archived: stmt = stmt.filter_by(archived=False) @@ -1082,6 +1107,7 @@ class ChatTable: stmt = select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at, Chat.last_read_at).filter_by( user_id=user_id ) + stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True)) if not include_folders: stmt = stmt.filter_by(folder_id=None) @@ -1123,9 +1149,9 @@ class ChatTable: db: AsyncSession | None = None, ) -> list[ChatModel]: async with get_async_db_context(db) as session: - result = await session.execute( - select(Chat).filter(Chat.id.in_(chat_ids)).filter_by(archived=False).order_by(Chat.updated_at.desc()) - ) + stmt = select(Chat).filter(Chat.id.in_(chat_ids)).filter_by(archived=False) + stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True)) + result = await session.execute(stmt.order_by(Chat.updated_at.desc())) all_chats = result.scalars().all() return [ChatModel.model_validate(chat) for chat in all_chats] @@ -1170,6 +1196,7 @@ class ChatTable: select(Chat.id, Chat.user_id, Chat.title, Chat.updated_at, User.name.label('user_name')) .join(chat_ids, chat_ids.c.chat_id == Chat.id) .outerjoin(User, User.id == Chat.user_id) + .where(Chat.meta['internal'].as_boolean().is_not(True)) ) order_by = filter.get('order_by') if filter else None @@ -1301,7 +1328,8 @@ class ChatTable: async def get_chats(self, skip: int = 0, limit: int = 50, db: AsyncSession | None = None) -> list[ChatModel]: async with get_async_db_context(db) as session: - result = await session.execute(select(Chat).order_by(Chat.updated_at.desc())) + stmt = select(Chat).where(Chat.meta['internal'].as_boolean().is_not(True)) + result = await session.execute(stmt.order_by(Chat.updated_at.desc())) all_chats = result.scalars().all() return [ChatModel.model_validate(chat) for chat in all_chats] @@ -1316,6 +1344,7 @@ class ChatTable: ) -> ChatListResponse: async with get_async_db_context(db) as session: stmt = select(Chat).filter_by(user_id=user_id) + stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True)) if filter: if filter.get('updated_at'): @@ -1359,11 +1388,11 @@ class ChatTable: self, user_id: str, db: AsyncSession | None = None ) -> list[ChatTitleIdResponse]: async with get_async_db_context(db) as session: - result = await session.execute( - select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at, Chat.last_read_at) - .filter_by(user_id=user_id, pinned=True, archived=False) - .order_by(Chat.updated_at.desc()) + stmt = select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at, Chat.last_read_at).filter_by( + user_id=user_id, pinned=True, archived=False ) + stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True)) + result = await session.execute(stmt.order_by(Chat.updated_at.desc())) all_chats = result.all() return [ ChatTitleIdResponse.model_validate( @@ -1380,9 +1409,9 @@ class ChatTable: async def get_archived_chats_by_user_id(self, user_id: str, db: AsyncSession | None = None) -> list[ChatModel]: async with get_async_db_context(db) as session: - result = await session.execute( - select(Chat).filter_by(user_id=user_id, archived=True).order_by(Chat.updated_at.desc()) - ) + stmt = select(Chat).filter_by(user_id=user_id, archived=True) + stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True)) + result = await session.execute(stmt.order_by(Chat.updated_at.desc())) return [ChatModel.model_validate(chat) for chat in result.scalars().all()] # search user conversations @@ -1453,6 +1482,7 @@ class ChatTable: async with get_async_db_context(db) as session: stmt = select(Chat).filter(Chat.user_id == user_id) + stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True)) if is_archived is not None: stmt = stmt.filter(Chat.archived == is_archived) @@ -1594,6 +1624,7 @@ class ChatTable: .filter_by(folder_id=folder_id, user_id=user_id) .filter(or_(Chat.pinned == False, Chat.pinned == None)) .filter_by(archived=False) + .where(Chat.meta['internal'].as_boolean().is_not(True)) .order_by(Chat.updated_at.desc(), Chat.id) ) @@ -1631,6 +1662,7 @@ class ChatTable: .filter_by(folder_id=folder_id) .filter(or_(Chat.pinned == False, Chat.pinned == None)) .filter_by(archived=False) + .where(Chat.meta['internal'].as_boolean().is_not(True)) .order_by(Chat.updated_at.desc(), Chat.id) ) @@ -1662,6 +1694,7 @@ class ChatTable: .filter(Chat.folder_id.in_(folder_ids), Chat.user_id == user_id) .filter(or_(Chat.pinned == False, Chat.pinned == None)) .filter_by(archived=False) + .where(Chat.meta['internal'].as_boolean().is_not(True)) .order_by(Chat.updated_at.desc()) ) @@ -1707,6 +1740,7 @@ class ChatTable: stmt = select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at, Chat.last_read_at).filter_by( user_id=user_id ) + stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True)) tag_id = tag_name.replace(' ', '_').lower() bind = await session.connection() @@ -1769,6 +1803,7 @@ class ChatTable: ) -> int: async with get_async_db_context(db) as session: stmt = select(func.count(Chat.id)).filter_by(user_id=user_id, archived=False) + stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True)) tag_id = tag_name.replace(' ', '_').lower() bind = await session.connection() @@ -1816,7 +1851,8 @@ class ChatTable: self, folder_id: str, user_id: str, db: AsyncSession | None = None ) -> int: async with get_async_db_context(db) as session: - result = await session.execute(select(func.count(Chat.id)).filter_by(user_id=user_id, folder_id=folder_id)) + stmt = select(func.count(Chat.id)).filter_by(user_id=user_id, folder_id=folder_id) + result = await session.execute(stmt.where(Chat.meta['internal'].as_boolean().is_not(True))) count = result.scalar() log.info(f"Count of chats for folder '{folder_id}': {count}") @@ -1829,9 +1865,8 @@ class ChatTable: return 0 async with get_async_db_context(db) as session: - result = await session.execute( - select(func.count(Chat.id)).filter(Chat.user_id == user_id, Chat.folder_id.in_(folder_ids)) - ) + stmt = select(func.count(Chat.id)).filter(Chat.user_id == user_id, Chat.folder_id.in_(folder_ids)) + result = await session.execute(stmt.where(Chat.meta['internal'].as_boolean().is_not(True))) count = result.scalar() log.info(f"Count of chats for folders '{folder_ids}': {count}") diff --git a/backend/open_webui/routers/chats.py b/backend/open_webui/routers/chats.py index 7f80ca1b6a..e3aa0ec659 100644 --- a/backend/open_webui/routers/chats.py +++ b/backend/open_webui/routers/chats.py @@ -26,6 +26,7 @@ from open_webui.models.chats import ( ChatStatsExport, ChatTitleIdResponse, ChatUsageStatsListResponse, + is_internal_chat, MessageStats, ) from open_webui.models.folders import Folders @@ -1178,8 +1179,10 @@ async def get_chat_by_id(id: str, user=Depends(get_verified_user), db: AsyncSess if not chat: # Check if user has access via access grants (shared_chat grants) - if user.role == 'admin' and ENABLE_ADMIN_CHAT_ACCESS: - chat = await Chats.get_chat_by_id(id, db=db) + if user.role == 'admin': + candidate = await Chats.get_chat_by_id(id, db=db) + if ENABLE_ADMIN_CHAT_ACCESS or (candidate and is_internal_chat(candidate.meta)): + chat = candidate else: has_grant = await AccessGrants.has_access( user_id=user.id, @@ -1430,6 +1433,13 @@ async def delete_chat_by_id( # before deleting the chat to prevent orphaned requests. await stop_item_tasks(request.app.state.redis, id) + async def delete_internal_children(owner_id: str) -> None: + child_ids = await Chats.get_internal_chat_ids_by_parent_id(id, owner_id) + for child_id in child_ids: + await stop_item_tasks(request.app.state.redis, child_id) + await Chats.delete_chat_by_id_and_user_id(child_id, owner_id) + await stop_item_tasks(request.app.state.redis, id) + if user.role == 'admin': chat = await Chats.get_chat_by_id(id, db=db) if not chat: @@ -1438,6 +1448,7 @@ async def delete_chat_by_id( detail=ERROR_MESSAGES.NOT_FOUND, ) await Chats.delete_orphan_tags_for_user(chat.meta.get('tags', []), user.id, threshold=1, db=db) + await delete_internal_children(chat.user_id) result = await Chats.delete_chat_by_id(id, db=db) @@ -1464,6 +1475,7 @@ async def delete_chat_by_id( detail=ERROR_MESSAGES.NOT_FOUND, ) await Chats.delete_orphan_tags_for_user(chat.meta.get('tags', []), user.id, threshold=1, db=db) + await delete_internal_children(user.id) result = await Chats.delete_chat_by_id_and_user_id(id, user.id, db=db) if result: diff --git a/backend/open_webui/routers/configs.py b/backend/open_webui/routers/configs.py index dde2015e3c..d93139396e 100644 --- a/backend/open_webui/routers/configs.py +++ b/backend/open_webui/routers/configs.py @@ -72,6 +72,8 @@ SUBAGENTS_CONFIG_KEYS = { 'SUBAGENTS_BACKGROUND_ENABLED': 'subagents.background_enabled', 'SUBAGENTS_MAX_CONCURRENT': 'subagents.max_concurrent', 'SUBAGENTS_MAX_ASYNC': 'subagents.max_async', + 'SUBAGENTS_MAX_ITERATIONS': 'subagents.max_iterations', + 'SUBAGENTS_MAX_OUTPUT': 'subagents.max_output', 'SUBAGENTS_SYSTEM_PROMPT': 'subagents.system_prompt', } @@ -767,6 +769,8 @@ class SubagentsConfigForm(BaseModel): SUBAGENTS_BACKGROUND_ENABLED: bool SUBAGENTS_MAX_CONCURRENT: int SUBAGENTS_MAX_ASYNC: int + SUBAGENTS_MAX_ITERATIONS: int + SUBAGENTS_MAX_OUTPUT: int SUBAGENTS_SYSTEM_PROMPT: str diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index 1cdd064b3a..3393b73bb2 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -925,12 +925,17 @@ async def get_event_emitter(request_info, update_db=True): user_id = request_info['user_id'] chat_id = request_info['chat_id'] message_id = request_info['message_id'] + internal = request_info.get('internal') is True + + if internal and event_data.get('type') == 'notification': + return await sio.emit( 'events', { 'chat_id': chat_id, 'message_id': message_id, + **({'internal': True} if internal else {}), 'data': event_data, }, room=f'user:{user_id}', diff --git a/backend/open_webui/tools/builtin.py b/backend/open_webui/tools/builtin.py index f6631691d3..c59ea54aaa 100644 --- a/backend/open_webui/tools/builtin.py +++ b/backend/open_webui/tools/builtin.py @@ -1348,6 +1348,48 @@ async def view_chat( return json.dumps({'error': str(e)}) +# ============================================================================= +# SUB-AGENT TOOL +# ============================================================================= + + +async def delegate_task( + task: str, + context: str = '', + background: bool = False, + __request__: Request = None, + __user__: dict = None, + __metadata__: dict = None, + __chat_id__: str = None, + __message_id__: str = None, +) -> str: + """ + Delegate focused work to a parallel sub-agent using the current model and tools. + + :param task: The specific task for the sub-agent to complete + :param context: Relevant context, decisions, or file paths for the task + :param background: Return immediately and continue this chat when the sub-agent finishes + :return: Foreground result text, or a JSON dispatch handle for background work + """ + if __request__ is None: + return 'Error: request context not available.' + if getattr(__request__.state, 'internal', False) is True: + return 'Error: sub-agents cannot delegate recursively.' + + from open_webui.utils.subagents import delegate + + return await delegate( + task, + context, + background, + request=__request__, + user_data=__user__ or {}, + metadata=__metadata__ or {}, + parent_chat_id=__chat_id__ or '', + parent_message_id=__message_id__, + ) + + # ============================================================================= # CHANNELS TOOLS # ============================================================================= diff --git a/backend/open_webui/utils/middleware.py b/backend/open_webui/utils/middleware.py index d72a4a63a8..cdc33db7ad 100644 --- a/backend/open_webui/utils/middleware.py +++ b/backend/open_webui/utils/middleware.py @@ -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') diff --git a/backend/open_webui/utils/subagents.py b/backend/open_webui/utils/subagents.py new file mode 100644 index 0000000000..512713373d --- /dev/null +++ b/backend/open_webui/utils/subagents.py @@ -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.' diff --git a/backend/open_webui/utils/tools.py b/backend/open_webui/utils/tools.py index 0e468ad313..93f6d11ffc 100644 --- a/backend/open_webui/utils/tools.py +++ b/backend/open_webui/utils/tools.py @@ -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: diff --git a/src/lib/components/admin/Settings.svelte b/src/lib/components/admin/Settings.svelte index 2d75066031..370b4f005e 100644 --- a/src/lib/components/admin/Settings.svelte +++ b/src/lib/components/admin/Settings.svelte @@ -22,10 +22,12 @@ import Evaluations from './Settings/Evaluations.svelte'; import CodeExecution from './Settings/CodeExecution.svelte'; import Integrations from './Settings/Integrations.svelte'; + import Subagents from './Settings/Subagents.svelte'; import ChartBar from '../icons/ChartBar.svelte'; import DocumentChartBar from '../icons/DocumentChartBar.svelte'; import Search from '../icons/Search.svelte'; + import User from '../icons/User.svelte'; import XMark from '../icons/XMark.svelte'; const i18n = getContext('i18n'); @@ -41,6 +43,7 @@ 'authentication', 'connections', 'models', + 'subagents', 'evaluations', 'integrations', 'documents', @@ -145,6 +148,12 @@ 'export' ] }, + { + id: 'subagents', + title: 'Sub-agents', + route: '/admin/settings/subagents', + keywords: ['sub-agents', 'subagents', 'delegation', 'background', 'agents'] + }, { id: 'evaluations', title: 'Evaluations', @@ -345,6 +354,7 @@ + @@ -431,6 +441,8 @@ clip-rule="evenodd" /> + {:else if tab.id === 'subagents'} + {:else if tab.id === 'documents'} {:else if selectedTab === 'models'} + {:else if selectedTab === 'subagents'} + {:else if selectedTab === 'evaluations'} {:else if selectedTab === 'integrations'} diff --git a/src/lib/components/admin/Settings/Subagents.svelte b/src/lib/components/admin/Settings/Subagents.svelte new file mode 100644 index 0000000000..aa9c212fbc --- /dev/null +++ b/src/lib/components/admin/Settings/Subagents.svelte @@ -0,0 +1,200 @@ + + +
+
+
{$i18n.t('Sub-agents')}
+ + {#if loading} +
+ {:else} +
+ +

+ {$i18n.t( + 'Allow the AI to delegate tasks to sub-agents. Each sub-agent creates a real chat with full tool access. Uses additional LLM calls.' + )} +

+ + {#if enabled} +
+ +
+ + + {$i18n.t('simultaneous sub-agents')} + +
+
+ +
+ +

+ {$i18n.t( + 'Allow delegated sub-agents to keep running while the parent chat continues.' + )} +

+
+ + {#if backgroundEnabled} +
+ +
+ + + {$i18n.t('background sub-agents')} + +
+
+ {/if} + +
+ +
+ + + {$i18n.t('tool loops per sub-agent')} + +
+
+ +
+ +
+ + chars +
+
+ +
+ + +

+ {$i18n.t('Leave empty for the built-in default.')} +

+
+ {/if} +
+ {/if} +
+ + {#if !loading} +
+ +
+ {/if} +
diff --git a/src/lib/components/chat/Messages/SubagentResultRow.svelte b/src/lib/components/chat/Messages/SubagentResultRow.svelte new file mode 100644 index 0000000000..83b95c3662 --- /dev/null +++ b/src/lib/components/chat/Messages/SubagentResultRow.svelte @@ -0,0 +1,67 @@ + + +
+ + {#if expanded} +
+ {content} +
+ {/if} +
diff --git a/src/lib/components/chat/Messages/UserMessage.svelte b/src/lib/components/chat/Messages/UserMessage.svelte index 8d540828e7..30e2d48b32 100644 --- a/src/lib/components/chat/Messages/UserMessage.svelte +++ b/src/lib/components/chat/Messages/UserMessage.svelte @@ -16,11 +16,19 @@ import Markdown from './Markdown.svelte'; import Image from '$lib/components/common/Image.svelte'; import DeleteConfirmDialog from '$lib/components/common/ConfirmDialog.svelte'; + import SubagentResultRow from './SubagentResultRow.svelte'; import localizedFormat from 'dayjs/plugin/localizedFormat'; const i18n = getContext('i18n'); dayjs.extend(localizedFormat); + type SubagentResult = { + async_subagent_result: true; + delegation_id?: string; + delegation_ids?: string[]; + subagent_chat_id?: string; + subagent_chat_ids?: string[]; + }; export let user; @@ -54,6 +62,7 @@ let editScrollContainer: HTMLDivElement; let message = structuredClone(history.messages[messageId]); + let subagentResult: SubagentResult | undefined; $: if (history.messages) { const source = history.messages[messageId]; if (source) { @@ -64,6 +73,7 @@ } } } + $: subagentResult = message?.meta?.async_subagent_result ? message.meta : undefined; const copyToClipboard = async (text) => { const res = await _copyToClipboard(text); @@ -133,7 +143,7 @@ id="message-{message.id}" style="scroll-margin-top: 3rem;" > - {#if !($settings?.chatBubble ?? true)} + {#if !($settings?.chatBubble ?? true) && !subagentResult}
{/if} -
- {#if !($settings?.chatBubble ?? true)} +
+ {#if !($settings?.chatBubble ?? true) && !subagentResult}
{#if message.user} @@ -179,7 +189,7 @@ {/if}
- {:else if message.timestamp} + {:else if message.timestamp && !subagentResult}
@@ -394,7 +406,7 @@
{/if} - {#if edit !== true} + {#if edit !== true && !subagentResult}
60 ? `${task.slice(0, 60)}...` : task}"`; + } catch { + name = 'Sub-agent'; + } + } return { summary: isDone ? 'Tool Executed' : 'Executing...', @@ -134,7 +148,7 @@ function buildToolCallToken(item: OutputItem, toolOutputByCallId: Record {$i18n.t('Open @@ -1627,7 +1627,7 @@
{$i18n.t('Open diff --git a/src/lib/components/layout/Sidebar/UserMenu.svelte b/src/lib/components/layout/Sidebar/UserMenu.svelte index 8131d96d46..c6a463460c 100644 --- a/src/lib/components/layout/Sidebar/UserMenu.svelte +++ b/src/lib/components/layout/Sidebar/UserMenu.svelte @@ -122,49 +122,13 @@
{#if profile} -
-
- profile -
- -
-
- {$user.name} -
- -
- {#if $user?.is_active ?? true} -
- - - -
- - {$i18n.t('Active')} - {:else} -
- - - -
- - {$i18n.t('Away')} - {/if} -
-
-
- {#if $user?.status_emoji || $user?.status_message} -
+
{:else} -
+
+ + diff --git a/src/routes/+layout.svelte b/src/routes/+layout.svelte index b7f0799652..201b45f37b 100644 --- a/src/routes/+layout.svelte +++ b/src/routes/+layout.svelte @@ -616,7 +616,10 @@ } } - if ((event.chat_id !== $chatId && !$temporaryChatEnabled) || isInBackground) { + if ( + !event?.internal && + ((event.chat_id !== $chatId && !$temporaryChatEnabled) || isInBackground) + ) { if (type === 'chat:completion') { const { done, content, output, title } = data; const displayTitle = title || $i18n.t('New Chat');