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

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

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

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

8from lexigram.logging import ( 

9 get_logger, 

10) 

11from lexigram.result import Err, Ok, Result 

12 

13logger = get_logger(__name__) 

14 

15 

16class PromptHandler: 

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

18 

19 Handles prompts/list and prompts/get methods by delegating to an 

20 MCPPromptProviderProtocol implementation. 

21 """ 

22 

23 def __init__( 

24 self, 

25 prompt_provider: Any | None = None, 

26 ) -> None: 

27 """Initialize the prompt handler. 

28 

29 Args: 

30 prompt_provider: Provider that handles prompt operations. 

31 """ 

32 self._provider = prompt_provider 

33 

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

35 """Handle prompts/list method. 

36 

37 Returns: 

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

39 """ 

40 if self._provider is None: 

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

42 

43 try: 

44 prompts = await self._provider.list_prompts() 

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

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

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

48 return Err( 

49 MCPPromptError( 

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

51 prompt_name="prompts/list", 

52 ) 

53 ) 

54 

55 async def get_prompt( 

56 self, 

57 name: str, 

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

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

60 """Handle prompts/get method. 

61 

62 Args: 

63 name: Name of the prompt to get. 

64 arguments: Arguments to fill in the prompt template. 

65 

66 Returns: 

67 ``Result[dict[str, Any], MCPError]`` with prompt payload. 

68 """ 

69 if self._provider is None: 

70 return Err( 

71 MCPPromptError( 

72 message="No prompt provider configured", 

73 prompt_name=name, 

74 ) 

75 ) 

76 

77 if arguments is None: 

78 arguments = {} 

79 

80 try: 

81 prompt = await self._provider.get_prompt(name, arguments) 

82 return Ok(prompt) 

83 except MCPPromptError as e: 

84 return Err(e) 

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

86 logger.error( 

87 "mcp_get_prompt_error", 

88 prompt=name, 

89 error=str(e), 

90 ) 

91 return Err( 

92 MCPPromptError( 

93 message=f"Failed to get prompt: {e!s}", 

94 prompt_name=name, 

95 ) 

96 ) 

97 

98 

99__all__ = ["PromptHandler"]