Replace loguru during tui

Author: Drummersbrother

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "open-edison"
-version = "0.1.53"
+version = "0.1.54"
 description = "Open-source MCP security, aggregation, and monitoring. Single-user, self-hosted MCP proxy."
 readme = "README.md"
 authors = [

src/mcp_importer/api.py

@@ -8,7 +8,8 @@
 from fastmcp import Client as FastMCPClient
 from fastmcp import FastMCP
 from fastmcp.client.auth.oauth import FileTokenStorage
-from loguru import logger as log
+import questionary
+from loguru import logger as log  # kept for non-TUI contexts; printing used in TUI flows
 
 from src.config import Config, MCPServerConfig, get_config_json_path
 from src.mcp_importer import paths as _paths
@@ -169,9 +170,9 @@ async def _verify_async() -> bool:  # noqa: C901
                             and no_tokens
                             and not has_inline_headers
                         ):
-                            log.info(
-                                "Skipping verification for remote server '{}' pending OAuth",
-                                server.name,
+                            questionary.print(
+                                f"Skipping verification for remote server '{server.name}' pending OAuth",
+                                style="bold fg:ansiyellow",
                             )
                             return True
         except Exception:
@@ -249,16 +250,20 @@ async def _list_tools_only() -> Any:
                         return ping_succeeded
                 except TimeoutError:
                     if ping_succeeded:
-                        log.warning(
-                            f"Ping received from '{server.name}' but shutting down client timed out. Marking as success."
+                        questionary.print(
+                            f"Ping received from '{server.name}' but shutdown timed out (treating as success)",
+                            style="bold fg:ansiyellow",
                         )
                     else:
-                        log.error(
-                            f"MCP remote verification timed out (more than {connection_timeout}s) for '{server.name}', marking as failure."
+                        questionary.print(
+                            f"Verification timed out (> {connection_timeout}s) for '{server.name}'",
+                            style="bold fg:ansired",
                         )
                     return ping_succeeded
                 except Exception as e:  # noqa: BLE001
-                    log.error("MCP remote verification failed for '{}': {}", server.name, e)
+                    questionary.print(
+                        f"Verification failed for '{server.name}': {e}", style="bold fg:ansired"
+                    )
                     return False
 
         # Local/stdio servers: mount via proxy and perform a single light operation (tools only)
@@ -286,7 +291,9 @@ async def _list_tools_only() -> Any:
             await asyncio.wait_for(_list_tools_only(), timeout=30.0)
             return True
         except Exception as e:
-            log.error("MCP verification failed for '{}': {}", server.name, e)
+            questionary.print(
+                f"Verification failed for '{server.name}': {e}", style="bold fg:ansired"
+            )
             return False
         finally:
             for obj in (host_local, proxy_local):

src/setup_tui/main.py

@@ -2,7 +2,9 @@
 import asyncio
 
 import questionary
+from loguru import logger as log
 
+import src.oauth_manager as oauth_mod
 from src.config import MCPServerConfig
 from src.mcp_importer.api import (
     CLIENT,
@@ -170,6 +172,35 @@ def show_manual_setup_screen() -> None:
 
 def run(*, dry_run: bool = False, skip_oauth: bool = False) -> None:
     """Run the complete setup process."""
+    # Suppress loguru output for a cleaner TUI experience
+    import contextlib
+
+    with contextlib.suppress(Exception):
+        log.remove()
+
+    # Route oauth_manager's log calls to questionary for TUI output
+    class _TuiLogger:
+        def _fmt(self, msg: object, *args: object) -> str:
+            try:
+                if isinstance(msg, str) and args:
+                    return msg.format(*args)
+            except Exception:
+                pass
+            return str(msg)
+
+        def info(self, msg: object, *args: object, **kwargs: object) -> None:
+            questionary.print(self._fmt(msg, *args), style="fg:ansiblue")
+
+        def debug(self, msg: object, *args: object, **kwargs: object) -> None:
+            questionary.print(self._fmt(msg, *args), style="fg:ansiblack")
+
+        def warning(self, msg: object, *args: object, **kwargs: object) -> None:
+            questionary.print(self._fmt(msg, *args), style="bold fg:ansiyellow")
+
+        def error(self, msg: object, *args: object, **kwargs: object) -> None:
+            questionary.print(self._fmt(msg, *args), style="bold fg:ansired")
+
+    oauth_mod.log = _TuiLogger()  # type: ignore[attr-defined]
     show_welcome_screen(dry_run=dry_run)
     # Additional setup steps will be added here