Added extra logs from stderr on commands on the /mcp list command. Allows more visibility for MCP connections.

pentestagent/interface/tui.py

@@ -552,6 +552,11 @@ class MCPScreen(ModalScreen):
             if mcp.last_error:
                 text.append(f"\nLast error: {mcp.last_error}\n", style="#e91b1b")
 
+            logs = mcp.get_logs()
+
+            if logs:
+                text.append(f"\n{logs}\n", style="#86e41a")
+
             desc_widget.update(text)
 
             toggle_btn.display = True

pentestagent/mcp/manager.py

@@ -66,6 +66,11 @@ class MCPServer:
     def disable(self):
         self.config.enabled = False
 
+    def get_logs(self) -> str:
+        if not self.transport:
+            return ""
+        return self.transport.get_logs()
+
 
 class MCPManager:
     """Manages MCP server connections and exposes tools to agents."""

pentestagent/mcp/transport.py

@@ -31,6 +31,10 @@ class MCPTransport(ABC):
         """Check if the transport is connected."""
         pass
 
+    @abstractmethod
+    async def get_logs(self) -> str:
+        pass
+
 
 class StdioTransport(MCPTransport):
     """MCP transport over stdio (for npx/uvx commands)."""
@@ -49,11 +53,28 @@ class StdioTransport(MCPTransport):
         self.env = env
         self.process: Optional[asyncio.subprocess.Process] = None
         self._lock = asyncio.Lock()
+        self.logstask = None
+        self.logs = ""
+
+    def get_logs(self) -> str:
+        return self.logs
 
     @property
     def is_connected(self) -> bool:
         """Check if the process is running."""
         return self.process is not None and self.process.returncode is None
+    
+    async def _read_stderr_loop(self):
+        try:
+            while True:
+                line = await self.process.stderr.readline()
+                if not line:
+                    break
+                self.logs += line.decode().rstrip() + "\n"
+        except asyncio.CancelledError:
+            # Optional: do any cleanup here
+            pass
+
 
     async def connect(self):
         """Start the MCP server process."""
@@ -84,6 +105,9 @@ class StdioTransport(MCPTransport):
             limit=1024 * 1024,  # 1MB buffer limit for large MCP responses
         )
 
+        self.logstask = asyncio.create_task(self._read_stderr_loop())
+
+
     async def send(self, message: dict, timeout: float = 15.0) -> dict:
         """
         Send a JSON-RPC message and wait for response.
@@ -129,6 +153,15 @@ class StdioTransport(MCPTransport):
         if not self.process:
             return
 
+        if self.logstask:
+
+            self.logstask.cancel()
+
+            try:
+                await self.logstask
+            except asyncio.CancelledError:
+                pass  # Task was successfully cancelled
+
         proc = self.process
         self.process = None
 
@@ -180,6 +213,9 @@ class SSETransport(MCPTransport):
         """Check if the session is active."""
         return self._connected and self.session is not None
 
+    def get_logs(self) -> str:
+        return ""
+
     async def connect(self):
         """Connect to the SSE endpoint."""
         try: