Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/_experimental/mcp_server/elicitation_handler.py: 19%

1""" 

2MCP Elicitation Handler 

3Handles `elicitation/create` requests from upstream MCP servers by either: 

41. Relaying them to the connected downstream MCP client (if it supports elicitation) 

52. Returning a decline/error response (if no downstream client or unsupported) 

6Supports both Form mode (structured data collection) and URL mode (external URL 

7navigation for sensitive interactions like OAuth). 

8MCP Spec Reference: 

9 https://modelcontextprotocol.io/specification/2025-11-25/client/elicitation 

10""" 

11 

12from typing import TYPE_CHECKING, Final, Protocol, Union 

13 

14from litellm._logging import verbose_logger 

15 

16if TYPE_CHECKING: 16 ↛ 17line 16 didn't jump to line 17 because the condition on line 16 was never true

17 from mcp.types import ( 

18 ElicitRequestFormParams, 

19 ElicitRequestParams, 

20 ElicitRequestURLParams, 

21 ElicitResult, 

22 ErrorData, 

23 ) 

24 

25# Guard imports that require the mcp package 

26try: 

27 from mcp.types import ( 

28 ElicitRequestFormParams, 

29 ElicitRequestParams, 

30 ElicitRequestURLParams, 

31 ElicitResult, 

32 ErrorData, 

33 ) 

34 

35 MCP_ELICITATION_AVAILABLE = True 

36except ImportError: 

37 MCP_ELICITATION_AVAILABLE = False 

38 

39 

40class _DownstreamElicitSession(Protocol): 

41 """The downstream MCP client session methods this module relays elicitation requests through.""" 

42 

43 async def elicit_url(self, message: str, url: str, elicitation_id: str) -> "ElicitResult": ... 43 ↛ exitline 43 didn't return from function 'elicit_url' because

44 

45 async def elicit_form(self, message: str, requested_schema: dict[str, object]) -> "ElicitResult": ... 45 ↛ exitline 45 didn't return from function 'elicit_form' because

46 

47 async def elicit(self, message: str, requested_schema: dict[str, object]) -> "ElicitResult": ... 47 ↛ exitline 47 didn't return from function 'elicit' because

48 

49 

50async def handle_elicitation_request( 

51 context: object, 

52 params: "ElicitRequestParams", 

53 downstream_session: _DownstreamElicitSession | None = None, 

54 downstream_capabilities: object = None, 

55) -> Union["ElicitResult", "ErrorData"]: 

56 """ 

57 Handle an MCP elicitation/create request from an upstream MCP server. 

58 In Gateway mode (Mode A), we relay the elicitation request to the 

59 connected downstream client if they declared elicitation capabilities. 

60 In Tool Bridge mode (Mode B), there's no persistent downstream MCP 

61 client, so we return a decline response. 

62 Args: 

63 context: MCP RequestContext from the upstream server connection. 

64 params: The ElicitRequestParams (either form or URL mode). 

65 downstream_session: The ServerSession to the downstream client, 

66 if available (for relaying). 

67 downstream_capabilities: The downstream client's declared 

68 capabilities, used to check elicitation support. 

69 Returns: 

70 ElicitResult with the user's response, or ErrorData on failure. 

71 """ 

72 if not MCP_ELICITATION_AVAILABLE: 

73 return ErrorData( 

74 code=-1, 

75 message="MCP elicitation is not available (mcp package not installed)", 

76 ) 

77 try: 

78 mode: Final = getattr(params, "mode", "form") 

79 verbose_logger.info( 

80 "MCP elicitation: received request mode=%s, message=%s", 

81 mode, 

82 getattr(params, "message", ""), 

83 ) 

84 # Check if we have a downstream session to relay to 

85 if downstream_session is not None: 

86 return await _relay_elicitation_to_downstream( 

87 params=params, 

88 downstream_session=downstream_session, 

89 downstream_capabilities=downstream_capabilities, 

90 ) 

91 # No downstream session — we're in Tool Bridge mode 

92 # or the client doesn't support elicitation 

93 verbose_logger.info("MCP elicitation: no downstream session available, declining") 

94 return ElicitResult( 

95 action="decline", 

96 ) 

97 except Exception as e: 

98 verbose_logger.exception("MCP elicitation handler failed: %s", e) 

99 return ErrorData( 

100 code=-1, 

101 message=f"Elicitation failed: {e}", 

102 ) 

103 

104 

105async def _relay_elicitation_to_downstream( 

106 params: "ElicitRequestParams", 

107 downstream_session: _DownstreamElicitSession, 

108 downstream_capabilities: object = None, 

109) -> Union["ElicitResult", "ErrorData"]: 

110 """ 

111 Relay an elicitation request to the downstream MCP client. 

112 Uses the ServerSession's elicit_form() or elicit_url() methods to 

113 send the elicitation request back to the connected client. 

114 Args: 

115 params: The elicitation request parameters. 

116 downstream_session: The ServerSession connected to the downstream client. 

117 downstream_capabilities: Client capabilities to check support. 

118 Returns: 

119 ElicitResult from the downstream client. 

120 """ 

121 mode: Final = getattr(params, "mode", "form") 

122 # Check if the downstream client supports the requested mode 

123 if downstream_capabilities is not None: 

124 elicit_caps: Final[object] = getattr(downstream_capabilities, "elicitation", None) 

125 if elicit_caps is None: 

126 verbose_logger.info("MCP elicitation: downstream client does not support elicitation") 

127 return ElicitResult(action="decline") 

128 if mode == "url": 

129 url_cap: Final[object] = getattr(elicit_caps, "url", None) 

130 if url_cap is None: 

131 verbose_logger.info("MCP elicitation: downstream client does not support URL mode") 

132 return ElicitResult(action="decline") 

133 if mode == "form": 

134 form_cap: Final[object] = getattr(elicit_caps, "form", None) 

135 if form_cap is None: 

136 verbose_logger.info("MCP elicitation: downstream client does not support form mode") 

137 return ElicitResult(action="decline") 

138 try: 

139 if mode == "url" and isinstance(params, ElicitRequestURLParams): 

140 # URL mode: relay URL to client for external navigation 

141 verbose_logger.info( 

142 "MCP elicitation: relaying URL mode to downstream, url=%s", 

143 getattr(params, "url", ""), 

144 ) 

145 result = await downstream_session.elicit_url( 

146 message=params.message, 

147 url=params.url, 

148 elicitation_id=params.elicitation_id, 

149 ) 

150 elif isinstance(params, ElicitRequestFormParams): 

151 # Form mode: relay structured form to client 

152 verbose_logger.info("MCP elicitation: relaying form mode to downstream") 

153 result = await downstream_session.elicit_form( 

154 message=params.message, 

155 requested_schema=params.requested_schema, 

156 ) 

157 else: 

158 # Fallback for generic ElicitRequestParams — pass an empty schema 

159 # since elicit() requires requested_schema as a positional arg. 

160 verbose_logger.info("MCP elicitation: relaying generic elicitation to downstream") 

161 result = await downstream_session.elicit( 

162 message=getattr(params, "message", ""), 

163 requested_schema=getattr(params, "requested_schema", {}), # mutable-ok: elicitation default schema 

164 ) 

165 verbose_logger.info( 

166 "MCP elicitation: downstream responded with action=%s", 

167 getattr(result, "action", "unknown"), 

168 ) 

169 return result 

170 except Exception as e: 

171 verbose_logger.warning("MCP elicitation: failed to relay to downstream: %s", e) 

172 # If relay fails, decline gracefully 

173 return ElicitResult(action="decline")