feat: add upload_file_to_terminal builtin tool with file_id and acces… (#170)

backend/open_webui/tools/builtin.py

@@ -2823,3 +2823,138 @@ async def delete_automation(
     except Exception as e:
         log.exception(f'delete_automation error: {e}')
         return json.dumps({'error': str(e)})
+
+
+# =============================================================================
+# TERMINAL TOOLS
+# =============================================================================
+
+
+async def upload_file_to_terminal(
+    file_id: str,
+    __request__: Request = None,
+    __user__: dict = None,
+    __metadata__: dict = None,
+) -> str:
+    """
+    Upload a file to the connected Open Terminal server's working directory.
+    The file can be a chat attachment or a file from a knowledge base.
+
+    :param file_id: The ID of the file to upload
+    :return: JSON with the upload result including the path on the terminal server
+    """
+    if __request__ is None:
+        return json.dumps({'error': 'Request context not available'})
+
+    if not __user__:
+        return json.dumps({'error': 'User context not available'})
+
+    metadata = __metadata__ or {}
+    terminal_id = metadata.get('terminal_id')
+    if not terminal_id:
+        return json.dumps({'error': 'No terminal server connected to this chat'})
+
+    try:
+        from open_webui.models.files import Files
+        from open_webui.storage.provider import Storage
+        from open_webui.utils.access_control.files import has_access_to_file
+
+        user_id = __user__.get('id')
+        user_role = __user__.get('role', 'user')
+
+        # --- 1. Retrieve file and check access ---
+        file_record = await Files.get_file_by_id(file_id)
+        if not file_record:
+            return json.dumps({'error': 'File not found'})
+
+        if (
+            file_record.user_id != user_id
+            and user_role != 'admin'
+            and not await has_access_to_file(
+                file_id=file_id,
+                access_type='read',
+                user=UserModel(**__user__),
+            )
+        ):
+            return json.dumps({'error': 'File not found'})
+
+        local_path = Storage.get_file(file_record.path)
+
+        # --- 2. Resolve terminal connection ---
+        terminal_url = None
+        headers = {}
+
+        connections = __request__.app.state.config.TERMINAL_SERVER_CONNECTIONS or []
+        connection = next(
+            (c for c in connections if c.get('id') == terminal_id), None
+        )
+        if connection:
+            terminal_url = connection.get('url', '').rstrip('/')
+            auth_type = connection.get('auth_type', 'bearer')
+            if auth_type == 'bearer':
+                key = connection.get('key', '')
+                if key:
+                    headers['Authorization'] = f'Bearer {key}'
+        else:
+            tool_servers = metadata.get('tool_servers') or []
+            for ts in tool_servers:
+                ts_url = (ts.get('url') or '').rstrip('/')
+                if ts_url and ts_url == terminal_id.rstrip('/'):
+                    terminal_url = ts_url
+                    key = ts.get('key', '')
+                    if key:
+                        headers['Authorization'] = f'Bearer {key}'
+                    break
+
+        if not terminal_url:
+            return json.dumps(
+                {'error': f"Terminal connection '{terminal_id}' could not be resolved"}
+            )
+
+        # --- 3. Upload to terminal server ---
+        import httpx
+
+        async with httpx.AsyncClient(timeout=60.0) as client:
+            upload_dir = '.'
+            try:
+                cwd_resp = await client.get(
+                    f'{terminal_url}/files/cwd',
+                    headers=headers,
+                )
+                if cwd_resp.status_code == 200:
+                    upload_dir = cwd_resp.json().get('cwd', '.')
+            except Exception:
+                pass
+
+            log.info(
+                f"upload_file_to_terminal: uploading '{file_record.filename}' "
+                f'to {terminal_url}/files/upload?directory={upload_dir}'
+            )
+
+            with open(local_path, 'rb') as fh:
+                response = await client.post(
+                    f'{terminal_url}/files/upload',
+                    params={'directory': upload_dir},
+                    files={'file': (file_record.filename, fh)},
+                    headers=headers,
+                )
+
+        if response.status_code != 200:
+            detail = response.text[:500]
+            return json.dumps(
+                {'error': f'Upload failed (HTTP {response.status_code}): {detail}'}
+            )
+
+        result = response.json()
+        return json.dumps(
+            {
+                'status': 'success',
+                'message': f"File '{file_record.filename}' uploaded to terminal server",
+                'path': result.get('path'),
+                'size': result.get('size'),
+            },
+            ensure_ascii=False,
+        )
+    except Exception as e:
+        log.exception(f'upload_file_to_terminal error: {e}')
+        return json.dumps({'error': str(e)})

backend/open_webui/utils/middleware.py

@@ -2458,6 +2458,7 @@ async def process_chat_payload(request, form_data, user, metadata, model):
         'files': files,
     }
     form_data['metadata'] = metadata
+    extra_params['__metadata__'] = metadata
 
     # When the caller provides an explicit OpenAI-style `tools` array in the
     # request body, skip all server-side tool resolution and pass the caller's

backend/open_webui/utils/tools.py

@@ -91,6 +91,7 @@ from open_webui.tools.builtin import (
     list_automations,
     toggle_automation,
     delete_automation,
+    upload_file_to_terminal,
 )
 
 import copy
@@ -546,6 +547,11 @@ async def get_builtin_tools(
     if is_builtin_tool_enabled('automations') and await has_user_permission('automations'):
         builtin_functions.extend([create_automation, update_automation, list_automations, toggle_automation, delete_automation])
 
+    # Terminal tools - upload files to connected terminal server
+    terminal_id = (extra_params.get('__metadata__') or {}).get('terminal_id')
+    if is_builtin_tool_enabled('terminal') and terminal_id:
+        builtin_functions.append(upload_file_to_terminal)
+
     for func in builtin_functions:
         callable = await get_async_tool_function_and_apply_extra_params(
             func,