Coverage for src/mcp_internal/webserver_manager.py: 77%

1""" 

2Webserver Manager Module 

3 

4Handles web server lifecycle management, port finding, and browser opening. 

5""" 

6 

7from typing import Dict, Any 

8import sys 

9import socket 

10import threading 

11import webbrowser 

12import uvicorn 

13 

14 

15class WebserverManager: 

16 """Manages web server lifecycle and configuration""" 

17 

18 def __init__(self, server: 'MCPDocumentationServer'): 

19 """ 

20 Initialize WebserverManager with reference to server instance 

21 

22 Args: 

23 server: MCPDocumentationServer instance for accessing shared state 

24 """ 

25 self.server = server 

26 self.webserver_url = None 

27 self.webserver_port = None 

28 self.webserver_thread = None 

29 self.webserver_started = False 

30 

31 def find_free_port(self, start_port: int = 8080) -> int: 

32 """Find first available port using OS-assigned port for tests""" 

33 import os 

34 

35 # For tests, always use OS-assigned port to avoid conflicts 

36 if 'pytest' in os.environ.get('_', '') or 'PYTEST_CURRENT_TEST' in os.environ: 

37 try: 

38 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

39 sock.bind(('localhost', 0)) # Let OS assign port 

40 port = sock.getsockname()[1] 

41 sock.close() 

42 return port 

43 except OSError: 

44 pass 

45 

46 # For normal operation, try preferred range first 

47 for port in range(start_port, start_port + 20): 

48 try: 

49 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

50 sock.bind(('localhost', port)) 

51 sock.close() 

52 return port 

53 except OSError: 

54 continue 

55 

56 # Final fallback: OS-assigned port 

57 try: 

58 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

59 sock.bind(('localhost', 0)) 

60 port = sock.getsockname()[1] 

61 sock.close() 

62 return port 

63 except OSError: 

64 return start_port 

65 

66 def start_webserver_thread(self): 

67 """Start webserver in background thread (like Serena)""" 

68 # Find free port 

69 self.webserver_port = self.find_free_port() 

70 self.webserver_url = f"http://localhost:{self.webserver_port}" 

71 

72 # Start uvicorn in daemon thread 

73 def run_server(): 

74 import sys 

75 import os 

76 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) 

77 import web_server 

78 # Set the global doc_server 

79 web_server.doc_server = self.server 

80 

81 # Configure uvicorn (disable logging config for MCP compatibility) 

82 config = uvicorn.Config( 

83 web_server.app, 

84 host="127.0.0.1", 

85 port=self.webserver_port, 

86 log_config=None, # Disable default logging (incompatible with MCP stdio) 

87 access_log=False 

88 ) 

89 server = uvicorn.Server(config) 

90 server.run() 

91 

92 self.webserver_thread = threading.Thread(target=run_server, daemon=True) 

93 self.webserver_thread.start() 

94 

95 # Don't wait - let it start in background 

96 print(f"✅ Webserver starting at {self.webserver_url}", file=sys.stderr) 

97 

98 # Open browser after short delay (non-blocking) 

99 def open_browser_delayed(): 

100 import time 

101 time.sleep(1.5) 

102 try: 

103 webbrowser.open(self.webserver_url) 

104 print(f"🚀 Browser opened", file=sys.stderr) 

105 except Exception as e: 

106 print(f"⚠️ Could not open browser: {e}", file=sys.stderr) 

107 

108 browser_thread = threading.Thread(target=open_browser_delayed, daemon=True) 

109 browser_thread.start() 

110 

111 self.webserver_started = True 

112 

113 def get_webserver_status(self) -> Dict[str, Any]: 

114 """Get current webserver status""" 

115 if not self.server.enable_webserver: 

116 return { 

117 'enabled': False, 

118 'running': False, 

119 'port': None, 

120 'url': None, 

121 'browser_opened': False, 

122 'warnings': [] 

123 } 

124 

125 return { 

126 'enabled': True, 

127 'running': self.webserver_url is not None, 

128 'port': self.webserver_port, 

129 'url': self.webserver_url, 

130 'browser_opened': self.webserver_url is not None, # Assume success if URL exists 

131 'warnings': [] 

132 } 

133 

134 def restart_webserver(self) -> bool: 

135 """Restart webserver on potentially different port""" 

136 if not self.server.enable_webserver: 

137 return False 

138 

139 try: 

140 # Note: Current implementation doesn't support graceful shutdown 

141 # This would need to be implemented if restart is needed 

142 print("Webserver restart not fully implemented - requires graceful shutdown", file=sys.stderr) 

143 return False 

144 

145 except Exception as e: 

146 print(f"Error restarting webserver: {e}", file=sys.stderr) 

147 return False