Coverage for mcpgateway / plugins / framework / observability.py: 100%

1# -*- coding: utf-8 -*- 

2"""Location: ./mcpgateway/plugins/framework/observability.py 

3Copyright 2026 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Fred Araujo 

6 

7Observability abstractions for the plugin framework. 

8 

9Provides a protocol-based interface for observability so that host 

10applications can inject their own tracing implementation. 

11""" 

12 

13# Standard 

14from contextvars import ContextVar 

15from typing import Any, Dict, Optional, Protocol 

16 

17# Context variable for tracking the current trace_id across async calls. 

18# NOTE: This is bridged from mcpgateway.services.observability_service.current_trace_id 

19# by ObservabilityMiddleware. Both must be set together; see the middleware for details. 

20current_trace_id: ContextVar[Optional[str]] = ContextVar("current_trace_id", default=None) 

21 

22 

23class ObservabilityProvider(Protocol): 

24 """Interface for observability - host application implements this.""" 

25 

26 def start_span( 

27 self, 

28 trace_id: str, 

29 name: str, 

30 kind: str = "internal", 

31 resource_type: Optional[str] = None, 

32 resource_name: Optional[str] = None, 

33 attributes: Optional[Dict[str, Any]] = None, 

34 ) -> Optional[str]: 

35 """Start a new span within a trace. 

36 

37 Args: 

38 trace_id: The trace identifier. 

39 name: The span name. 

40 kind: The span kind (e.g. "internal", "client", "server"). 

41 resource_type: Optional resource type being traced. 

42 resource_name: Optional resource name being traced. 

43 attributes: Optional key-value attributes for the span. 

44 """ 

45 ... # pylint: disable=unnecessary-ellipsis 

46 

47 def end_span( 

48 self, 

49 span_id: Optional[str], 

50 status: str = "ok", 

51 attributes: Optional[Dict[str, Any]] = None, 

52 ) -> None: 

53 """End a previously started span. 

54 

55 Args: 

56 span_id: The span identifier returned by start_span. 

57 status: The span status (e.g. "ok", "error"). 

58 attributes: Optional additional attributes to attach. 

59 """ 

60 ... # pylint: disable=unnecessary-ellipsis 

61 

62 

63class NullObservability: 

64 """Default no-op implementation for standalone operation.""" 

65 

66 def start_span( # pylint: disable=unused-argument 

67 self, 

68 trace_id: str, 

69 name: str, 

70 kind: str = "internal", 

71 resource_type: Optional[str] = None, 

72 resource_name: Optional[str] = None, 

73 attributes: Optional[Dict[str, Any]] = None, 

74 ) -> Optional[str]: 

75 """No-op span start for standalone operation. 

76 

77 Args: 

78 trace_id: The trace identifier. 

79 name: The span name. 

80 kind: The span kind. 

81 resource_type: Optional resource type. 

82 resource_name: Optional resource name. 

83 attributes: Optional span attributes. 

84 

85 Returns: 

86 Always None (no-op implementation). 

87 """ 

88 return None 

89 

90 def end_span( # pylint: disable=unused-argument 

91 self, 

92 span_id: Optional[str], 

93 status: str = "ok", 

94 attributes: Optional[Dict[str, Any]] = None, 

95 ) -> None: 

96 """No-op span end for standalone operation. 

97 

98 Args: 

99 span_id: The span identifier. 

100 status: The span status. 

101 attributes: Optional span attributes. 

102 """