Coverage for session_buddy / tools / health_tools.py: 76.27%

1"""Health check MCP tools for session-mgmt-mcp. 

2 

3Provides health status endpoints compatible with Docker and Kubernetes 

4orchestration systems. 

5 

6Phase 10.1: Production Hardening - Health Check Tools 

7""" 

8 

9from __future__ import annotations 

10 

11import time 

12import typing as t 

13 

14from mcp_common.health import HealthCheckResponse 

15 

16# Server start time for uptime calculation 

17_SERVER_START_TIME = time.time() 

18 

19 

20def _normalize_dict_component( 

21 component: dict[str, t.Any], 

22) -> t.Any: # Returns MCPComponentHealth 

23 """Normalize dictionary component to MCPComponentHealth.""" 

24 from mcp_common.health import ComponentHealth as MCPComponentHealth 

25 from mcp_common.health import HealthStatus as MCPHealthStatus 

26 

27 status_value = component.get("status", "degraded") 

28 try: 

29 status = MCPHealthStatus(status_value) 

30 except ValueError: 

31 status = MCPHealthStatus.DEGRADED 

32 

33 return MCPComponentHealth( 

34 name=component.get("name", "unknown"), 

35 status=status, 

36 message=component.get("message"), 

37 latency_ms=component.get("latency_ms"), 

38 metadata=component.get("metadata", {}), 

39 ) 

40 

41 

42def _normalize_object_component( 

43 component: t.Any, 

44) -> t.Any: # Returns MCPComponentHealth 

45 """Normalize object component to MCPComponentHealth.""" 

46 from mcp_common.health import ComponentHealth as MCPComponentHealth 

47 from mcp_common.health import HealthStatus as MCPHealthStatus 

48 

49 status_attr = getattr(component, "status", MCPHealthStatus.DEGRADED) 

50 try: 

51 status = ( 

52 status_attr 

53 if isinstance(status_attr, MCPHealthStatus) 

54 else MCPHealthStatus(str(status_attr)) 

55 ) 

56 except ValueError: 

57 status = MCPHealthStatus.DEGRADED 

58 

59 return MCPComponentHealth( 

60 name=getattr(component, "name", "unknown"), 

61 status=status, 

62 message=getattr(component, "message", None), 

63 latency_ms=getattr(component, "latency_ms", None), 

64 metadata=getattr(component, "metadata", {}), 

65 ) 

66 

67 

68def _normalize_components( 

69 components: list[t.Any], 

70) -> list[t.Any]: # Returns list[MCPComponentHealth] 

71 """Normalize health check components to standard format.""" 

72 from mcp_common.health import ComponentHealth as MCPComponentHealth 

73 

74 normalized: list[t.Any] = [] # list[MCPComponentHealth] 

75 for component in components: 

76 if isinstance(component, MCPComponentHealth): 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 normalized.append(component) 

78 elif isinstance(component, dict): 78 ↛ 81line 78 didn't jump to line 81 because the condition on line 78 was always true

79 normalized.append(_normalize_dict_component(component)) 

80 else: 

81 normalized.append(_normalize_object_component(component)) 

82 

83 return normalized 

84 

85 

86def _prepare_readiness_result( 

87 response: t.Any, # HealthCheckResponse 

88) -> dict[str, t.Any]: 

89 """Prepare result for readiness check.""" 

90 result: dict[str, t.Any] = response.to_dict() 

91 result["ready"] = response.is_healthy() 

92 return result 

93 

94 

95def _prepare_liveness_result( 

96 response: t.Any, # HealthCheckResponse 

97) -> dict[str, t.Any]: 

98 """Prepare result for liveness check.""" 

99 result: dict[str, t.Any] = response.to_dict() 

100 result["alive"] = response.is_ready() 

101 return result 

102 

103 

104async def get_health_status(ready: bool = False) -> dict[str, t.Any]: 

105 """Get comprehensive health status of the session management server. 

106 

107 Args: 

108 ready: If True, use readiness check logic (stricter, for K8s readiness probes) 

109 If False, use liveness check logic (looser, for K8s liveness probes) 

110 

111 Returns: 

112 Dictionary with health status suitable for JSON serialization 

113 

114 Example Response: 

115 { 

116 "status": "healthy", 

117 "timestamp": "2025-10-28T12:00:00Z", 

118 "version": "1.0.0", 

119 "uptime_seconds": 3600.5, 

120 "components": [ 

121 { 

122 "name": "database", 

123 "status": "healthy", 

124 "message": "Database operational", 

125 "latency_ms": 12.5 

126 }, 

127 ... 

128 ] 

129 } 

130 

131 Usage: 

132 # Kubernetes liveness probe (checks if server should be restarted) 

133 await get_health_status(ready=False) 

134 

135 # Kubernetes readiness probe (checks if server should receive traffic) 

136 await get_health_status(ready=True) 

137 

138 # Docker health check 

139 await get_health_status() 

140 

141 """ 

142 from session_buddy.health_checks import get_all_health_checks 

143 

144 # Get server version 

145 try: 

146 from session_buddy import __version__ 

147 

148 version = __version__ 

149 except (ImportError, AttributeError): 

150 version = "unknown" 

151 

152 # Run all health checks and normalize 

153 components = await get_all_health_checks() 

154 normalized_components = _normalize_components(components) 

155 

156 # Create health response 

157 response = HealthCheckResponse.create( 

158 components=normalized_components, 

159 version=version, 

160 start_time=_SERVER_START_TIME, 

161 metadata={"check_type": "readiness" if ready else "liveness"}, 

162 ) 

163 

164 # Return appropriate result based on check type 

165 return ( 

166 _prepare_readiness_result(response) 

167 if ready 

168 else _prepare_liveness_result(response) 

169 ) 

170 

171 

172__all__ = ["get_health_status"]