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 ....utilities.url_utils import normalize_url 

8from ..base import Exposure 

9from ..openai_base import OpenAICompatibleProvider 

10 

11 

12class CustomOpenAIEndpointProvider(OpenAICompatibleProvider): 

13 """Custom OpenAI-compatible endpoint provider. 

14 

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

16 by specifying a custom URL in the settings. 

17 """ 

18 

19 provider_name = "OpenAI-Compatible Endpoint" 

20 api_key_setting = "llm.openai_endpoint.api_key" 

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

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

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

24 api_key_optional = True 

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

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

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

28 

29 # Metadata for auto-discovery 

30 provider_key = "OPENAI_ENDPOINT" 

31 company_name = "OpenAI-Compatible" 

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

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

34 

35 @classmethod 

36 def requires_auth_for_models(cls): 

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

38 

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

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

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

42 """ 

43 return False 

44 

45 @classmethod 

46 def is_available(cls, settings_snapshot=None): 

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

48 

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

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

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

52 URL that differs from the default OpenAI endpoint. 

53 """ 

54 api_key = None 

55 try: 

56 api_key = get_setting_from_snapshot( 

57 cls.api_key_setting, 

58 default=None, 

59 settings_snapshot=settings_snapshot, 

60 ) 

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

62 return True 

63 except Exception as e: 

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

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

66 # a redacted exception message instead. 

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

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

69 

70 try: 

71 custom_url = get_setting_from_snapshot( 

72 cls.url_setting, 

73 default=None, 

74 settings_snapshot=settings_snapshot, 

75 ) 

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

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

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

79 return True 

80 except Exception: 

81 logger.debug( 

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

83 exc_info=True, 

84 ) 

85 

86 return False 

87 

88 @classmethod 

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

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

91 settings_snapshot = kwargs.get("settings_snapshot") 

92 

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

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

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

96 if cls.resolve_api_key(settings_snapshot) is None: 

97 logger.warning( 

98 "No API key configured for openai_endpoint provider; " 

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

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

101 ) 

102 

103 # Get custom endpoint URL from settings 

104 custom_url = get_setting_from_snapshot( 

105 "llm.openai_endpoint.url", 

106 default=cls.default_base_url, 

107 settings_snapshot=settings_snapshot, 

108 ) 

109 

110 # Normalize and pass the custom URL to parent implementation 

111 kwargs["base_url"] = ( 

112 normalize_url(custom_url) if custom_url else cls.default_base_url 

113 ) 

114 

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

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

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

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

119 stream_usage = get_setting_from_snapshot( 

120 "llm.openai_endpoint.stream_usage", 

121 default=False, 

122 settings_snapshot=settings_snapshot, 

123 ) 

124 if stream_usage: 

125 kwargs["stream_usage"] = True 

126 

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