Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_endpoints/mcp_connector_import.py: 79%

1""" 

2Convert Anthropic MCP connector definitions into LiteLLM MCP server create requests. 

3 

4Two interchange shapes are accepted: 

5- the ``mcpServers`` mapping used by Claude Desktop / Claude Code config files 

6- the ``mcp_servers`` array used by the Anthropic Messages API MCP connector 

7""" 

8 

9import re 

10from collections.abc import Mapping 

11from dataclasses import dataclass 

12from typing import Final 

13 

14from pydantic import AliasChoices, BaseModel, ConfigDict, Field, ValidationError 

15 

16from litellm.proxy._types import MCPApprovalStatus, NewMCPServerRequest 

17from litellm.types.mcp import MCPAuth, MCPAuthType, MCPCredentials, MCPTransport 

18 

19 

20class MCPConnectorEntry(BaseModel): 

21 model_config = ConfigDict(populate_by_name=True) 

22 

23 name: str | None = None 

24 type: str | None = None 

25 url: str | None = None 

26 authorization_token: str | None = Field( 

27 default=None, validation_alias=AliasChoices("authorization_token", "authorizationToken") 

28 ) 

29 headers: Mapping[str, str] | None = None 

30 command: str | None = None 

31 args: tuple[str, ...] = Field(default_factory=tuple) 

32 env: Mapping[str, str] = Field(default_factory=dict) 

33 description: str | None = None 

34 

35 

36class MCPConnectorImportRequest(BaseModel): 

37 model_config = ConfigDict(populate_by_name=True) 

38 

39 mcp_servers: Mapping[str, MCPConnectorEntry] | tuple[MCPConnectorEntry, ...] = Field( 

40 validation_alias=AliasChoices("mcp_servers", "mcpServers") 

41 ) 

42 

43 

44@dataclass(frozen=True, slots=True) 

45class ConvertedConnector: 

46 name: str 

47 request: NewMCPServerRequest 

48 

49 

50@dataclass(frozen=True, slots=True) 

51class ConnectorConversionError: 

52 name: str 

53 error: str 

54 

55 

56class MCPConnectorImportResult(BaseModel): 

57 name: str 

58 server_id: str 

59 alias: str 

60 

61 

62class MCPConnectorImportSkipped(BaseModel): 

63 name: str 

64 reason: str 

65 

66 

67class MCPConnectorImportFailure(BaseModel): 

68 name: str 

69 error: str 

70 

71 

72class MCPConnectorImportResponse(BaseModel): 

73 imported: tuple[MCPConnectorImportResult, ...] 

74 skipped: tuple[MCPConnectorImportSkipped, ...] 

75 errors: tuple[MCPConnectorImportFailure, ...] 

76 

77 

78_INVALID_SERVER_NAME_CHARS: Final = re.compile(r"[^A-Za-z0-9_]") 

79 

80 

81def sanitize_connector_name(name: str) -> str: 

82 sanitized: Final = re.sub(r"_+", "_", _INVALID_SERVER_NAME_CHARS.sub("_", name.strip())).strip("_") 

83 return sanitized 

84 

85 

86_SSE_TYPES: Final = frozenset({"sse"}) 

87_URL_TYPES: Final = frozenset({"url", "http", "streamable_http", "streamable-http", "sse", ""}) 

88 

89 

90def _convert_entry(name: str, entry: MCPConnectorEntry) -> ConvertedConnector | ConnectorConversionError: 

91 sanitized_name: Final = sanitize_connector_name(name) 

92 if not sanitized_name: 

93 return ConnectorConversionError(name=name, error="Connector name is empty after sanitization.") 

94 

95 if entry.url and entry.command: 

96 return ConnectorConversionError(name=name, error="Connector cannot have both a url and a command.") 

97 

98 if entry.command: 

99 try: 

100 stdio_request: Final = NewMCPServerRequest( 

101 server_name=sanitized_name, 

102 alias=sanitized_name, 

103 description=entry.description, 

104 approval_status=MCPApprovalStatus.active, 

105 transport=MCPTransport.stdio, 

106 command=entry.command, 

107 args=list(entry.args), 

108 env=dict(entry.env), 

109 ) 

110 except ValidationError as e: 

111 return ConnectorConversionError(name=name, error=_first_validation_message(e)) 

112 return ConvertedConnector(name=name, request=stdio_request) 

113 

114 if not entry.url: 

115 return ConnectorConversionError(name=name, error="Connector must have either a url or a command.") 

116 

117 entry_type: Final = (entry.type or "").lower() 

118 if entry_type not in _URL_TYPES: 118 ↛ 121line 118 didn't jump to line 121 because the condition on line 118 was always true

119 return ConnectorConversionError(name=name, error=f"Unsupported connector type '{entry.type}'.") 

120 

121 transport: Final = MCPTransport.sse if entry_type in _SSE_TYPES else MCPTransport.http 

122 auth: Final = _remote_auth(entry) 

123 try: 

124 remote_request: Final = NewMCPServerRequest( 

125 server_name=sanitized_name, 

126 alias=sanitized_name, 

127 description=entry.description, 

128 approval_status=MCPApprovalStatus.active, 

129 transport=transport, 

130 url=entry.url, 

131 auth_type=auth.auth_type, 

132 credentials=auth.credentials, 

133 static_headers=auth.static_headers, 

134 ) 

135 except ValidationError as e: 

136 return ConnectorConversionError(name=name, error=_first_validation_message(e)) 

137 return ConvertedConnector(name=name, request=remote_request) 

138 

139 

140@dataclass(frozen=True, slots=True) 

141class _RemoteAuth: 

142 auth_type: MCPAuthType 

143 credentials: MCPCredentials | None 

144 static_headers: dict[str, str] | None 

145 

146 

147_AUTHORIZATION_HEADER: Final = "authorization" 

148_BEARER_PREFIX: Final = "bearer " 

149 

150 

151def _remote_auth(entry: MCPConnectorEntry) -> _RemoteAuth: 

152 headers: Final[Mapping[str, str]] = entry.headers or {} 

153 header_value: Final = next((value for key, value in headers.items() if key.lower() == _AUTHORIZATION_HEADER), None) 

154 remaining: Final = {key: value for key, value in headers.items() if key.lower() != _AUTHORIZATION_HEADER} or None 

155 if entry.authorization_token: 

156 return _RemoteAuth(MCPAuth.bearer_token, {"auth_value": entry.authorization_token}, remaining) 

157 if not header_value: 

158 return _RemoteAuth(MCPAuth.none, None, remaining) 

159 if header_value.lower().startswith(_BEARER_PREFIX): 

160 return _RemoteAuth(MCPAuth.bearer_token, {"auth_value": header_value[len(_BEARER_PREFIX) :]}, remaining) 

161 return _RemoteAuth(MCPAuth.authorization, {"auth_value": header_value}, remaining) 

162 

163 

164def _first_validation_message(error: ValidationError) -> str: 

165 messages: Final = tuple(str(detail.get("msg", "")) for detail in error.errors()) 

166 return messages[0] if messages else str(error) 

167 

168 

169def convert_connector_entries( 

170 payload: MCPConnectorImportRequest, 

171) -> tuple[ConvertedConnector | ConnectorConversionError, ...]: 

172 servers: Final = payload.mcp_servers 

173 if isinstance(servers, Mapping): 

174 return tuple(_convert_entry(name, entry) for name, entry in servers.items()) 

175 return tuple( 

176 _convert_entry(entry.name or "", entry) if entry.name else _named_entry_error(index, entry) 

177 for index, entry in enumerate(servers) 

178 ) 

179 

180 

181def _named_entry_error(index: int, entry: MCPConnectorEntry) -> ConnectorConversionError: 

182 return ConnectorConversionError( 

183 name=entry.url or f"entry {index}", 

184 error="Connector entries in list form must have a name.", 

185 )