feat(mcp): add MCP server log panel backend (#2308)

- Add log buffer (_log_buffer: deque with maxlen=500) to RuntimeMCPSession
- Capture stderr from Box managed process in monitor_process_health loop
- Add get_mcp_server_logs service method with limit and level filtering
- Add GET /servers/<server_name>/logs HTTP endpoint
- Parse log level from stderr lines (error/warning/debug/info)
- Tests passing: 211 passed, 12 skipped

Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>

src/langbot/pkg/api/http/controller/groups/resources/mcp.py

@@ -96,6 +96,19 @@ class MCPRouterGroup(group.RouterGroup):
             except Exception as e:
                 return self.http_status(500, -1, f'Failed to get resource templates: {str(e)}')
 
+        @self.route('/servers/<path:server_name>/logs', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
+        async def _(server_name: str) -> str:
+            """Get logs from an MCP server"""
+            server_name = unquote(server_name)
+            try:
+                limit = int(quart.request.args.get('limit', 200))
+            except (TypeError, ValueError):
+                limit = 200
+            limit = min(limit, 500)
+            level = quart.request.args.get('level') or None
+            logs = await self.ap.mcp_service.get_mcp_server_logs(server_name, limit=limit, level=level)
+            return self.success(data={'logs': logs})
+
         @self.route('/servers/<path:server_name>/resources/read', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
         async def _(server_name: str) -> str:
             """Read a resource from an MCP server"""

src/langbot/pkg/api/http/service/mcp.py

@@ -244,3 +244,19 @@ class MCPService:
             context=ctx,
         )
         return wrapper.id
+
+    async def get_mcp_server_logs(self, server_name: str, limit: int = 200, level: str | None = None) -> list[dict]:
+        """Get recent log lines captured from the MCP server's stderr."""
+        session = self.ap.tool_mgr.mcp_tool_loader.get_session(server_name)
+        if not session:
+            return []
+
+        # Get logs from the session's buffer
+        logs = list(session._log_buffer)
+
+        # Filter by level if specified
+        if level:
+            logs = [log for log in logs if log.get('level') == level]
+
+        # Return the most recent 'limit' logs
+        return logs[-limit:]

src/langbot/pkg/provider/tools/loaders/mcp.py

@@ -268,6 +268,13 @@ class RuntimeMCPSession:
         # process (it will be re-attached on the next initialize()).
         self._preserve_managed_process = False
 
+        # Log buffer for capturing stderr from Box managed process (maxlen=500 keeps
+        # recent lines without unbounded memory growth)
+        import collections as _collections
+
+        self._log_buffer: _collections.deque = _collections.deque(maxlen=500)
+        self._last_stderr_text: str = ''
+
         self._box_stdio_runtime = BoxStdioSessionRuntime(self)
         self.box_config = self._box_stdio_runtime.config
 

src/langbot/pkg/provider/tools/loaders/mcp_stdio.py

@@ -297,6 +297,33 @@ class BoxStdioSessionRuntime:
                 )
                 if consecutive_errors >= self.owner._MONITOR_MAX_CONSECUTIVE_ERRORS:
                     return
+
+            # Capture stderr logs from the managed process
+            if isinstance(info, dict):
+                stderr_text = info.get('stderr', '') or info.get('stderr_preview', '')
+            else:
+                stderr_text = getattr(info, 'stderr', '') or getattr(info, 'stderr_preview', '')
+
+            if stderr_text and stderr_text != self.owner._last_stderr_text:
+                # Find new lines not in the previous snapshot
+                old_lines = set(self.owner._last_stderr_text.splitlines()) if self.owner._last_stderr_text else set()
+                new_lines = [l for l in stderr_text.splitlines() if l and l not in old_lines]
+                self.owner._last_stderr_text = stderr_text
+
+                import time as _time
+
+                for line in new_lines:
+                    level = (
+                        'error'
+                        if any(k in line.upper() for k in ('ERROR', 'CRITICAL'))
+                        else 'warning'
+                        if 'WARNING' in line.upper()
+                        else 'debug'
+                        if 'DEBUG' in line.upper()
+                        else 'info'
+                    )
+                    self.owner._log_buffer.append({'ts': _time.time(), 'level': level, 'text': line})
+
             await asyncio.sleep(self.owner._MONITOR_POLL_INTERVAL)
 
     async def _managed_process_is_running(self) -> bool: