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

feat: open terminal integration

Browse Source
This commit is contained in:
Timothy Jaeryang Baek
2026-02-27 13:08:59 -06:00
parent 7ea6afdf95
commit 4737e1f118
15 changed files with 767 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/open_webui/config.py
+14
View File
@@ -1183,6 +1183,20 @@ TOOL_SERVER_CONNECTIONS = PersistentConfig(
tool_server_connections,
)
####################################
# TERMINAL_SERVER
####################################
terminal_server_connections = json.loads(
os.environ.get("TERMINAL_SERVER_CONNECTIONS", "[]")
)
TERMINAL_SERVER_CONNECTIONS = PersistentConfig(
"TERMINAL_SERVER_CONNECTIONS",
"terminal_server.connections",
terminal_server_connections,
)
####################################
# WEBUI
####################################
backend/open_webui/main.py
+19 -2
View File
@@ -96,6 +96,7 @@ from open_webui.routers import (
users,
utils,
scim,
terminals,
)
from open_webui.routers.retrieval import (
@@ -132,6 +133,8 @@ from open_webui.config import (
THREAD_POOL_SIZE,
# Tool Server Configs
TOOL_SERVER_CONNECTIONS,
# Terminal Server
TERMINAL_SERVER_CONNECTIONS,
# Code Execution
ENABLE_CODE_EXECUTION,
CODE_EXECUTION_ENGINE,
@@ -524,7 +527,7 @@ from open_webui.utils.middleware import (
process_chat_payload,
process_chat_response,
)
from open_webui.utils.tools import set_tool_servers
from open_webui.utils.tools import set_tool_servers, set_terminal_servers
from open_webui.utils.auth import (
get_license_data,
@@ -690,8 +693,11 @@ async def lifespan(app: FastAPI):
)
await set_tool_servers(mock_request)
log.info(f"Initialized {len(app.state.TOOL_SERVERS)} tool server(s)")
await set_terminal_servers(mock_request)
log.info(f"Initialized {len(app.state.TERMINAL_SERVERS)} terminal server(s)")
except Exception as e:
log.warning(f"Failed to initialize tool servers at startup: {e}")
log.warning(f"Failed to initialize tool/terminal servers at startup: {e}")
yield
@@ -775,6 +781,15 @@ app.state.OPENAI_MODELS = {}
app.state.config.TOOL_SERVER_CONNECTIONS = TOOL_SERVER_CONNECTIONS
app.state.TOOL_SERVERS = []
########################################
#
# TERMINAL SERVER
#
########################################
app.state.config.TERMINAL_SERVER_CONNECTIONS = TERMINAL_SERVER_CONNECTIONS
app.state.TERMINAL_SERVERS = []
########################################
#
# DIRECT CONNECTIONS
@@ -1540,6 +1555,7 @@ app.include_router(
if ENABLE_ADMIN_ANALYTICS:
app.include_router(analytics.router, prefix="/api/v1/analytics", tags=["analytics"])
app.include_router(utils.router, prefix="/api/v1/utils", tags=["utils"])
app.include_router(terminals.router, prefix="/api/v1/terminals", tags=["terminals"])
# SCIM 2.0 API for identity management
if ENABLE_SCIM:
@@ -2204,6 +2220,7 @@ async def get_app_config(request: Request):
"pending_user_overlay_content": app.state.config.PENDING_USER_OVERLAY_CONTENT,
"response_watermark": app.state.config.RESPONSE_WATERMARK,
},
"license_metadata": app.state.LICENSE_METADATA,
**(
{
backend/open_webui/routers/configs.py
+40
View File
@@ -15,6 +15,7 @@ from open_webui.utils.tools import (
get_tool_server_data,
get_tool_server_url,
set_tool_servers,
set_terminal_servers,
)
from open_webui.utils.mcp.client import MCPClient
from open_webui.models.oauth_sessions import OAuthSessions
@@ -214,6 +215,45 @@ async def set_tool_servers_config(
}
class TerminalServerConnection(BaseModel):
id: str
url: str
key: Optional[str] = ""
name: Optional[str] = ""
auth_type: Optional[str] = "bearer"
config: Optional[dict] = None # holds access_grants, etc.
model_config = ConfigDict(extra="allow")
class TerminalServersConfigForm(BaseModel):
TERMINAL_SERVER_CONNECTIONS: list[TerminalServerConnection]
@router.get("/terminal_servers")
async def get_terminal_servers_config(request: Request, user=Depends(get_admin_user)):
return {
"TERMINAL_SERVER_CONNECTIONS": request.app.state.config.TERMINAL_SERVER_CONNECTIONS,
}
@router.post("/terminal_servers")
async def set_terminal_servers_config(
request: Request,
form_data: TerminalServersConfigForm,
user=Depends(get_admin_user),
):
request.app.state.config.TERMINAL_SERVER_CONNECTIONS = [
connection.model_dump() for connection in form_data.TERMINAL_SERVER_CONNECTIONS
]
await set_terminal_servers(request)
return {
"TERMINAL_SERVER_CONNECTIONS": request.app.state.config.TERMINAL_SERVER_CONNECTIONS,
}
@router.post("/tool_servers/verify")
async def verify_tool_servers_config(
request: Request, form_data: ToolServerConnection, user=Depends(get_admin_user)
backend/open_webui/routers/terminals.py
+137
View File
@@ -0,0 +1,137 @@
"""Reverse proxy for admin-configured terminal servers.
Routes:
GET / — list terminals the user has access to
* /{server_id}/{path:path} — proxy request to terminal server
"""
import logging
import aiohttp
from fastapi import APIRouter, Depends, Request, Response
from fastapi.responses import JSONResponse, StreamingResponse
from starlette.background import BackgroundTask
from open_webui.utils.auth import get_verified_user
from open_webui.utils.access_control import has_connection_access
from open_webui.models.groups import Groups
log = logging.getLogger(__name__)
router = APIRouter()
STREAMING_CONTENT_TYPES = ("application/octet-stream", "image/", "application/pdf")
STRIPPED_RESPONSE_HEADERS = frozenset(
("transfer-encoding", "connection", "content-encoding", "content-length")
)
@router.get("/")
async def list_terminal_servers(request: Request, user=Depends(get_verified_user)):
"""Return terminal servers the authenticated user has access to."""
connections = request.app.state.config.TERMINAL_SERVER_CONNECTIONS or []
user_group_ids = {group.id for group in Groups.get_groups_by_member_id(user.id)}
return [
{"id": connection.get("id", ""), "url": connection.get("url", ""), "name": connection.get("name", "")}
for connection in connections
if has_connection_access(user, connection, user_group_ids)
]
PROXY_METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"]
@router.api_route("/{server_id}/{path:path}", methods=PROXY_METHODS)
async def proxy_terminal(
server_id: str,
path: str,
request: Request,
user=Depends(get_verified_user),
):
"""Proxy a request to the admin terminal server identified by *server_id*."""
connections = request.app.state.config.TERMINAL_SERVER_CONNECTIONS or []
connection = next((c for c in connections if c.get("id") == server_id), None)
if connection is None:
return JSONResponse({"error": f"Terminal server '{server_id}' not found"}, status_code=404)
user_group_ids = {group.id for group in Groups.get_groups_by_member_id(user.id)}
if not has_connection_access(user, connection, user_group_ids):
return JSONResponse({"error": "Access denied"}, status_code=403)
base_url = (connection.get("url") or "").rstrip("/")
if not base_url:
return JSONResponse({"error": "Terminal server URL not configured"}, status_code=503)
target_url = f"{base_url}/{path}"
if request.query_params:
target_url += f"?{request.query_params}"
headers = {"X-User-Id": user.id}
cookies = {}
auth_type = connection.get("auth_type", "bearer")
if auth_type == "bearer":
headers["Authorization"] = f"Bearer {connection.get('key', '')}"
elif auth_type == "session":
cookies = request.cookies
headers["Authorization"] = f"Bearer {request.state.token.credentials}"
elif auth_type == "system_oauth":
cookies = request.cookies
oauth_token = request.headers.get("x-oauth-access-token", "")
if oauth_token:
headers["Authorization"] = f"Bearer {oauth_token}"
# auth_type == "none": no Authorization header
content_type = request.headers.get("content-type")
if content_type:
headers["Content-Type"] = content_type
body = await request.body()
session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=300, connect=10),
trust_env=True,
)
try:
upstream_response = await session.request(
method=request.method,
url=target_url,
headers=headers,
cookies=cookies,
data=body or None,
)
upstream_content_type = upstream_response.headers.get("content-type", "")
filtered_headers = {
key: value
for key, value in upstream_response.headers.items()
if key.lower() not in STRIPPED_RESPONSE_HEADERS
}
# Stream binary responses directly
if any(t in upstream_content_type for t in STREAMING_CONTENT_TYPES):
async def cleanup():
await upstream_response.release()
await session.close()
return StreamingResponse(
content=upstream_response.content.iter_any(),
status_code=upstream_response.status,
headers=filtered_headers,
background=BackgroundTask(cleanup),
)
# Buffer text/JSON responses
response_body = await upstream_response.read()
status_code = upstream_response.status
await upstream_response.release()
await session.close()
return Response(content=response_body, status_code=status_code, headers=filtered_headers)
except Exception as error:
await session.close()
log.exception("Terminal proxy error: %s", error)
return JSONResponse({"error": f"Terminal proxy error: {error}"}, status_code=502)
backend/open_webui/utils/access_control.py
+28
View File
@@ -153,6 +153,34 @@ def has_access(
return False
def has_connection_access(
user: UserModel,
connection: dict,
user_group_ids: Optional[Set[str]] = None,
) -> bool:
"""
Check if a user can access a server connection (tool server, terminal, etc.)
based on ``config.access_grants`` within the connection dict.
- Admin with BYPASS_ADMIN_ACCESS_CONTROL → always allowed
- Empty / missing access_grants → allowed for all users
- Otherwise → delegates to ``has_access``
"""
from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL
if user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL:
return True
if user_group_ids is None:
user_group_ids = {group.id for group in Groups.get_groups_by_member_id(user.id)}
access_grants = (connection.get("config") or {}).get("access_grants", [])
if not access_grants:
return True
return has_access(user.id, "read", access_grants, user_group_ids)
def migrate_access_control(
data: dict, ac_key: str = "access_control", grants_key: str = "access_grants"
) -> None:
backend/open_webui/utils/middleware.py
+16 -2
View File
@@ -99,8 +99,9 @@ from open_webui.utils.misc import (
from open_webui.utils.tools import (
get_tools,
get_updated_tool_function,
has_tool_server_access,
get_terminal_tools,
)
from open_webui.utils.access_control import has_connection_access
from open_webui.utils.plugin import load_function_module_by_id
from open_webui.utils.filter import (
get_sorted_filter_ids,
@@ -2225,6 +2226,7 @@ async def process_chat_payload(request, form_data, user, metadata, model):
)
tool_ids = form_data.pop("tool_ids", None)
terminal_id = form_data.pop("terminal_id", None)
files = form_data.pop("files", None)
# Caller-provided OpenAI-style tools take precedence over server-side
@@ -2298,6 +2300,7 @@ async def process_chat_payload(request, form_data, user, metadata, model):
metadata = {
**metadata,
"tool_ids": tool_ids,
"terminal_id": terminal_id,
"files": files,
}
form_data["metadata"] = metadata
@@ -2342,7 +2345,7 @@ async def process_chat_payload(request, form_data, user, metadata, model):
continue
# Check access control for MCP server
if not has_tool_server_access(user, mcp_server_connection):
if not has_connection_access(user, mcp_server_connection):
log.warning(
f"Access denied to MCP server {server_id} for user {user.id}"
)
@@ -2479,6 +2482,17 @@ async def process_chat_payload(request, form_data, user, metadata, model):
if mcp_tools_dict:
tools_dict = {**tools_dict, **mcp_tools_dict}
# Resolve terminal tools if terminal_id is set
if terminal_id:
terminal_tools = await get_terminal_tools(
request,
terminal_id,
user,
extra_params,
)
if terminal_tools:
tools_dict = {**tools_dict, **terminal_tools}
if direct_tool_servers:
for tool_server in direct_tool_servers:
tool_specs = tool_server.pop("specs", [])
backend/open_webui/utils/tools.py
+188 -17
View File
@@ -40,7 +40,7 @@ from open_webui.models.users import UserModel
from open_webui.models.groups import Groups
from open_webui.models.access_grants import AccessGrants
from open_webui.utils.plugin import load_tool_module_by_id
from open_webui.utils.access_control import has_access
from open_webui.utils.access_control import has_access, has_connection_access
from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL
from open_webui.env import (
AIOHTTP_CLIENT_TIMEOUT,
@@ -144,19 +144,6 @@ def get_updated_tool_function(function: Callable, extra_params: dict):
return function
def has_tool_server_access(
user: UserModel, server_connection: dict, user_group_ids: set = None
) -> bool:
"""Check if user has access to a tool server (MCP or OpenAPI)."""
if user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL:
return True
if user_group_ids is None:
user_group_ids = {group.id for group in Groups.get_groups_by_member_id(user.id)}
server_config = server_connection.get("config", {})
access_grants = server_config.get("access_grants", [])
return has_access(user.id, "read", access_grants, user_group_ids)
async def get_tools(
@@ -297,7 +284,7 @@ async def get_tools(
)
# Check access control for tool server
if not has_tool_server_access(
if not has_connection_access(
user, tool_server_connection, user_group_ids
):
log.warning(
@@ -394,7 +381,7 @@ async def get_tools(
tool_dict = {
"tool_id": tool_id,
"callable": callable,
"spec": spec,
"spec": clean_openai_tool_schema(spec),
# Misc info
"type": "external",
}
@@ -810,7 +797,7 @@ def convert_openapi_to_tool_payload(openapi_spec):
f". Possible values: {', '.join(param_schema.get('enum'))}"
)
param_property = {
"type": param_schema.get("type"),
"type": param_schema.get("type") or "string",
"description": description,
}
@@ -818,6 +805,11 @@ def convert_openapi_to_tool_payload(openapi_spec):
if param_schema.get("type") == "array" and "items" in param_schema:
param_property["items"] = param_schema["items"]
# Filter out None values to prevent schema validation errors
param_property = {
k: v for k, v in param_property.items() if v is not None
}
tool["parameters"]["properties"][param_name] = param_property
if param.get("required"):
tool["parameters"]["required"].append(param_name)
@@ -881,6 +873,180 @@ async def get_tool_servers(request: Request):
return tool_servers
async def get_terminal_cwd(
base_url: str,
headers: dict,
cookies: Optional[dict] = None,
) -> Optional[str]:
"""Fetch the current working directory from a terminal server."""
try:
cwd_url = f"{base_url.rstrip('/')}/files/cwd"
async with aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=5),
trust_env=True,
) as session:
async with session.get(
cwd_url, headers=headers, cookies=cookies or {}
) as resp:
if resp.status == 200:
data = await resp.json()
return data.get("cwd")
except Exception as e:
log.debug(f"Failed to fetch terminal CWD: {e}")
return None
async def set_terminal_servers(request: Request):
"""Load and cache OpenAPI specs from all TERMINAL_SERVER_CONNECTIONS."""
connections = request.app.state.config.TERMINAL_SERVER_CONNECTIONS or []
# Build server configs with info containing the connection ID
server_configs = []
for connection in connections:
conn_id = connection.get("id", "")
if not connection.get("url"):
continue
auth_type = connection.get("auth_type", "bearer")
token = None
if auth_type == "bearer":
token = connection.get("key", "")
server_configs.append({
"url": connection.get("url", ""),
"key": token or "",
"auth_type": auth_type,
"path": "openapi.json",
"spec_type": "url",
"config": {"enable": True},
"info": {"id": conn_id, "name": connection.get("name", "")},
})
request.app.state.TERMINAL_SERVERS = await get_tool_servers_data(server_configs)
if request.app.state.redis is not None:
await request.app.state.redis.set(
"terminal_servers", json.dumps(request.app.state.TERMINAL_SERVERS)
)
return request.app.state.TERMINAL_SERVERS
async def get_terminal_servers(request: Request):
"""Return cached terminal server specs, loading if needed."""
terminal_servers = []
if request.app.state.redis is not None:
try:
terminal_servers = json.loads(
await request.app.state.redis.get("terminal_servers")
)
request.app.state.TERMINAL_SERVERS = terminal_servers
except Exception as e:
log.error(f"Error fetching terminal_servers from Redis: {e}")
if not terminal_servers:
terminal_servers = await set_terminal_servers(request)
return terminal_servers
async def get_terminal_tools(
request: Request,
terminal_id: str,
user: UserModel,
extra_params: dict,
) -> dict[str, dict]:
"""Resolve tools for a terminal server identified by terminal_id.
- Finds the connection in TERMINAL_SERVER_CONNECTIONS
- Checks access_grants
- Loads specs from cache
- Builds callables that route through the terminal proxy
"""
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 is None:
log.warning(f"Terminal server not found: {terminal_id}")
return {}
user_group_ids = {group.id for group in Groups.get_groups_by_member_id(user.id)}
if not has_connection_access(user, connection, user_group_ids):
log.warning(f"Access denied to terminal {terminal_id} for user {user.id}")
return {}
# 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 {}
specs = server_data.get("specs", [])
if not specs:
return {}
# Build auth headers
auth_type = connection.get("auth_type", "bearer")
cookies = {}
headers = {"Content-Type": "application/json", "X-User-Id": user.id}
if auth_type == "bearer":
headers["Authorization"] = f"Bearer {connection.get('key', '')}"
elif auth_type == "session":
cookies = request.cookies
headers["Authorization"] = f"Bearer {request.state.token.credentials}"
elif auth_type == "system_oauth":
cookies = request.cookies
oauth_token = extra_params.get("__oauth_token__", None)
if oauth_token:
headers["Authorization"] = f"Bearer {oauth_token.get('access_token', '')}"
# auth_type == "none": no Authorization header
terminal_cwd = await get_terminal_cwd(
connection.get("url", ""), headers, cookies
)
tools_dict = {}
for spec in specs:
function_name = spec["name"]
# Inject CWD into run_command description
tool_spec = clean_openai_tool_schema(spec)
if function_name == "run_command" and terminal_cwd:
tool_spec["description"] = (
tool_spec.get("description", "")
+ f"\n\nThe current working directory is: {terminal_cwd}"
)
def make_tool_function(fn_name, srv_data, hdrs, cks):
async def tool_function(**kwargs):
return await execute_tool_server(
url=srv_data["url"],
headers=hdrs,
cookies=cks,
name=fn_name,
params=kwargs,
server_data=srv_data,
)
return tool_function
tool_function = make_tool_function(function_name, server_data, headers, cookies)
callable = get_async_tool_function_and_apply_extra_params(tool_function, {})
tools_dict[function_name] = {
"tool_id": f"terminal:{terminal_id}",
"callable": callable,
"spec": tool_spec,
"type": "terminal",
}
return tools_dict
async def get_tool_server_data(url: str, headers: Optional[dict]) -> Dict[str, Any]:
_headers = {
"Accept": "application/json",
@@ -997,6 +1163,11 @@ async def get_tool_servers_data(servers: List[Dict[str, Any]]) -> List[Dict[str,
log.error(f"Failed to connect to {url} OpenAPI tool server")
continue
# Guard against invalid or non-OpenAPI specs (e.g., MCP-style configs)
if not isinstance(response, dict) or "paths" not in response:
log.warning(f"Invalid OpenAPI spec from {url}: missing 'paths'")
continue
response = {
"openapi": response,
"info": response.get("info", {}),