Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/vertex_ai_endpoints/langfuse_endpoints.py: 25%

1""" 

2What is this? 

3 

4Logging Pass-Through Endpoints 

5""" 

6 

7""" 

81. Create pass-through endpoints for any LITELLM_BASE_URL/langfuse/<endpoint> map to LANGFUSE_BASE_URL/<endpoint> 

9""" 

10 

11import base64 

12import os 

13from base64 import b64encode 

14from typing import Final 

15from urllib.parse import unquote 

16 

17import httpx 

18from fastapi import APIRouter, HTTPException, Request, Response, status 

19 

20import litellm 

21from litellm.litellm_core_utils.url_utils import SSRFError, validate_url 

22from litellm.proxy._types import * 

23from litellm.proxy.auth.user_api_key_auth import user_api_key_auth 

24from litellm.proxy.common_utils.http_parsing_utils import _safe_get_request_headers 

25from litellm.proxy.litellm_pre_call_utils import _get_dynamic_logging_metadata 

26from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( 

27 create_pass_through_route, 

28) 

29 

30router: Final = APIRouter() 

31default_vertex_config: Final = None 

32_DEFAULT_LANGFUSE_HOST: Final = "https://cloud.langfuse.com" 

33 

34 

35def create_request_copy(request: Request): 

36 return { 

37 "method": request.method, 

38 "url": str(request.url), 

39 "headers": _safe_get_request_headers(request).copy(), 

40 "cookies": request.cookies, 

41 "query_params": dict(request.query_params), 

42 } 

43 

44 

45def _decode_to_convergence(value: str) -> str: 

46 previous = value 

47 while True: 

48 decoded = unquote(previous) 

49 if decoded == previous: 

50 return decoded 

51 previous = decoded 

52 

53 

54def _normalize_langfuse_base_url(base_target_url: str) -> str: 

55 if not (base_target_url.startswith("http://") or base_target_url.startswith("https://")): 

56 # Existing behavior allows host-only Langfuse settings. 

57 base_target_url = "http://" + base_target_url 

58 

59 try: 

60 base_url: Final = httpx.URL(base_target_url) 

61 except Exception as e: 

62 raise HTTPException( 

63 status_code=status.HTTP_400_BAD_REQUEST, 

64 detail={"error": f"Invalid Langfuse host: {e}"}, 

65 ) 

66 

67 if base_url.scheme not in ("http", "https") or not base_url.host: 

68 raise HTTPException( 

69 status_code=status.HTTP_400_BAD_REQUEST, 

70 detail={"error": "Invalid Langfuse host"}, 

71 ) 

72 

73 if base_url.userinfo: 

74 raise HTTPException( 

75 status_code=status.HTTP_400_BAD_REQUEST, 

76 detail={"error": "Langfuse host must not include credentials"}, 

77 ) 

78 

79 return str(base_url) 

80 

81 

82def _validate_langfuse_proxy_path(endpoint: str) -> str: 

83 decoded_endpoint: Final = _decode_to_convergence(endpoint) 

84 if any(ord(char) < 32 for char in decoded_endpoint): 

85 raise HTTPException( 

86 status_code=status.HTTP_400_BAD_REQUEST, 

87 detail={"error": "Invalid Langfuse endpoint path"}, 

88 ) 

89 if "\\" in decoded_endpoint or decoded_endpoint.startswith("//"): 

90 raise HTTPException( 

91 status_code=status.HTTP_400_BAD_REQUEST, 

92 detail={"error": "Invalid Langfuse endpoint path"}, 

93 ) 

94 

95 endpoint_path: Final = "/" + decoded_endpoint.lstrip("/") 

96 if any(segment in (".", "..") for segment in endpoint_path.split("/")): 

97 raise HTTPException( 

98 status_code=status.HTTP_400_BAD_REQUEST, 

99 detail={"error": "Invalid Langfuse endpoint path"}, 

100 ) 

101 return endpoint_path 

102 

103 

104def _get_langfuse_proxy_credentials( 

105 *, 

106 dynamic_host_supplied: bool, 

107 dynamic_langfuse_public_key: str | None, 

108 dynamic_langfuse_secret_key: str | None, 

109): 

110 if dynamic_host_supplied: 

111 if not dynamic_langfuse_public_key or not dynamic_langfuse_secret_key: 

112 raise HTTPException( 

113 status_code=status.HTTP_400_BAD_REQUEST, 

114 detail={"error": "Dynamic Langfuse hosts must include dynamic Langfuse credentials"}, 

115 ) 

116 return dynamic_langfuse_public_key, dynamic_langfuse_secret_key 

117 

118 return ( 

119 dynamic_langfuse_public_key or litellm.utils.get_secret(secret_name="LANGFUSE_PUBLIC_KEY"), 

120 dynamic_langfuse_secret_key or litellm.utils.get_secret(secret_name="LANGFUSE_SECRET_KEY"), 

121 ) 

122 

123 

124def _build_langfuse_proxy_target( 

125 *, 

126 endpoint: str, 

127 base_target_url: str, 

128 dynamic_host_supplied: bool, 

129): 

130 endpoint_path: Final = _validate_langfuse_proxy_path(endpoint) 

131 base_url: Final = httpx.URL(_normalize_langfuse_base_url(base_target_url)) 

132 updated_url: Final = base_url.copy_with(path=endpoint_path) 

133 custom_headers: Final = {} 

134 

135 if dynamic_host_supplied and getattr(litellm, "user_url_validation", True): 

136 try: 

137 target_url, host_header = validate_url(str(updated_url)) 

138 except SSRFError as e: 

139 raise HTTPException( 

140 status_code=status.HTTP_400_BAD_REQUEST, 

141 detail={"error": f"Invalid Langfuse host: {e}"}, 

142 ) 

143 custom_headers["Host"] = host_header 

144 return target_url, custom_headers 

145 

146 return str(updated_url), custom_headers 

147 

148 

149@router.api_route( 

150 "/langfuse/{endpoint:path}", 

151 methods=["GET", "POST", "PUT", "DELETE", "PATCH"], 

152 tags=["Langfuse Pass-through", "pass-through"], 

153) 

154async def langfuse_proxy_route( 

155 endpoint: str, 

156 request: Request, 

157 fastapi_response: Response, 

158): 

159 """ 

160 Call Langfuse via LiteLLM proxy. Works with Langfuse SDK. 

161 

162 [Docs](https://docs.litellm.ai/docs/pass_through/langfuse) 

163 """ 

164 from litellm.proxy.proxy_server import proxy_config 

165 

166 ## CHECK FOR LITELLM API KEY IN THE QUERY PARAMS - ?..key=LITELLM_API_KEY 

167 api_key = request.headers.get("Authorization") or "" 

168 

169 ## decrypt base64 hash 

170 api_key = api_key.replace("Basic ", "") 

171 

172 decoded_bytes: Final = base64.b64decode(api_key) 

173 decoded_str: Final = decoded_bytes.decode("utf-8") 

174 api_key = decoded_str.split(":")[1] # assume api key is passed in as secret key 

175 

176 user_api_key_dict: Final = await user_api_key_auth(request=request, api_key=f"Bearer {api_key}") 

177 

178 callback_settings_obj: Final[TeamCallbackMetadata | None] = _get_dynamic_logging_metadata( 

179 user_api_key_dict=user_api_key_dict, proxy_config=proxy_config 

180 ) 

181 

182 dynamic_langfuse_public_key: str | None = None 

183 dynamic_langfuse_secret_key: str | None = None 

184 dynamic_langfuse_host: str | None = None 

185 if callback_settings_obj is not None and callback_settings_obj.callback_vars is not None: 

186 for k, v in callback_settings_obj.callback_vars.items(): 

187 if k == "langfuse_public_key": 

188 dynamic_langfuse_public_key = v 

189 elif k == "langfuse_secret_key": 

190 dynamic_langfuse_secret_key = v 

191 elif k == "langfuse_host": 

192 dynamic_langfuse_host = v 

193 

194 dynamic_host_supplied: Final = dynamic_langfuse_host is not None 

195 base_target_url: Final[str] = ( 

196 dynamic_langfuse_host or os.getenv("LANGFUSE_HOST", _DEFAULT_LANGFUSE_HOST) or _DEFAULT_LANGFUSE_HOST 

197 ) 

198 langfuse_public_key, langfuse_secret_key = _get_langfuse_proxy_credentials( 

199 dynamic_host_supplied=dynamic_host_supplied, 

200 dynamic_langfuse_public_key=dynamic_langfuse_public_key, 

201 dynamic_langfuse_secret_key=dynamic_langfuse_secret_key, 

202 ) 

203 target_url, target_headers = _build_langfuse_proxy_target( 

204 endpoint=endpoint, 

205 base_target_url=base_target_url, 

206 dynamic_host_supplied=dynamic_host_supplied, 

207 ) 

208 

209 langfuse_combined_key: Final = "Basic " + b64encode(f"{langfuse_public_key}:{langfuse_secret_key}".encode()).decode( 

210 "ascii" 

211 ) 

212 target_headers["Authorization"] = langfuse_combined_key 

213 

214 ## CREATE PASS-THROUGH 

215 endpoint_func: Final = create_pass_through_route( 

216 endpoint=endpoint, 

217 target=target_url, 

218 custom_headers=target_headers, 

219 query_params=dict(request.query_params), 

220 ) # dynamically construct pass-through endpoint based on incoming path 

221 received_value: Final = await endpoint_func( 

222 request, 

223 fastapi_response, 

224 user_api_key_dict, 

225 ) 

226 

227 return received_value