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

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

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.ai.mcp.types import MCPToolResult 

8from lexigram.contracts.mcp.exceptions import MCPError, MCPToolCallError 

9from lexigram.logging import ( 

10 get_logger, 

11) 

12from lexigram.result import Err, Ok, Result 

13 

14logger = get_logger(__name__) 

15 

16 

17class ToolHandler: 

18 """Handler for MCP tool-related methods. 

19 

20 Handles tools/list and tools/call methods by delegating to a 

21 MCPToolProviderProtocol implementation. 

22 

23 All handler methods return ``Result`` so callers can handle expected 

24 failures without raising exceptions. 

25 """ 

26 

27 def __init__( 

28 self, 

29 tool_provider: Any | None = None, 

30 ) -> None: 

31 """Initialize the tool handler. 

32 

33 Args: 

34 tool_provider: Provider that handles tool listing and execution. 

35 """ 

36 self._provider = tool_provider 

37 

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

39 """Handle tools/list method. 

40 

41 Returns: 

42 ``Result`` containing tools list in MCP format. 

43 """ 

44 if self._provider is None: 

45 return Ok({"tools": []}) 

46 

47 try: 

48 tools = await self._provider.list_tools() 

49 return Ok({"tools": tools}) 

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

51 logger.error("mcp_list_tools_error", error=str(e)) 

52 return Err( 

53 MCPToolCallError( 

54 message=f"Failed to list tools: {e!s}", 

55 tool_name="tools/list", 

56 ) 

57 ) 

58 

59 async def call_tool( 

60 self, 

61 name: str, 

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

63 ) -> Result[dict[str, Any], MCPError]: 

64 """Handle tools/call method. 

65 

66 Returns a ``Result`` — ``Ok`` wraps the serialized MCP tool result dict; 

67 ``Err`` wraps an :class:`~lexigram.contracts.mcp.exceptions.MCPToolCallError` 

68 describing what went wrong. 

69 

70 Args: 

71 name: Name of the tool to call. 

72 arguments: Arguments to pass to the tool. 

73 

74 Returns: 

75 ``Result[dict[str, Any], MCPError]`` — never raises expected errors. 

76 """ 

77 if self._provider is None: 

78 return Err( 

79 MCPToolCallError( 

80 message="No tool provider configured", 

81 tool_name=name, 

82 ) 

83 ) 

84 

85 if arguments is None: 

86 arguments = {} 

87 

88 try: 

89 result = await self._provider.call_tool(name, arguments) 

90 except MCPToolCallError as e: 

91 logger.error("mcp_tool_call_error", tool=name, error=str(e)) 

92 return Err(e) 

93 except (ValueError, TypeError, KeyError) as e: 

94 logger.error("mcp_tool_call_invalid_args", tool=name, error=str(e)) 

95 return Err( 

96 MCPToolCallError( 

97 message=f"Invalid arguments for tool '{name}': {e}", 

98 tool_name=name, 

99 ) 

100 ) 

101 except (AttributeError, LookupError, OSError, RuntimeError) as e: 

102 logger.error("mcp_tool_call_error", tool=name, error=str(e)) 

103 return Err( 

104 MCPToolCallError( 

105 message=f"Tool call failed: {e!s}", 

106 tool_name=name, 

107 ) 

108 ) 

109 

110 # Normalise result to MCP wire format 

111 if isinstance(result, MCPToolResult): 

112 return Ok(result.to_dict()) 

113 return Ok(MCPToolResult.text(str(result)).to_dict()) 

114 

115 

116__all__ = ["ToolHandler"]