Coverage for src/local_deep_research/llm/providers/implementations/custom_openai_endpoint.py: 100%

1"""Custom OpenAI-compatible endpoint provider for Local Deep Research.""" 

2 

3from ....security.secure_logging import logger 

4 

5from ....config.thread_settings import get_setting_from_snapshot 

6from ....security.log_sanitizer import redact_secrets 

7from ....settings.exceptions import SettingsValueError 

8from ....utilities.url_utils import normalize_url 

9from ..base import Exposure 

10from ..openai_base import OpenAICompatibleProvider 

11 

12 

13class CustomOpenAIEndpointProvider(OpenAICompatibleProvider): 

14 """Custom OpenAI-compatible endpoint provider. 

15 

16 This provider allows users to connect to any OpenAI-compatible API endpoint 

17 by specifying a custom URL in the settings. 

18 """ 

19 

20 provider_name = "OpenAI-Compatible Endpoint" 

21 api_key_setting = "llm.openai_endpoint.api_key" 

22 # Many OpenAI-compatible servers (vLLM, local LLMs, etc.) don't require 

23 # auth. Optional flag lets the resolver fall back to a placeholder when 

24 # no key is configured, instead of raising at LLM construction time. 

25 api_key_optional = True 

26 url_setting = "llm.openai_endpoint.url" # type: ignore[assignment] # Settings key for URL 

27 default_base_url = "https://api.openai.com/v1" 

28 default_model = "" # User must explicitly pick a model — no silent fallback 

29 

30 # Metadata for auto-discovery 

31 provider_key = "OPENAI_ENDPOINT" 

32 company_name = "OpenAI-Compatible" 

33 is_cloud = None # Unknown — could be local or cloud 

34 egress_exposure = Exposure.EXPOSING # URL-configurable; fail closed to exposing until classifier refines by URL 

35 

36 @classmethod 

37 def requires_auth_for_models(cls): 

38 """Custom endpoints may or may not require authentication for listing models. 

39 

40 Many OpenAI-compatible servers (vLLM, local LLMs, etc.) don't require 

41 authentication. Return False to allow model listing without an API key. 

42 If the endpoint requires auth, the OpenAI client will raise an error. 

43 """ 

44 return False 

45 

46 @classmethod 

47 def is_available(cls, settings_snapshot=None): 

48 """Custom endpoints are available with either an API key or a custom URL. 

49 

50 Unlike cloud-only providers, custom endpoints support keyless local 

51 servers (vLLM, text-generation-webui, etc.). The provider is 

52 considered configured when the user has set either an API key or a 

53 URL that differs from the default OpenAI endpoint. 

54 """ 

55 api_key = None 

56 try: 

57 api_key = get_setting_from_snapshot( 

58 cls.api_key_setting, 

59 default=None, 

60 settings_snapshot=settings_snapshot, 

61 ) 

62 if api_key and str(api_key).strip(): 

63 return True 

64 except SettingsValueError: 

65 raise 

66 except Exception as e: 

67 # Drop exc_info — the cause chain may embed the api_key value 

68 # if a settings-layer error message surfaces it. Interpolate 

69 # a redacted exception message instead. 

70 safe_msg = redact_secrets(str(e), api_key) 

71 logger.debug(f"Error checking provider availability: {safe_msg}") 

72 

73 try: 

74 custom_url = get_setting_from_snapshot( 

75 cls.url_setting, 

76 default=None, 

77 settings_snapshot=settings_snapshot, 

78 ) 

79 if custom_url and str(custom_url).strip(): 

80 normalized = normalize_url(str(custom_url).strip()) 

81 if normalized.rstrip("/") != cls.default_base_url.rstrip("/"): 

82 return True 

83 except SettingsValueError: 

84 raise 

85 except Exception: 

86 logger.debug( 

87 f"Error reading URL setting '{cls.url_setting}'", 

88 exc_info=True, 

89 ) 

90 

91 return False 

92 

93 @classmethod 

94 def create_llm(cls, model_name=None, temperature=0.7, **kwargs): 

95 """Override to get URL from settings.""" 

96 settings_snapshot = kwargs.get("settings_snapshot") 

97 

98 # Keyless construction is supported (vLLM, text-generation-webui), 

99 # but a key-requiring endpoint (e.g. OpenRouter) fails with an 

100 # opaque upstream 401 — warn so the misconfiguration is traceable. 

101 if cls.resolve_api_key(settings_snapshot) is None: 

102 logger.warning( 

103 "No API key configured for openai_endpoint provider; " 

104 "proceeding with a placeholder. If your endpoint requires " 

105 "an API key, set llm.openai_endpoint.api_key in settings." 

106 ) 

107 

108 # Get custom endpoint URL from settings 

109 custom_url = get_setting_from_snapshot( 

110 "llm.openai_endpoint.url", 

111 default=cls.default_base_url, 

112 settings_snapshot=settings_snapshot, 

113 ) 

114 

115 # Normalize and pass the custom URL to parent implementation 

116 kwargs["base_url"] = ( 

117 normalize_url(custom_url) if custom_url else cls.default_base_url 

118 ) 

119 

120 # Opt-in token usage on streamed responses. Off by default because 

121 # some OpenAI-compatible gateways reject stream_options with a 400 

122 # (e.g. xAI, Databricks AI Gateway, Azure "on your data"), which 

123 # would break the call entirely — worse than missing token counts. 

124 stream_usage = get_setting_from_snapshot( 

125 "llm.openai_endpoint.stream_usage", 

126 default=False, 

127 settings_snapshot=settings_snapshot, 

128 ) 

129 if stream_usage: 

130 kwargs["stream_usage"] = True 

131 

132 return super().create_llm(model_name, temperature, **kwargs)