Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-mcp/src/lexigram/ai/mcp/server/handlers/resources.py: 29%

1"""MCP resource handler for the MCP server.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.contracts.mcp.exceptions import MCPError, MCPResourceError 

8from lexigram.logging import ( 

9 get_logger, 

10) 

11from lexigram.result import Err, Ok, Result 

12 

13logger = get_logger(__name__) 

14 

15 

16class ResourceHandler: 

17 """Handler for MCP resource-related methods. 

18 

19 Handles resources/list, resources/read, and resources/templates/list 

20 methods by delegating to an MCPResourceProviderProtocol implementation. 

21 """ 

22 

23 def __init__( 

24 self, 

25 resource_provider: Any | None = None, 

26 ) -> None: 

27 """Initialize the resource handler. 

28 

29 Args: 

30 resource_provider: Provider that handles resource operations. 

31 """ 

32 self._provider = resource_provider 

33 

34 async def list_resources(self) -> Result[dict[str, Any], MCPError]: 

35 """Handle resources/list method. 

36 

37 Returns: 

38 ``Result`` containing resources list in MCP format. 

39 """ 

40 if self._provider is None: 

41 return Ok({"resources": []}) 

42 

43 try: 

44 resources = await self._provider.list_resources() 

45 return Ok({"resources": resources}) 

46 except (RuntimeError, TypeError, AttributeError, LookupError, OSError) as e: 

47 logger.error("mcp_list_resources_error", error=str(e)) 

48 return Err( 

49 MCPResourceError( 

50 message=f"Failed to list resources: {e!s}", 

51 uri="resources/list", 

52 ) 

53 ) 

54 

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

56 """Handle resources/read method. 

57 

58 Args: 

59 uri: URI of the resource to read. 

60 

61 Returns: 

62 ``Result[dict[str, Any], MCPError]`` with MCP-formatted contents. 

63 """ 

64 if self._provider is None: 

65 return Err( 

66 MCPResourceError( 

67 message="No resource provider configured", 

68 uri=uri, 

69 ) 

70 ) 

71 

72 try: 

73 content = await self._provider.read_resource(uri) 

74 return Ok({"contents": [content]}) 

75 except MCPResourceError as e: 

76 return Err(e) 

77 except (RuntimeError, TypeError, AttributeError, LookupError, OSError) as e: 

78 logger.error( 

79 "mcp_read_resource_error", 

80 uri=uri, 

81 error=str(e), 

82 ) 

83 return Err( 

84 MCPResourceError( 

85 message=f"Failed to read resource: {e!s}", 

86 uri=uri, 

87 ) 

88 ) 

89 

90 async def list_templates(self) -> Result[dict[str, Any], MCPError]: 

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

92 

93 Returns: 

94 ``Result`` containing URI templates in MCP format. 

95 """ 

96 if self._provider is None: 

97 return Ok({"resourceTemplates": []}) 

98 

99 try: 

100 # Try to get templates if the provider supports it 

101 if hasattr(self._provider, "list_templates"): 

102 templates = await self._provider.list_templates() 

103 return Ok({"resourceTemplates": templates}) 

104 return Ok({"resourceTemplates": []}) 

105 except (RuntimeError, TypeError, AttributeError, LookupError, OSError) as e: 

106 logger.error("mcp_list_templates_error", error=str(e)) 

107 return Err( 

108 MCPResourceError( 

109 message=f"Failed to list resource templates: {e!s}", 

110 uri="resources/templates/list", 

111 ) 

112 ) 

113 

114 

115__all__ = ["ResourceHandler"]