Coverage for mcpgateway / transports / base.py: 100%

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

2"""Location: ./mcpgateway/transports/base.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Mihai Criveti 

6 

7Base Transport Interface. 

8This module defines the base protocol for MCP transports. 

9""" 

10 

11# Standard 

12from abc import ABC, abstractmethod 

13from typing import Any, AsyncGenerator, Dict 

14 

15 

16class Transport(ABC): 

17 """Base class for MCP transport implementations. 

18 

19 This abstract base class defines the interface that all MCP transport 

20 implementations must follow. It provides the core methods for connection 

21 management and message exchange. 

22 

23 Examples: 

24 >>> # Transport is abstract and cannot be instantiated directly 

25 >>> try: 

26 ... Transport() 

27 ... except TypeError as e: 

28 ... print("Cannot instantiate abstract class") 

29 Cannot instantiate abstract class 

30 

31 >>> # Check if Transport is an abstract base class 

32 >>> from abc import ABC 

33 >>> issubclass(Transport, ABC) 

34 True 

35 

36 >>> # Verify abstract methods are defined 

37 >>> hasattr(Transport, 'connect') 

38 True 

39 >>> hasattr(Transport, 'disconnect') 

40 True 

41 >>> hasattr(Transport, 'send_message') 

42 True 

43 >>> hasattr(Transport, 'receive_message') 

44 True 

45 >>> hasattr(Transport, 'is_connected') 

46 True 

47 """ 

48 

49 @abstractmethod 

50 async def connect(self) -> None: 

51 """Initialize transport connection. 

52 

53 This method should establish the underlying connection for the transport. 

54 It must be called before sending or receiving messages. 

55 

56 Examples: 

57 >>> # This is an abstract method - implementation required in subclasses 

58 >>> import inspect 

59 >>> inspect.ismethod(Transport.connect) 

60 False 

61 >>> hasattr(Transport, 'connect') 

62 True 

63 """ 

64 

65 @abstractmethod 

66 async def disconnect(self) -> None: 

67 """Close transport connection. 

68 

69 This method should clean up the underlying connection and any associated 

70 resources. It should be called when the transport is no longer needed. 

71 

72 Examples: 

73 >>> # This is an abstract method - implementation required in subclasses 

74 >>> import inspect 

75 >>> inspect.ismethod(Transport.disconnect) 

76 False 

77 >>> hasattr(Transport, 'disconnect') 

78 True 

79 """ 

80 

81 @abstractmethod 

82 async def send_message(self, message: Dict[str, Any]) -> None: 

83 """Send a message over the transport. 

84 

85 Args: 

86 message: Message to send 

87 

88 Examples: 

89 >>> # This is an abstract method - implementation required in subclasses 

90 >>> import inspect 

91 >>> inspect.ismethod(Transport.send_message) 

92 False 

93 >>> hasattr(Transport, 'send_message') 

94 True 

95 """ 

96 

97 @abstractmethod 

98 async def receive_message(self) -> AsyncGenerator[Dict[str, Any], None]: 

99 """Receive messages from the transport. 

100 

101 Yields: 

102 Received messages 

103 

104 Examples: 

105 >>> # This is an abstract method - implementation required in subclasses 

106 >>> import inspect 

107 >>> inspect.ismethod(Transport.receive_message) 

108 False 

109 >>> hasattr(Transport, 'receive_message') 

110 True 

111 """ 

112 

113 @abstractmethod 

114 async def is_connected(self) -> bool: 

115 """Check if transport is connected. 

116 

117 Returns: 

118 True if connected 

119 

120 Examples: 

121 >>> # This is an abstract method - implementation required in subclasses 

122 >>> import inspect 

123 >>> inspect.ismethod(Transport.is_connected) 

124 False 

125 >>> hasattr(Transport, 'is_connected') 

126 True 

127 """