Coverage for src / mcp_server_langgraph / api / error_handlers.py: 77%

1""" 

2FastAPI exception handlers for custom exceptions. 

3 

4Provides structured error responses with proper HTTP status codes, 

5trace IDs, and user-friendly messages. 

6 

7See ADR-0029 for design rationale. 

8""" 

9 

10import logging 

11from typing import Any 

12 

13from fastapi import FastAPI, Request 

14from fastapi.responses import JSONResponse 

15 

16from mcp_server_langgraph.core.exceptions import MCPServerException 

17 

18logger = logging.getLogger(__name__) 

19 

20 

21def register_exception_handlers(app: FastAPI) -> None: 

22 """ 

23 Register custom exception handlers with FastAPI application. 

24 

25 Args: 

26 app: FastAPI application instance 

27 

28 Usage: 

29 from fastapi import FastAPI 

30 from mcp_server_langgraph.api.error_handlers import register_exception_handlers 

31 

32 app = FastAPI() 

33 register_exception_handlers(app) 

34 """ 

35 

36 @app.exception_handler(MCPServerException) 

37 async def mcp_exception_handler(request: Request, exc: MCPServerException) -> JSONResponse: 

38 """ 

39 Handle MCP server exceptions. 

40 

41 Provides: 

42 - Structured JSON error response 

43 - Trace ID in response headers 

44 - Metrics emission 

45 - Detailed logging with context 

46 """ 

47 # Log with full context 

48 logger.error( 

49 f"Exception: {exc.error_code}", 

50 extra={ 

51 "error_code": exc.error_code, 

52 "status_code": exc.status_code, 

53 "category": exc.category.value, 

54 "retry_policy": exc.retry_policy.value, 

55 "trace_id": exc.trace_id, 

56 "metadata": exc.metadata, 

57 "user_message": exc.user_message, 

58 "request_method": request.method, 

59 "request_url": str(request.url), 

60 }, 

61 ) 

62 

63 # Record metric 

64 try: 

65 from mcp_server_langgraph.observability.telemetry import error_counter 

66 

67 error_counter.add( 

68 1, 

69 attributes={ 

70 "error_code": exc.error_code, 

71 "status_code": str(exc.status_code), 

72 "category": exc.category.value, 

73 "retry_policy": exc.retry_policy.value, 

74 }, 

75 ) 

76 except Exception as metric_error: 

77 logger.warning(f"Failed to record error metric: {metric_error}") 

78 

79 # Build response headers 

80 headers = {} 

81 if exc.trace_id: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true

82 headers["X-Trace-ID"] = exc.trace_id 

83 

84 # Add Retry-After header for rate limit errors 

85 if exc.status_code == 429: 

86 retry_after = exc.metadata.get("retry_after", 60) 

87 headers["Retry-After"] = str(retry_after) 

88 

89 # Return JSON response 

90 return JSONResponse( 

91 status_code=exc.status_code, 

92 content=exc.to_dict(), 

93 headers=headers, 

94 ) 

95 

96 @app.exception_handler(Exception) 

97 async def generic_exception_handler(request: Request, exc: Exception) -> JSONResponse: 

98 """ 

99 Handle unexpected exceptions. 

100 

101 Wraps generic exceptions in InternalServerError for consistent handling. 

102 """ 

103 # Log unexpected error 

104 logger.error( 

105 f"Unexpected exception: {type(exc).__name__}", 

106 extra={ 

107 "exception_type": type(exc).__name__, 

108 "request_method": request.method, 

109 "request_url": str(request.url), 

110 }, 

111 ) 

112 

113 # Wrap in InternalServerError 

114 from mcp_server_langgraph.core.exceptions import UnexpectedError 

115 

116 wrapped_exc = UnexpectedError( 

117 message=str(exc), 

118 metadata={ 

119 "original_exception": type(exc).__name__, 

120 "request_method": request.method, 

121 "request_path": str(request.url.path), 

122 }, 

123 cause=exc, 

124 ) 

125 

126 return await mcp_exception_handler(request, wrapped_exc) 

127 

128 logger.info("Exception handlers registered") 

129 

130 

131def create_error_response( 

132 error_code: str, 

133 message: str, 

134 status_code: int = 500, 

135 metadata: dict[str, Any] | None = None, 

136 trace_id: str | None = None, 

137) -> dict[str, Any]: 

138 """ 

139 Create a standardized error response dictionary. 

140 

141 Args: 

142 error_code: Machine-readable error code 

143 message: Human-readable error message 

144 status_code: HTTP status code 

145 metadata: Additional error context 

146 trace_id: OpenTelemetry trace ID 

147 

148 Returns: 

149 Standardized error response dictionary 

150 

151 Usage: 

152 return create_error_response( 

153 error_code="auth.token_expired", 

154 message="Your session has expired", 

155 status_code=401, 

156 metadata={"user_id": "user_123"}, 

157 trace_id="abc123def456" 

158 ) 

159 """ 

160 return { 

161 "error": { 

162 "code": error_code, 

163 "message": message, 

164 "status_code": status_code, 

165 "metadata": metadata or {}, 

166 "trace_id": trace_id, 

167 } 

168 }