diff --git a/mcpgateway/admin.py b/mcpgateway/admin.py
index 325e29406..46cd87458 100644
--- a/mcpgateway/admin.py
+++ b/mcpgateway/admin.py
@@ -1910,7 +1910,7 @@ async def admin_add_server(request: Request, db: Session = Depends(get_db), user
>>> server_service.register_server = original_register_server
"""
form = await request.form()
- # root_path = request.scope.get("root_path", "")
+ # root_path = settings.app_root_path
# is_inactive_checked = form.get("is_inactive_checked", "false")
# Parse tags from comma-separated string
@@ -2268,14 +2268,14 @@ async def admin_toggle_server(
>>>
>>> # Happy path: Activate server
>>> form_data_activate = FormData([("activate", "true"), ("is_inactive_checked", "false")])
- >>> mock_request_activate = MagicMock(spec=Request, scope={"root_path": ""})
+ >>> mock_request_activate = MagicMock(spec=Request)
>>> mock_request_activate.form = AsyncMock(return_value=form_data_activate)
>>> original_toggle_server_status = server_service.toggle_server_status
>>> server_service.toggle_server_status = AsyncMock()
>>>
>>> async def test_admin_toggle_server_activate():
... result = await admin_toggle_server(server_id, mock_request_activate, mock_db, mock_user)
- ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin#catalog" in result.headers["location"]
+ ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin/#catalog" in result.headers["location"]
>>>
>>> asyncio.run(test_admin_toggle_server_activate())
True
@@ -2285,16 +2285,23 @@ async def admin_toggle_server(
>>> mock_request_deactivate = MagicMock(spec=Request, scope={"root_path": "/api"})
>>> mock_request_deactivate.form = AsyncMock(return_value=form_data_deactivate)
>>>
+ >>> # Mock settings.app_root_path
+ >>> import mcpgateway.admin as admin_module
+ >>> original_app_root_path = admin_module.settings.app_root_path
+ >>> admin_module.settings.app_root_path = "/api"
+ >>>
>>> async def test_admin_toggle_server_deactivate():
... result = await admin_toggle_server(server_id, mock_request_deactivate, mock_db, mock_user)
- ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/api/admin#catalog" in result.headers["location"]
+ ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/api/admin/#catalog" in result.headers["location"]
>>>
>>> asyncio.run(test_admin_toggle_server_deactivate())
True
+ >>> # Restore
+ >>> admin_module.settings.app_root_path = original_app_root_path
>>>
>>> # Edge case: Toggle with inactive checkbox checked
>>> form_data_inactive = FormData([("activate", "true"), ("is_inactive_checked", "true")])
- >>> mock_request_inactive = MagicMock(spec=Request, scope={"root_path": ""})
+ >>> mock_request_inactive = MagicMock(spec=Request)
>>> mock_request_inactive.form = AsyncMock(return_value=form_data_inactive)
>>>
>>> async def test_admin_toggle_server_inactive_checked():
@@ -2306,7 +2313,7 @@ async def admin_toggle_server(
>>>
>>> # Error path: Simulate an exception during toggle
>>> form_data_error = FormData([("activate", "true")])
- >>> mock_request_error = MagicMock(spec=Request, scope={"root_path": ""})
+ >>> mock_request_error = MagicMock(spec=Request)
>>> mock_request_error.form = AsyncMock(return_value=form_data_error)
>>> server_service.toggle_server_status = AsyncMock(side_effect=Exception("Toggle failed"))
>>>
@@ -2342,7 +2349,7 @@ async def admin_toggle_server(
LOGGER.error(f"Error toggling server status: {e}")
error_message = "Error toggling server status. Please try again."
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
# Build redirect URL with error message if present
if error_message:
@@ -2353,7 +2360,7 @@ async def admin_toggle_server(
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#catalog", status_code=303)
- return RedirectResponse(f"{root_path}/admin#catalog", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#catalog", status_code=303)
@admin_router.post("/servers/{server_id}/delete")
@@ -2394,7 +2401,7 @@ async def admin_delete_server(server_id: str, request: Request, db: Session = De
>>>
>>> async def test_admin_delete_server_success():
... result = await admin_delete_server(server_id, mock_request_delete, mock_db, mock_user)
- ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin#catalog" in result.headers["location"]
+ ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin/#catalog" in result.headers["location"]
>>>
>>> asyncio.run(test_admin_delete_server_success())
True
@@ -2442,6 +2449,9 @@ async def admin_delete_server(server_id: str, request: Request, db: Session = De
LOGGER.error(f"Error deleting server: {e}")
error_message = "Failed to delete server. Please try again."
+ form = await request.form()
+ is_inactive_checked = str(form.get("is_inactive_checked", "false"))
+ root_path = settings.app_root_path
root_path = request.scope.get("root_path", "")
# Build redirect URL with error message if present
@@ -2453,7 +2463,7 @@ async def admin_delete_server(server_id: str, request: Request, db: Session = De
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#catalog", status_code=303)
- return RedirectResponse(f"{root_path}/admin#catalog", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#catalog", status_code=303)
@admin_router.get("/resources", response_model=PaginatedResponse)
@@ -2771,33 +2781,40 @@ async def admin_toggle_gateway(
>>>
>>> # Happy path: Activate gateway
>>> form_data_activate = FormData([("activate", "true"), ("is_inactive_checked", "false")])
- >>> mock_request_activate = MagicMock(spec=Request, scope={"root_path": ""})
+ >>> mock_request_activate = MagicMock(spec=Request)
>>> mock_request_activate.form = AsyncMock(return_value=form_data_activate)
>>> original_toggle_gateway_status = gateway_service.toggle_gateway_status
>>> gateway_service.toggle_gateway_status = AsyncMock()
>>>
>>> async def test_admin_toggle_gateway_activate():
... result = await admin_toggle_gateway(gateway_id, mock_request_activate, mock_db, mock_user)
- ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin#gateways" in result.headers["location"]
+ ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin/#gateways" in result.headers["location"]
>>>
>>> asyncio.run(test_admin_toggle_gateway_activate())
True
>>>
>>> # Happy path: Deactivate gateway
>>> form_data_deactivate = FormData([("activate", "false"), ("is_inactive_checked", "false")])
- >>> mock_request_deactivate = MagicMock(spec=Request, scope={"root_path": "/api"})
+ >>> mock_request_deactivate = MagicMock(spec=Request)
>>> mock_request_deactivate.form = AsyncMock(return_value=form_data_deactivate)
>>>
+ >>> # Mock settings.app_root_path
+ >>> import mcpgateway.admin as admin_module
+ >>> original_app_root_path = admin_module.settings.app_root_path
+ >>> admin_module.settings.app_root_path = "/api"
+ >>>
>>> async def test_admin_toggle_gateway_deactivate():
... result = await admin_toggle_gateway(gateway_id, mock_request_deactivate, mock_db, mock_user)
- ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/api/admin#gateways" in result.headers["location"]
+ ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/api/admin/#gateways" in result.headers["location"]
>>>
>>> asyncio.run(test_admin_toggle_gateway_deactivate())
True
+ >>> # Restore
+ >>> admin_module.settings.app_root_path = original_app_root_path
>>>
>>> # Error path: Simulate an exception during toggle
>>> form_data_error = FormData([("activate", "true")])
- >>> mock_request_error = MagicMock(spec=Request, scope={"root_path": ""})
+ >>> mock_request_error = MagicMock(spec=Request)
>>> mock_request_error.form = AsyncMock(return_value=form_data_error)
>>> gateway_service.toggle_gateway_status = AsyncMock(side_effect=Exception("Toggle failed"))
>>>
@@ -2833,7 +2850,7 @@ async def admin_toggle_gateway(
LOGGER.error(f"Error toggling gateway status: {e}")
error_message = "Failed to toggle gateway status. Please try again."
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
# Build redirect URL with error message if present
if error_message:
@@ -2844,7 +2861,7 @@ async def admin_toggle_gateway(
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#gateways", status_code=303)
- return RedirectResponse(f"{root_path}/admin#gateways", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#gateways", status_code=303)
@admin_router.get("/", name="admin_home", response_class=HTMLResponse)
@@ -3487,8 +3504,8 @@ async def admin_login_page(request: Request) -> Response:
"""
# Check if email auth is enabled
if not getattr(settings, "email_auth_enabled", False):
- root_path = request.scope.get("root_path", "")
- return RedirectResponse(url=f"{root_path}/admin", status_code=303)
+ root_path = settings.app_root_path
+ return RedirectResponse(url=f"{root_path}/admin/", status_code=303)
root_path = settings.app_root_path
@@ -3543,8 +3560,8 @@ async def admin_login_handler(request: Request, db: Session = Depends(get_db)) -
True
"""
if not getattr(settings, "email_auth_enabled", False):
- root_path = request.scope.get("root_path", "")
- return RedirectResponse(url=f"{root_path}/admin", status_code=303)
+ root_path = settings.app_root_path
+ return RedirectResponse(url=f"{root_path}/admin/", status_code=303)
try:
form = await request.form()
@@ -3554,7 +3571,7 @@ async def admin_login_handler(request: Request, db: Session = Depends(get_db)) -
password = password_val if isinstance(password_val, str) else None
if not email or not password:
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/login?error=missing_fields", status_code=303)
# Authenticate using the email auth service
@@ -3568,7 +3585,7 @@ async def admin_login_handler(request: Request, db: Session = Depends(get_db)) -
if not user:
LOGGER.warning(f"Authentication failed for {email} - user is None")
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/login?error=invalid_credentials", status_code=303)
# Password change enforcement respects master switch and toggles
@@ -3615,7 +3632,7 @@ async def admin_login_handler(request: Request, db: Session = Depends(get_db)) -
token, _ = await create_access_token(user)
# Create redirect response to password change page
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
response = RedirectResponse(url=f"{root_path}/admin/change-password-required", status_code=303)
# Set JWT token as secure cookie for the password change process
@@ -3627,8 +3644,8 @@ async def admin_login_handler(request: Request, db: Session = Depends(get_db)) -
token, _ = await create_access_token(user) # expires_seconds not needed here
# Create redirect response
- root_path = request.scope.get("root_path", "")
- response = RedirectResponse(url=f"{root_path}/admin", status_code=303)
+ root_path = settings.app_root_path
+ response = RedirectResponse(url=f"{root_path}/admin/", status_code=303)
# Set JWT token as secure cookie
set_auth_cookie(response, token, remember_me=False)
@@ -3642,12 +3659,12 @@ async def admin_login_handler(request: Request, db: Session = Depends(get_db)) -
if settings.secure_cookies and settings.environment == "development":
LOGGER.warning("Login failed - set SECURE_COOKIES to false in config for HTTP development")
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/login?error=invalid_credentials", status_code=303)
except Exception as e:
LOGGER.error(f"Login handler error: {e}")
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/login?error=server_error", status_code=303)
@@ -3750,11 +3767,11 @@ async def change_password_required_page(request: Request) -> HTMLResponse:
True
"""
if not getattr(settings, "email_auth_enabled", False):
- root_path = request.scope.get("root_path", "")
- return RedirectResponse(url=f"{root_path}/admin", status_code=303)
+ root_path = settings.app_root_path
+ return RedirectResponse(url=f"{root_path}/admin/", status_code=303)
# Get root path for template
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return request.app.state.templates.TemplateResponse(
"change-password-required.html",
@@ -3815,8 +3832,8 @@ async def change_password_required_handler(request: Request, db: Session = Depen
True
"""
if not getattr(settings, "email_auth_enabled", False):
- root_path = request.scope.get("root_path", "")
- return RedirectResponse(url=f"{root_path}/admin", status_code=303)
+ root_path = settings.app_root_path
+ return RedirectResponse(url=f"{root_path}/admin/", status_code=303)
try:
form = await request.form()
@@ -3829,18 +3846,18 @@ async def change_password_required_handler(request: Request, db: Session = Depen
confirm_password = confirm_password_val if isinstance(confirm_password_val, str) else None
if not all([current_password, new_password, confirm_password]):
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/change-password-required?error=missing_fields", status_code=303)
if new_password != confirm_password:
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/change-password-required?error=mismatch", status_code=303)
# Get user from JWT token in cookie
try:
jwt_token = request.cookies.get("jwt_token")
if not jwt_token:
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/login?error=session_expired", status_code=303)
# Authenticate using the token
@@ -3850,11 +3867,11 @@ async def change_password_required_handler(request: Request, db: Session = Depen
current_user = await get_current_user(credentials, request=request)
if not current_user:
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/login?error=session_expired", status_code=303)
except Exception as e:
LOGGER.error(f"Authentication error: {e}")
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/login?error=session_expired", status_code=303)
# Authenticate using the email auth service
@@ -3896,8 +3913,8 @@ async def change_password_required_handler(request: Request, db: Session = Depen
token, _ = await create_access_token(current_user)
# Create redirect response to admin panel
- root_path = request.scope.get("root_path", "")
- response = RedirectResponse(url=f"{root_path}/admin", status_code=303)
+ root_path = settings.app_root_path
+ response = RedirectResponse(url=f"{root_path}/admin/", status_code=303)
# Update JWT token cookie
set_auth_cookie(response, token, remember_me=False)
@@ -3905,24 +3922,24 @@ async def change_password_required_handler(request: Request, db: Session = Depen
LOGGER.info(f"User {current_user.email} successfully changed their expired password")
return response
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/change-password-required?error=change_failed", status_code=303)
except AuthenticationError:
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/change-password-required?error=invalid_password", status_code=303)
except PasswordValidationError as e:
LOGGER.warning(f"Password validation failed for {current_user.email}: {e}")
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/change-password-required?error=weak_password", status_code=303)
except Exception as e:
LOGGER.error(f"Password change failed for {current_user.email}: {e}", exc_info=True)
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/change-password-required?error=server_error", status_code=303)
except Exception as e:
LOGGER.error(f"Password change handler error: {e}")
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return RedirectResponse(url=f"{root_path}/admin/change-password-required?error=server_error", status_code=303)
@@ -4492,7 +4509,7 @@ async def admin_list_teams(
if not current_user:
return HTMLResponse(content='', status_code=200)
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
if unified:
# Generate unified team view
@@ -4576,7 +4593,7 @@ async def admin_create_team(
try:
# Get root path for URL construction
- root_path = request.scope.get("root_path", "") if request else ""
+ root_path = settings.app_root_path
form = await request.form()
name = form.get("name")
@@ -4718,7 +4735,7 @@ async def admin_view_team_members(
try:
# Get root_path from request
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
# Get current user context for logging and authorization
user_email = get_user_email(user)
@@ -4993,7 +5010,7 @@ async def admin_get_team_edit(
try:
# Get root path for URL construction
- root_path = _request.scope.get("root_path", "") if _request else ""
+ root_path = settings.app_root_path if _request else ""
team_service = TeamManagementService(db)
team = await team_service.get_team_by_id(team_id)
@@ -5072,7 +5089,7 @@ async def admin_update_team(
Response: Result of team update operation
"""
# Ensure root_path is available for URL construction in all branches
- root_path = request.scope.get("root_path", "") if request else ""
+ root_path = settings.app_root_path
if not settings.email_auth_enabled:
return HTMLResponse(content='Email authentication is disabled
', status_code=403)
@@ -5794,7 +5811,6 @@ async def admin_cancel_join_request(
@require_permission("teams.manage_members")
async def admin_list_join_requests(
team_id: str,
- request: Request,
db: Session = Depends(get_db),
user=Depends(get_current_user_with_permissions),
) -> HTMLResponse:
@@ -5815,7 +5831,6 @@ async def admin_list_join_requests(
try:
team_service = TeamManagementService(db)
user_email = get_user_email(user)
- request.scope.get("root_path", "")
# Get team and verify ownership
team = await team_service.get_team_by_id(team_id)
@@ -6131,6 +6146,10 @@ async def admin_list_users(
status_code=200,
)
+ # Get root_path from request
+ root_path = settings.app_root_path
+
+ # First-Party
LOGGER.debug(f"User {get_user_email(user)} requested user list (page={page}, per_page={per_page})")
auth_service = EmailAuthService(db)
@@ -6542,6 +6561,9 @@ async def admin_create_user(
HTMLResponse: Success message or error response
"""
try:
+ # Get root path for URL construction
+ root_path = settings.app_root_path
+
form = await request.form()
# Validate password strength
@@ -6602,7 +6624,7 @@ async def admin_get_user_edit(
try:
# Get root path for URL construction
- root_path = _request.scope.get("root_path", "") if _request else ""
+ root_path = settings.app_root_path if _request else ""
# First-Party
@@ -6836,7 +6858,7 @@ async def admin_activate_user(
try:
# Get root path for URL construction
- root_path = _request.scope.get("root_path", "") if _request else ""
+ root_path = settings.app_root_path if _request else ""
# First-Party
@@ -6881,7 +6903,7 @@ async def admin_deactivate_user(
try:
# Get root path for URL construction
- root_path = _request.scope.get("root_path", "") if _request else ""
+ root_path = settings.app_root_path if _request else ""
# First-Party
@@ -7011,7 +7033,7 @@ async def admin_force_password_change(
try:
# Get root path for URL construction
- root_path = _request.scope.get("root_path", "") if _request else ""
+ root_path = settings.app_root_path if _request else ""
auth_service = EmailAuthService(db)
@@ -7238,7 +7260,7 @@ async def admin_tools_partial_html(
"hx_target": "#tools-table-body",
"hx_indicator": "#tools-loading",
"query_params": query_params_dict,
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
},
)
@@ -7250,7 +7272,7 @@ async def admin_tools_partial_html(
"request": request,
"data": data,
"pagination": pagination.model_dump(),
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
"gateway_id": gateway_id,
},
)
@@ -7263,7 +7285,7 @@ async def admin_tools_partial_html(
"data": data,
"pagination": pagination.model_dump(),
"links": links.model_dump() if links else None,
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
"include_inactive": include_inactive,
},
)
@@ -7630,7 +7652,7 @@ async def admin_prompts_partial_html(
"hx_target": "#prompts-table-body",
"hx_indicator": "#prompts-loading",
"query_params": query_params,
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
},
)
@@ -7641,7 +7663,7 @@ async def admin_prompts_partial_html(
"request": request,
"data": data,
"pagination": pagination.model_dump(),
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
"gateway_id": gateway_id,
},
)
@@ -7653,7 +7675,169 @@ async def admin_prompts_partial_html(
"data": data,
"pagination": pagination.model_dump(),
"links": links.model_dump() if links else None,
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
+ "include_inactive": include_inactive,
+ },
+ )
+
+
+@admin_router.get("/gateways/partial", response_class=HTMLResponse)
+async def admin_gateways_partial_html(
+ request: Request,
+ page: int = Query(1, ge=1, description="Page number (1-indexed)"),
+ per_page: int = Query(settings.pagination_default_page_size, ge=1, le=settings.pagination_max_page_size, description="Items per page"),
+ include_inactive: bool = False,
+ render: Optional[str] = Query(None),
+ team_id: Optional[str] = Depends(_validated_team_id_param),
+ db: Session = Depends(get_db),
+ user=Depends(get_current_user_with_permissions),
+):
+ """Return paginated gateways HTML partials for the admin UI.
+
+ This HTMX endpoint returns only the partial HTML used by the admin UI for
+ gateways. It supports three render modes:
+
+ - default: full table partial (rows + controls)
+ - ``render="controls"``: return only pagination controls
+ - ``render="selector"``: return selector items for infinite scroll
+
+ Args:
+ request (Request): FastAPI request object used by the template engine.
+ page (int): Page number (1-indexed).
+ per_page (int): Number of items per page (bounded by settings).
+ include_inactive (bool): If True, include inactive gateways in results.
+ render (Optional[str]): Render mode; one of None, "controls", "selector".
+ team_id (Optional[str]): Filter by team ID.
+ db (Session): Database session (dependency-injected).
+ user: Authenticated user object from dependency injection.
+
+ Returns:
+ Union[HTMLResponse, TemplateResponse]: A rendered template response
+ containing either the table partial, pagination controls, or selector
+ items depending on ``render``. The response contains JSON-serializable
+ encoded gateway data when templates expect it.
+ """
+ user_email = get_user_email(user)
+ LOGGER.info(f"🔷 GATEWAYS PARTIAL REQUEST - User: {user_email}, team_id: {team_id}, page: {page}, render: {render}, referer: {request.headers.get('referer', 'none')}")
+ # Normalize per_page within configured bounds
+ per_page = max(settings.pagination_min_page_size, min(per_page, settings.pagination_max_page_size))
+
+ # Team scoping
+ team_service = TeamManagementService(db)
+ user_teams = await team_service.get_user_teams(user_email)
+ team_ids = [t.id for t in user_teams]
+
+ # Build base query
+ query = select(DbGateway)
+
+ if not include_inactive:
+ query = query.where(DbGateway.enabled.is_(True))
+
+ # Build access conditions
+ # When team_id is specified, show ONLY items from that team (simpler, team-scoped view)
+ # When team_id is NOT specified, show all accessible items (owned + team + public)
+ if team_id:
+ # Team-specific view: only show gateways from the specified team if user is a member
+ if team_id in team_ids:
+ # Apply visibility check: team/public resources + user's own resources (including private)
+ team_access = [
+ and_(DbGateway.team_id == team_id, DbGateway.visibility.in_(["team", "public"])),
+ and_(DbGateway.team_id == team_id, DbGateway.owner_email == user_email),
+ ]
+ query = query.where(or_(*team_access))
+ LOGGER.debug(f"Filtering gateways by team_id: {team_id}")
+ else:
+ # User is not a member of this team, return no results
+ LOGGER.warning(f"User {user_email} attempted to filter by team {team_id} but is not a member")
+ query = query.where(false())
+ else:
+ # All Teams view: apply standard access conditions
+ access_conditions = []
+ access_conditions.append(DbGateway.owner_email == user_email)
+ if team_ids:
+ access_conditions.append(and_(DbGateway.team_id.in_(team_ids), DbGateway.visibility.in_(["team", "public"])))
+ access_conditions.append(DbGateway.visibility == "public")
+
+ query = query.where(or_(*access_conditions))
+
+ # Apply pagination ordering for cursor support
+ query = query.order_by(desc(DbGateway.created_at), desc(DbGateway.id))
+
+ # Build query params for pagination links
+ query_params = {}
+ if include_inactive:
+ query_params["include_inactive"] = "true"
+ if team_id:
+ query_params["team_id"] = team_id
+
+ # Use unified pagination function
+ paginated_result = await paginate_query(
+ db=db,
+ query=query,
+ page=page,
+ per_page=per_page,
+ cursor=None, # HTMX partials use page-based navigation
+ base_url=f"{settings.app_root_path}/admin/gateways/partial",
+ query_params=query_params,
+ use_cursor_threshold=False, # Disable auto-cursor switching for UI
+ )
+
+ # Extract paginated gateways (DbGateway objects)
+ gateways_db = paginated_result["data"]
+ pagination = paginated_result["pagination"]
+ links = paginated_result["links"]
+
+ # Batch fetch team names for the gateways to avoid N+1 queries
+ team_ids_set = {p.team_id for p in gateways_db if p.team_id}
+ team_map = {}
+ if team_ids_set:
+ teams = db.execute(select(EmailTeam.id, EmailTeam.name).where(EmailTeam.id.in_(team_ids_set), EmailTeam.is_active.is_(True))).all()
+ team_map = {team.id: team.name for team in teams}
+
+ # Apply team names to DB objects before conversion
+ for p in gateways_db:
+ p.team = team_map.get(p.team_id) if p.team_id else None
+
+ # Batch convert to Pydantic models using gateway service
+ # This eliminates the N+1 query problem from calling get_gateway_details() in a loop
+ gateways_pydantic = [gateway_service.convert_gateway_to_read(p) for p in gateways_db]
+
+ data = jsonable_encoder(gateways_pydantic)
+ base_url = f"{settings.app_root_path}/admin/gateways/partial"
+
+ # End the read-only transaction before template rendering to avoid idle-in-transaction timeouts.
+ db.commit()
+
+ LOGGER.info(f"🔷 GATEWAYS PARTIAL RESPONSE - Returning {len(data)} gateways, render mode: {render or 'default'}, team_id used in query: {team_id}")
+
+ if render == "controls":
+ return request.app.state.templates.TemplateResponse(
+ "pagination_controls.html",
+ {
+ "request": request,
+ "pagination": pagination.model_dump(),
+ "base_url": base_url,
+ "hx_target": "#gateways-table-body",
+ "hx_indicator": "#gateways-loading",
+ "query_params": query_params,
+ "root_path": request.scope.get("root_path", ""),
+ },
+ )
+
+ if render == "selector":
+ return request.app.state.templates.TemplateResponse(
+ "gateways_selector_items.html",
+ {"request": request, "data": data, "pagination": pagination.model_dump(), "root_path": request.scope.get("root_path", "")},
+ )
+
+ return request.app.state.templates.TemplateResponse(
+ "gateways_partial.html",
+ {
+ "request": request,
+ "data": data,
+ "pagination": pagination.model_dump(),
+ "links": links.model_dump() if links else None,
+ "root_path": settings.app_root_path,
"include_inactive": include_inactive,
},
)
@@ -8298,7 +8482,7 @@ async def admin_resources_partial_html(
"hx_target": "#resources-table-body",
"hx_indicator": "#resources-loading",
"query_params": query_params,
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
},
)
@@ -8309,7 +8493,7 @@ async def admin_resources_partial_html(
"request": request,
"data": data,
"pagination": pagination.model_dump(),
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
"gateway_id": gateway_id,
},
)
@@ -8321,7 +8505,7 @@ async def admin_resources_partial_html(
"data": data,
"pagination": pagination.model_dump(),
"links": links.model_dump() if links else None,
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
"include_inactive": include_inactive,
},
)
@@ -9682,7 +9866,7 @@ async def admin_delete_tool(tool_id: str, request: Request, db: Session = Depend
>>>
>>> async def test_admin_delete_tool_success():
... result = await admin_delete_tool(tool_id, mock_request_delete, mock_db, mock_user)
- ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin#tools" in result.headers["location"]
+ ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin/#tools" in result.headers["location"]
>>>
>>> asyncio.run(test_admin_delete_tool_success())
True
@@ -9730,6 +9914,9 @@ async def admin_delete_tool(tool_id: str, request: Request, db: Session = Depend
LOGGER.error(f"Error deleting tool: {e}")
error_message = "Failed to delete tool. Please try again."
+ form = await request.form()
+ is_inactive_checked = str(form.get("is_inactive_checked", "false"))
+ root_path = settings.app_root_path
root_path = request.scope.get("root_path", "")
# Build redirect URL with error message if present
@@ -9741,7 +9928,7 @@ async def admin_delete_tool(tool_id: str, request: Request, db: Session = Depend
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#tools", status_code=303)
- return RedirectResponse(f"{root_path}/admin#tools", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#tools", status_code=303)
@admin_router.post("/tools/{tool_id}/toggle")
@@ -9782,33 +9969,40 @@ async def admin_toggle_tool(
>>>
>>> # Happy path: Activate tool
>>> form_data_activate = FormData([("activate", "true"), ("is_inactive_checked", "false")])
- >>> mock_request_activate = MagicMock(spec=Request, scope={"root_path": ""})
+ >>> mock_request_activate = MagicMock(spec=Request)
>>> mock_request_activate.form = AsyncMock(return_value=form_data_activate)
>>> original_toggle_tool_status = tool_service.toggle_tool_status
>>> tool_service.toggle_tool_status = AsyncMock()
>>>
>>> async def test_admin_toggle_tool_activate():
... result = await admin_toggle_tool(tool_id, mock_request_activate, mock_db, mock_user)
- ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin#tools" in result.headers["location"]
+ ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin/#tools" in result.headers["location"]
>>>
>>> asyncio.run(test_admin_toggle_tool_activate())
True
>>>
>>> # Happy path: Deactivate tool
>>> form_data_deactivate = FormData([("activate", "false"), ("is_inactive_checked", "false")])
- >>> mock_request_deactivate = MagicMock(spec=Request, scope={"root_path": "/api"})
+ >>> mock_request_deactivate = MagicMock(spec=Request)
>>> mock_request_deactivate.form = AsyncMock(return_value=form_data_deactivate)
+ >>>
+ >>> # Mock settings.app_root_path
+ >>> import mcpgateway.admin as admin_module
+ >>> original_app_root_path = admin_module.settings.app_root_path
+ >>> admin_module.settings.app_root_path = "/api"
>>>
>>> async def test_admin_toggle_tool_deactivate():
... result = await admin_toggle_tool(tool_id, mock_request_deactivate, mock_db, mock_user)
- ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/api/admin#tools" in result.headers["location"]
+ ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/api/admin/#tools" in result.headers["location"]
>>>
>>> asyncio.run(test_admin_toggle_tool_deactivate())
True
+ >>> # Restore
+ >>> admin_module.settings.app_root_path = original_app_root_path
>>>
>>> # Edge case: Toggle with inactive checkbox checked
>>> form_data_inactive = FormData([("activate", "true"), ("is_inactive_checked", "true")])
- >>> mock_request_inactive = MagicMock(spec=Request, scope={"root_path": ""})
+ >>> mock_request_inactive = MagicMock(spec=Request)
>>> mock_request_inactive.form = AsyncMock(return_value=form_data_inactive)
>>>
>>> async def test_admin_toggle_tool_inactive_checked():
@@ -9820,7 +10014,7 @@ async def admin_toggle_tool(
>>>
>>> # Error path: Simulate an exception during toggle
>>> form_data_error = FormData([("activate", "true")])
- >>> mock_request_error = MagicMock(spec=Request, scope={"root_path": ""})
+ >>> mock_request_error = MagicMock(spec=Request)
>>> mock_request_error.form = AsyncMock(return_value=form_data_error)
>>> tool_service.toggle_tool_status = AsyncMock(side_effect=Exception("Toggle failed"))
>>>
@@ -9856,7 +10050,7 @@ async def admin_toggle_tool(
LOGGER.error(f"Error toggling tool status: {e}")
error_message = "Failed to toggle tool status. Please try again."
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
# Build redirect URL with error message if present
if error_message:
@@ -9867,7 +10061,7 @@ async def admin_toggle_tool(
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#tools", status_code=303)
- return RedirectResponse(f"{root_path}/admin#tools", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#tools", status_code=303)
@admin_router.get("/gateways/{gateway_id}", response_model=GatewayRead)
@@ -10597,33 +10791,40 @@ async def admin_delete_gateway(gateway_id: str, request: Request, db: Session =
>>>
>>> # Happy path: Delete gateway
>>> form_data_delete = FormData([("is_inactive_checked", "false")])
- >>> mock_request_delete = MagicMock(spec=Request, scope={"root_path": ""})
+ >>> mock_request_delete = MagicMock(spec=Request)
>>> mock_request_delete.form = AsyncMock(return_value=form_data_delete)
>>> original_delete_gateway = gateway_service.delete_gateway
>>> gateway_service.delete_gateway = AsyncMock()
>>>
>>> async def test_admin_delete_gateway_success():
... result = await admin_delete_gateway(gateway_id, mock_request_delete, mock_db, mock_user)
- ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin#gateways" in result.headers["location"]
+ ... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/admin/#gateways" in result.headers["location"]
>>>
>>> asyncio.run(test_admin_delete_gateway_success())
True
>>>
>>> # Edge case: Delete with inactive checkbox checked
>>> form_data_inactive = FormData([("is_inactive_checked", "true")])
- >>> mock_request_inactive = MagicMock(spec=Request, scope={"root_path": "/api"})
+ >>> mock_request_inactive = MagicMock(spec=Request)
>>> mock_request_inactive.form = AsyncMock(return_value=form_data_inactive)
>>>
+ >>> # Mock settings.app_root_path
+ >>> import mcpgateway.admin as admin_module
+ >>> original_app_root_path = admin_module.settings.app_root_path
+ >>> admin_module.settings.app_root_path = "/api"
+ >>>
>>> async def test_admin_delete_gateway_inactive_checked():
... result = await admin_delete_gateway(gateway_id, mock_request_inactive, mock_db, mock_user)
... return isinstance(result, RedirectResponse) and result.status_code == 303 and "/api/admin/?include_inactive=true#gateways" in result.headers["location"]
>>>
>>> asyncio.run(test_admin_delete_gateway_inactive_checked())
True
+ >>> # Restore
+ >>> admin_module.settings.app_root_path = original_app_root_path
>>>
>>> # Error path: Simulate an exception during deletion
>>> form_data_error = FormData([])
- >>> mock_request_error = MagicMock(spec=Request, scope={"root_path": ""})
+ >>> mock_request_error = MagicMock(spec=Request)
>>> mock_request_error.form = AsyncMock(return_value=form_data_error)
>>> gateway_service.delete_gateway = AsyncMock(side_effect=Exception("Deletion failed"))
>>>
@@ -10651,7 +10852,7 @@ async def admin_delete_gateway(gateway_id: str, request: Request, db: Session =
form = await request.form()
is_inactive_checked = str(form.get("is_inactive_checked", "false"))
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
# Build redirect URL with error message if present
if error_message:
@@ -10662,7 +10863,7 @@ async def admin_delete_gateway(gateway_id: str, request: Request, db: Session =
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#gateways", status_code=303)
- return RedirectResponse(f"{root_path}/admin#gateways", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#gateways", status_code=303)
@admin_router.get("/resources/test/{resource_uri:path}")
@@ -11212,6 +11413,9 @@ async def admin_delete_resource(resource_id: str, request: Request, db: Session
except Exception as e:
LOGGER.error(f"Error deleting resource: {e}")
error_message = "Failed to delete resource. Please try again."
+ form = await request.form()
+ is_inactive_checked: str = str(form.get("is_inactive_checked", "false"))
+ root_path = settings.app_root_path
root_path = request.scope.get("root_path", "")
# Build redirect URL with error message if present
@@ -11223,7 +11427,7 @@ async def admin_delete_resource(resource_id: str, request: Request, db: Session
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#resources", status_code=303)
- return RedirectResponse(f"{root_path}/admin#resources", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#resources", status_code=303)
@admin_router.post("/resources/{resource_id}/toggle")
@@ -11337,7 +11541,7 @@ async def admin_toggle_resource(
LOGGER.error(f"Error toggling resource status: {e}")
error_message = "Failed to toggle resource status. Please try again."
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
# Build redirect URL with error message if present
if error_message:
@@ -11348,7 +11552,7 @@ async def admin_toggle_resource(
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#resources", status_code=303)
- return RedirectResponse(f"{root_path}/admin#resources", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#resources", status_code=303)
@admin_router.get("/prompts/{prompt_id}")
@@ -11773,6 +11977,9 @@ async def admin_delete_prompt(prompt_id: str, request: Request, db: Session = De
except Exception as e:
LOGGER.error(f"Error deleting prompt: {e}")
error_message = "Failed to delete prompt. Please try again."
+ form = await request.form()
+ is_inactive_checked: str = str(form.get("is_inactive_checked", "false"))
+ root_path = settings.app_root_path
root_path = request.scope.get("root_path", "")
# Build redirect URL with error message if present
@@ -11784,7 +11991,7 @@ async def admin_delete_prompt(prompt_id: str, request: Request, db: Session = De
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#prompts", status_code=303)
- return RedirectResponse(f"{root_path}/admin#prompts", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#prompts", status_code=303)
@admin_router.post("/prompts/{prompt_id}/toggle")
@@ -11898,7 +12105,7 @@ async def admin_toggle_prompt(
LOGGER.error(f"Error toggling prompt status: {e}")
error_message = "Failed to toggle prompt status. Please try again."
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
# Build redirect URL with error message if present
if error_message:
@@ -11909,7 +12116,7 @@ async def admin_toggle_prompt(
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#prompts", status_code=303)
- return RedirectResponse(f"{root_path}/admin#prompts", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#prompts", status_code=303)
@admin_router.post("/roots")
@@ -11963,8 +12170,8 @@ async def admin_add_root(request: Request, user=Depends(get_current_user_with_pe
if isinstance(name_value, str):
name = name_value
await root_service.add_root(uri, name)
- root_path = request.scope.get("root_path", "")
- return RedirectResponse(f"{root_path}/admin#roots", status_code=303)
+ root_path = settings.app_root_path
+ return RedirectResponse(f"{root_path}/admin/#roots", status_code=303)
@admin_router.post("/roots/{uri:path}/delete")
@@ -12024,11 +12231,11 @@ async def admin_delete_root(uri: str, request: Request, user=Depends(get_current
LOGGER.debug(f"User {get_user_email(user)} is deleting root URI {uri}")
await root_service.remove_root(uri)
form = await request.form()
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
is_inactive_checked: str = str(form.get("is_inactive_checked", "false"))
if is_inactive_checked.lower() == "true":
return RedirectResponse(f"{root_path}/admin/?include_inactive=true#roots", status_code=303)
- return RedirectResponse(f"{root_path}/admin#roots", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#roots", status_code=303)
# Metrics
@@ -12190,7 +12397,7 @@ async def admin_metrics_partial_html(
"entity_type": entity_type,
"data": data,
"pagination": pagination.model_dump(),
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
},
)
@@ -14510,8 +14717,8 @@ async def admin_toggle_a2a_agent(
HTTPException: If A2A features are disabled
"""
if not a2a_service or not settings.mcpgateway_a2a_enabled:
- root_path = request.scope.get("root_path", "")
- return RedirectResponse(f"{root_path}/admin#a2a-agents", status_code=303)
+ root_path = settings.app_root_path
+ return RedirectResponse(f"{root_path}/admin/#a2a-agents", status_code=303)
error_message = None
try:
@@ -14522,29 +14729,29 @@ async def admin_toggle_a2a_agent(
user_email = get_user_email(user)
await a2a_service.toggle_agent_status(db, agent_id, activate, user_email=user_email)
- root_path = request.scope.get("root_path", "")
- return RedirectResponse(f"{root_path}/admin#a2a-agents", status_code=303)
+ root_path = settings.app_root_path
+ return RedirectResponse(f"{root_path}/admin/#a2a-agents", status_code=303)
except PermissionError as e:
LOGGER.warning(f"Permission denied for user {user_email} toggling A2A agent status{agent_id}: {e}")
error_message = str(e)
except A2AAgentNotFoundError as e:
LOGGER.error(f"A2A agent toggle failed - not found: {e}")
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
error_message = "A2A agent not found."
except Exception as e:
LOGGER.error(f"Error toggling A2A agent: {e}")
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
error_message = "Failed to toggle status of A2A agent. Please try again."
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
# Build redirect URL with error message if present
if error_message:
error_param = f"?error={urllib.parse.quote(error_message)}"
return RedirectResponse(f"{root_path}/admin/{error_param}#a2a-agents", status_code=303)
- return RedirectResponse(f"{root_path}/admin#a2a-agents", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#a2a-agents", status_code=303)
@admin_router.post("/a2a/{agent_id}/delete")
@@ -14569,8 +14776,8 @@ async def admin_delete_a2a_agent(
HTTPException: If A2A features are disabled
"""
if not a2a_service or not settings.mcpgateway_a2a_enabled:
- root_path = request.scope.get("root_path", "")
- return RedirectResponse(f"{root_path}/admin#a2a-agents", status_code=303)
+ root_path = settings.app_root_path
+ return RedirectResponse(f"{root_path}/admin/#a2a-agents", status_code=303)
form = await request.form()
purge_metrics = str(form.get("purge_metrics", "false")).lower() == "true"
@@ -14588,14 +14795,14 @@ async def admin_delete_a2a_agent(
LOGGER.error(f"Error deleting A2A agent: {e}")
error_message = "Failed to delete A2A agent. Please try again."
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
# Build redirect URL with error message if present
if error_message:
error_param = f"?error={urllib.parse.quote(error_message)}"
return RedirectResponse(f"{root_path}/admin/{error_param}#a2a-agents", status_code=303)
- return RedirectResponse(f"{root_path}/admin#a2a-agents", status_code=303)
+ return RedirectResponse(f"{root_path}/admin/#a2a-agents", status_code=303)
@admin_router.post("/a2a/{agent_id}/test")
@@ -15185,7 +15392,7 @@ async def get_plugins_partial(request: Request, db: Session = Depends(get_db), u
stats = await plugin_service.get_plugin_statistics()
# Prepare context for template
- context = {"request": request, "plugins": plugins, "stats": stats, "plugins_enabled": plugin_manager is not None, "root_path": request.scope.get("root_path", "")}
+ context = {"request": request, "plugins": plugins, "stats": stats, "plugins_enabled": plugin_manager is not None, "root_path": settings.app_root_path}
# Render the partial template
return request.app.state.templates.TemplateResponse("plugins_partial.html", context)
@@ -15668,7 +15875,7 @@ async def catalog_partial(
if not settings.mcpgateway_catalog_enabled:
raise HTTPException(status_code=404, detail="Catalog feature is disabled")
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
# Calculate pagination
page_size = settings.mcpgateway_catalog_page_size
@@ -15787,7 +15994,7 @@ async def get_system_stats(
{
"request": request,
"stats": stats,
- "root_path": request.scope.get("root_path", ""),
+ "root_path": settings.app_root_path,
"db_metrics_recording_enabled": settings.db_metrics_recording_enabled,
},
)
@@ -15952,7 +16159,7 @@ async def get_observability_partial(request: Request, _user=Depends(get_current_
Returns:
HTMLResponse: Rendered observability dashboard template
"""
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return request.app.state.templates.TemplateResponse("observability_partial.html", {"request": request, "root_path": root_path})
@@ -15968,7 +16175,7 @@ async def get_observability_metrics_partial(request: Request, _user=Depends(get_
Returns:
HTMLResponse: Rendered metrics dashboard template
"""
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return request.app.state.templates.TemplateResponse("observability_metrics.html", {"request": request, "root_path": root_path})
@@ -16105,7 +16312,7 @@ async def get_observability_traces(
# Get traces ordered by most recent
traces = query.order_by(ObservabilityTrace.start_time.desc()).limit(limit).all()
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return request.app.state.templates.TemplateResponse("observability_traces_list.html", {"request": request, "traces": traces, "root_path": root_path})
finally:
# Ensure close() always runs even if commit() fails
@@ -16138,7 +16345,7 @@ async def get_observability_trace_detail(request: Request, trace_id: str, _user=
if not trace:
raise HTTPException(status_code=404, detail="Trace not found")
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return request.app.state.templates.TemplateResponse("observability_trace_detail.html", {"request": request, "trace": trace, "root_path": root_path})
finally:
# Ensure close() always runs even if commit() fails
@@ -17465,7 +17672,7 @@ async def get_tools_partial(
Returns:
HTMLResponse: Rendered tool metrics dashboard partial
"""
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return request.app.state.templates.TemplateResponse(
"observability_tools.html",
{
@@ -17672,7 +17879,7 @@ async def get_prompts_partial(
Returns:
HTMLResponse: Rendered prompt metrics dashboard partial
"""
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return request.app.state.templates.TemplateResponse(
"observability_prompts.html",
{
@@ -17879,7 +18086,7 @@ async def get_resources_partial(
Returns:
HTMLResponse: Rendered resource metrics dashboard partial
"""
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
return request.app.state.templates.TemplateResponse(
"observability_resources.html",
{
diff --git a/mcpgateway/routers/oauth_router.py b/mcpgateway/routers/oauth_router.py
index eb3c68c64..7b6280a9d 100644
--- a/mcpgateway/routers/oauth_router.py
+++ b/mcpgateway/routers/oauth_router.py
@@ -260,7 +260,7 @@ async def oauth_callback(
try:
# Get root path for URL construction
- root_path = request.scope.get("root_path", "") if request else ""
+ root_path = settings.app_root_path
# Extract gateway_id from state parameter
# Try new base64-encoded JSON format first
diff --git a/mcpgateway/routers/sso.py b/mcpgateway/routers/sso.py
index 962ab1459..67d4c2aa3 100644
--- a/mcpgateway/routers/sso.py
+++ b/mcpgateway/routers/sso.py
@@ -202,7 +202,7 @@ async def handle_sso_callback(
raise HTTPException(status_code=404, detail="SSO authentication is disabled")
# Get root path for URL construction
- root_path = request.scope.get("root_path", "") if request else ""
+ root_path = settings.app_root_path
sso_service = SSOService(db)
@@ -228,7 +228,7 @@ async def handle_sso_callback(
# Third-Party
from fastapi.responses import RedirectResponse
- redirect_response = RedirectResponse(url=f"{root_path}/admin", status_code=302)
+ redirect_response = RedirectResponse(url=f"{root_path}/admin/", status_code=302)
# Set secure HTTP-only cookie using the same method as email auth
# First-Party
diff --git a/mcpgateway/utils/verify_credentials.py b/mcpgateway/utils/verify_credentials.py
index 69c1a28fb..7e98562fe 100644
--- a/mcpgateway/utils/verify_credentials.py
+++ b/mcpgateway/utils/verify_credentials.py
@@ -889,7 +889,7 @@ async def require_admin_auth(
accept_header = request.headers.get("accept", "")
if "text/html" in accept_header:
# Redirect browser to login page with error
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
raise HTTPException(status_code=status.HTTP_302_FOUND, detail="Admin privileges required", headers={"Location": f"{root_path}/admin/login?error=admin_required"})
else:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin privileges required")
@@ -906,7 +906,7 @@ async def require_admin_auth(
# For 401, check if we should redirect browser users
accept_header = request.headers.get("accept", "")
if "text/html" in accept_header:
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
raise HTTPException(status_code=status.HTTP_302_FOUND, detail="Authentication required", headers={"Location": f"{root_path}/admin/login"})
# If JWT auth fails, fall back to basic auth for backward compatibility
except Exception:
@@ -929,7 +929,7 @@ async def require_admin_auth(
accept_header = request.headers.get("accept", "")
is_htmx = request.headers.get("hx-request") == "true"
if "text/html" in accept_header or is_htmx:
- root_path = request.scope.get("root_path", "")
+ root_path = settings.app_root_path
raise HTTPException(status_code=status.HTTP_302_FOUND, detail="Authentication required", headers={"Location": f"{root_path}/admin/login"})
else:
raise HTTPException(