Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-mcp/src/lexigram/ai/mcp/types.py: 72%

1"""MCP type definitions — data structures for the MCP protocol. 

2 

3These dataclasses map directly to the MCP (Model Context Protocol) format. 

4They provide serialization to the JSON format expected by MCP clients. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from typing import Any 

11 

12 

13@dataclass 

14class MCPToolDefinition: 

15 """MCP tool definition sent to clients during tools/list. 

16 

17 Maps directly to the MCP tool schema format. 

18 """ 

19 

20 name: str 

21 """Unique tool identifier.""" 

22 

23 description: str = "" 

24 """Human-readable description shown to the AI client.""" 

25 

26 input_schema: dict[str, Any] = field(default_factory=dict) 

27 """JSON Schema describing the tool's parameters.""" 

28 

29 def to_dict(self) -> dict[str, Any]: 

30 """Serialize to MCP protocol format.""" 

31 return { 

32 "name": self.name, 

33 "description": self.description, 

34 "inputSchema": self.input_schema, 

35 } 

36 

37 

38@dataclass 

39class MCPToolResult: 

40 """Result of an MCP tool call returned to the client.""" 

41 

42 content: list[dict[str, Any]] = field(default_factory=list) 

43 """Content items (text, image, etc.).""" 

44 

45 is_error: bool = False 

46 """Whether the tool call resulted in an error.""" 

47 

48 @staticmethod 

49 def text(text: str, is_error: bool = False) -> MCPToolResult: 

50 """Create a text result.""" 

51 return MCPToolResult( 

52 content=[{"type": "text", "text": text}], 

53 is_error=is_error, 

54 ) 

55 

56 @staticmethod 

57 def error(message: str) -> MCPToolResult: 

58 """Create an error result.""" 

59 return MCPToolResult( 

60 content=[{"type": "text", "text": message}], 

61 is_error=True, 

62 ) 

63 

64 def to_dict(self) -> dict[str, Any]: 

65 """Serialize to MCP protocol format.""" 

66 return { 

67 "content": self.content, 

68 "isError": self.is_error, 

69 } 

70 

71 

72@dataclass 

73class MCPResource: 

74 """An MCP resource that clients can browse and read.""" 

75 

76 uri: str 

77 """Unique resource identifier (URI).""" 

78 

79 name: str 

80 """Human-readable resource name.""" 

81 

82 description: str = "" 

83 """Resource description.""" 

84 

85 mime_type: str = "text/plain" 

86 """MIME type of the resource content.""" 

87 

88 def to_dict(self) -> dict[str, Any]: 

89 """Serialize to MCP protocol format.""" 

90 return { 

91 "uri": self.uri, 

92 "name": self.name, 

93 "description": self.description, 

94 "mimeType": self.mime_type, 

95 } 

96 

97 

98@dataclass 

99class MCPResourceContent: 

100 """Content of an MCP resource returned to the client.""" 

101 

102 uri: str 

103 """Resource URI.""" 

104 

105 mime_type: str = "text/plain" 

106 """Content MIME type.""" 

107 

108 text: str | None = None 

109 """Text content (for text-based resources).""" 

110 

111 blob: str | None = None 

112 """Base64-encoded binary content.""" 

113 

114 def to_dict(self) -> dict[str, Any]: 

115 """Serialize to MCP protocol format.""" 

116 result: dict[str, Any] = { 

117 "uri": self.uri, 

118 "mimeType": self.mime_type, 

119 } 

120 if self.text is not None: 

121 result["text"] = self.text 

122 if self.blob is not None: 

123 result["blob"] = self.blob 

124 return result 

125 

126 

127@dataclass 

128class MCPPrompt: 

129 """An MCP prompt template that clients can use.""" 

130 

131 name: str 

132 """Unique prompt identifier.""" 

133 

134 description: str = "" 

135 """Prompt description.""" 

136 

137 arguments: list[dict[str, Any]] = field(default_factory=list) 

138 """Prompt arguments (name, description, required).""" 

139 

140 def to_dict(self) -> dict[str, Any]: 

141 """Serialize to MCP protocol format.""" 

142 return { 

143 "name": self.name, 

144 "description": self.description, 

145 "arguments": self.arguments, 

146 } 

147 

148 

149@dataclass 

150class MCPPromptMessage: 

151 """A message in an MCP prompt response.""" 

152 

153 role: str 

154 """Message role (user, assistant, system).""" 

155 

156 content: dict[str, Any] = field(default_factory=dict) 

157 """Message content (type + text/image/etc.).""" 

158 

159 def to_dict(self) -> dict[str, Any]: 

160 """Serialize to MCP protocol format.""" 

161 return { 

162 "role": self.role, 

163 "content": self.content, 

164 } 

165 

166 

167@dataclass 

168class MCPServerCapabilities: 

169 """Server capabilities sent during initialization.""" 

170 

171 tools: bool = True 

172 """Whether the server provides tools.""" 

173 

174 resources: bool = False 

175 """Whether the server provides resources.""" 

176 

177 prompts: bool = False 

178 """Whether the server provides prompts.""" 

179 

180 logging: bool = False 

181 """Whether the server supports the logging capability.""" 

182 

183 sampling: bool = False 

184 """Whether the server supports the sampling/createMessage capability.""" 

185 

186 def to_dict(self) -> dict[str, Any]: 

187 """Serialize to MCP protocol format.""" 

188 caps: dict[str, Any] = {} 

189 if self.tools: 

190 caps["tools"] = {} 

191 if self.resources: 

192 caps["resources"] = {} 

193 if self.prompts: 

194 caps["prompts"] = {} 

195 if self.logging: 

196 caps["logging"] = {} 

197 if self.sampling: 

198 caps["sampling"] = {} 

199 return caps 

200 

201 

202@dataclass 

203class MCPServerInfo: 

204 """Server metadata sent during initialization.""" 

205 

206 name: str = "lexigram-mcp" 

207 """Server name.""" 

208 

209 version: str = "1.0.0" 

210 """Server version.""" 

211 

212 def to_dict(self) -> dict[str, Any]: 

213 """Serialize to MCP protocol format.""" 

214 return { 

215 "name": self.name, 

216 "version": self.version, 

217 } 

218 

219 

220@dataclass 

221class MCPInitializeResult: 

222 """Result of the initialize method.""" 

223 

224 protocol_version: str = "2024-11-05" 

225 """MCP protocol version.""" 

226 

227 capabilities: MCPServerCapabilities = field( 

228 default_factory=MCPServerCapabilities, 

229 ) 

230 """Server capabilities.""" 

231 

232 server_info: MCPServerInfo = field(default_factory=MCPServerInfo) 

233 """Server information.""" 

234 

235 def to_dict(self) -> dict[str, Any]: 

236 """Serialize to MCP protocol format.""" 

237 return { 

238 "protocolVersion": self.protocol_version, 

239 "capabilities": self.capabilities.to_dict(), 

240 "serverInfo": self.server_info.to_dict(), 

241 } 

242 

243 

244@dataclass 

245class MCPJSONRPCRequest: 

246 """JSON-RPC request message.""" 

247 

248 jsonrpc: str = "2.0" 

249 """JSON-RPC version.""" 

250 

251 id: int | str | None = None 

252 """Request identifier.""" 

253 

254 method: str = "" 

255 """Method name to invoke.""" 

256 

257 params: dict[str, Any] | None = None 

258 """Method parameters.""" 

259 

260 def to_dict(self) -> dict[str, Any]: 

261 """Serialize to JSON-RPC format.""" 

262 result: dict[str, Any] = { 

263 "jsonrpc": self.jsonrpc, 

264 "method": self.method, 

265 } 

266 if self.id is not None: 

267 result["id"] = self.id 

268 if self.params is not None: 

269 result["params"] = self.params 

270 return result 

271 

272 

273@dataclass 

274class MCPJSONRPCResponse: 

275 """JSON-RPC response message.""" 

276 

277 jsonrpc: str = "2.0" 

278 """JSON-RPC version.""" 

279 

280 id: int | str | None = None 

281 """Request identifier this response is for.""" 

282 

283 result: dict[str, Any] | None = None 

284 """Success result (if not error).""" 

285 

286 error: dict[str, Any] | None = None 

287 """Error object (if error occurred).""" 

288 

289 def to_dict(self) -> dict[str, Any]: 

290 """Serialize to JSON-RPC format.""" 

291 result: dict[str, Any] = { 

292 "jsonrpc": self.jsonrpc, 

293 } 

294 if self.id is not None: 

295 result["id"] = self.id 

296 

297 # We must return either error or result 

298 if self.error is not None: 

299 result["error"] = self.error 

300 else: 

301 result["result"] = self.result if self.result is not None else {} 

302 

303 return result 

304 

305 @staticmethod 

306 def create_success( 

307 result: dict[str, Any], 

308 request_id: int | str | None = None, 

309 ) -> MCPJSONRPCResponse: 

310 """Create a success response.""" 

311 return MCPJSONRPCResponse( 

312 id=request_id, 

313 result=result, 

314 ) 

315 

316 # the method was formerly "success" but since that's not a field we can alias it 

317 success = create_success 

318 

319 @staticmethod 

320 def create_error( 

321 code: int, 

322 message: str, 

323 request_id: int | str | None = None, 

324 data: Any | None = None, 

325 ) -> MCPJSONRPCResponse: 

326 """Create an error response.""" 

327 error_obj: dict[str, Any] = { 

328 "code": code, 

329 "message": message, 

330 } 

331 if data is not None: 

332 error_obj["data"] = data 

333 return MCPJSONRPCResponse( 

334 id=request_id, 

335 error=error_obj, 

336 )