Coverage for src / lexigram / contracts / mcp / protocols.py: 100%

1"""MCP protocol definitions. 

2 

3These protocols define the structural contracts for MCP server 

4components. Packages can type-hint against these without importing 

5the concrete implementations from ``lexigram-mcp``. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import Any, Protocol, runtime_checkable 

11 

12 

13@runtime_checkable 

14class MCPToolProviderProtocol(Protocol): 

15 """Protocol for providing tools to the MCP server. 

16 

17 Satisfied by: 

18 - ``ToolRegistryAdapter`` (bridges lexigram-agents ToolRegistry) 

19 - Any custom tool provider 

20 """ 

21 

22 async def list_tools(self) -> list[dict[str, Any]]: 

23 """List all available tools in MCP format. 

24 

25 Returns list of dicts with ``name``, ``description``, 

26 ``inputSchema`` keys. 

27 """ 

28 ... 

29 

30 async def call_tool( 

31 self, 

32 name: str, 

33 arguments: dict[str, Any], 

34 ) -> dict[str, Any]: 

35 """Execute a tool by name with given arguments.""" 

36 ... 

37 

38 

39@runtime_checkable 

40class MCPResourceProviderProtocol(Protocol): 

41 """Protocol for providing resources to the MCP server. 

42 

43 Resources are data that AI clients can browse and read — 

44 database records, files, configuration, etc. 

45 """ 

46 

47 async def list_resources(self) -> list[dict[str, Any]]: 

48 """List available resources in MCP format.""" 

49 ... 

50 

51 async def read_resource(self, uri: str) -> dict[str, Any]: 

52 """Read a resource by URI.""" 

53 ... 

54 

55 

56@runtime_checkable 

57class MCPPromptProviderProtocol(Protocol): 

58 """Protocol for providing prompt templates to the MCP server.""" 

59 

60 async def list_prompts(self) -> list[dict[str, Any]]: 

61 """List available prompt templates.""" 

62 ... 

63 

64 async def get_prompt( 

65 self, 

66 name: str, 

67 arguments: dict[str, Any] | None = None, 

68 ) -> dict[str, Any]: 

69 """Get a prompt template with arguments filled.""" 

70 ... 

71 

72 

73@runtime_checkable 

74class MCPTransportProtocol(Protocol): 

75 """Protocol for MCP transport implementations. 

76 

77 A transport handles the I/O layer — reading requests and 

78 writing responses. The MCP server is transport-agnostic. 

79 """ 

80 

81 async def start(self) -> None: 

82 """Start the transport.""" 

83 ... 

84 

85 async def stop(self) -> None: 

86 """Stop the transport.""" 

87 ... 

88 

89 

90@runtime_checkable 

91class MCPServerProtocol(Protocol): 

92 """Protocol for the MCP server. 

93 

94 The server routes JSON-RPC messages to handlers and returns 

95 JSON-RPC responses. 

96 """ 

97 

98 async def handle_message( 

99 self, 

100 message: dict[str, Any], 

101 ) -> dict[str, Any] | None: 

102 """Handle a JSON-RPC message and return the response. 

103 

104 Returns None for notifications (no response expected). 

105 """ 

106 ... 

107 

108 

109@runtime_checkable 

110class MCPToolHandlerProtocol(Protocol): 

111 """Protocol for handling MCP tool-related methods. 

112 

113 Handles tools/list and tools/call methods. 

114 """ 

115 

116 async def list_tools(self) -> list[dict[str, Any]]: 

117 """Handle tools/list method. 

118 

119 Returns list of available tools with their definitions. 

120 """ 

121 ... 

122 

123 async def call_tool( 

124 self, 

125 name: str, 

126 arguments: dict[str, Any], 

127 ) -> dict[str, Any]: 

128 """Handle tools/call method. 

129 

130 Executes a tool and returns the result. 

131 """ 

132 ... 

133 

134 

135@runtime_checkable 

136class MCPResourceHandlerProtocol(Protocol): 

137 """Protocol for handling MCP resource-related methods. 

138 

139 Handles resources/list, resources/read, and resources/templates/list. 

140 """ 

141 

142 async def list_resources(self) -> list[dict[str, Any]]: 

143 """Handle resources/list method. 

144 

145 Returns list of available resources. 

146 """ 

147 ... 

148 

149 async def read_resource(self, uri: str) -> dict[str, Any]: 

150 """Handle resources/read method. 

151 

152 Returns content of a specific resource. 

153 """ 

154 ... 

155 

156 async def list_templates(self) -> list[dict[str, Any]]: 

157 """Handle resources/templates/list method. 

158 

159 Returns list of URI templates for resources. 

160 """ 

161 ... 

162 

163 

164@runtime_checkable 

165class MCPPromptHandlerProtocol(Protocol): 

166 """Protocol for handling MCP prompt-related methods. 

167 

168 Handles prompts/list and prompts/get methods. 

169 """ 

170 

171 async def list_prompts(self) -> list[dict[str, Any]]: 

172 """Handle prompts/list method. 

173 

174 Returns list of available prompt templates. 

175 """ 

176 ... 

177 

178 async def get_prompt( 

179 self, 

180 name: str, 

181 arguments: dict[str, Any] | None = None, 

182 ) -> dict[str, Any]: 

183 """Handle prompts/get method. 

184 

185 Returns a prompt with arguments filled in. 

186 """ 

187 ... 

188 

189 

190@runtime_checkable 

191class MCPAuthorizerProtocol(Protocol): 

192 """Authorize an initialized client's method invocation. 

193 

194 Satisfied by any identity/scope resolver the host binds; the 

195 server core consults it once per non-handshake request. 

196 """ 

197 

198 async def authorize( 

199 self, 

200 method: str, 

201 params: dict[str, Any], 

202 client_info: dict[str, Any], 

203 ) -> bool: 

204 """Return True to allow dispatch, False to reject with -32000.""" 

205 ... 

206 

207 

208__all__ = [ 

209 "MCPAuthorizerProtocol", 

210 "MCPPromptHandlerProtocol", 

211 "MCPPromptProviderProtocol", 

212 "MCPResourceHandlerProtocol", 

213 "MCPResourceProviderProtocol", 

214 "MCPServerProtocol", 

215 "MCPToolHandlerProtocol", 

216 "MCPToolProviderProtocol", 

217 "MCPTransportProtocol", 

218]