"""Common OCI Generative AI chat-completion SDK boundary. This module is intentionally small: callers provide a validated model route and a JSON schema, and this boundary performs one OCI GenAI chat call. It does not know about MCP, Streamlit, business payloads, or bearer tokens. """ from __future__ import annotations from collections.abc import Mapping, Sequence from dataclasses import dataclass from functools import lru_cache import os from pathlib import Path import re from typing import Dict, Optional, Protocol ROOT = Path(__file__).resolve().parents[1] DOTENV_PATH = ROOT / ".env" ALLOWED_OCI_SETTINGS = frozenset( { "OCI_AUTH_TYPE", "OCI_CONFIG_FILE", "OCI_GENAI_COMPARTMENT_ID", "OCI_PROFILE", } ) _COMPARTMENT_ID = re.compile(r"^ocid1\.compartment\.[A-Za-z0-9._-]+$") class CompletionClient(Protocol): """Minimal completion client contract shared by app layers.""" def complete( self, system_prompt: str, user_prompt: str, response_schema: Mapping[str, object], max_tokens: int, temperature: Optional[float], ) -> str: """Return the assistant message text.""" def read_allowed_dotenv(path: Optional[Path] = None) -> Dict[str, str]: """Read only non-secret OCI routing/auth-mode settings from .env.""" selected_path = DOTENV_PATH if path is None else path try: lines = selected_path.read_text(encoding="utf-8").splitlines() except OSError: return {} values: Dict[str, str] = {} for raw_line in lines: line = raw_line.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[7:].lstrip() key, separator, raw_value = line.partition("=") key = key.strip() if not separator or key not in ALLOWED_OCI_SETTINGS: continue value = raw_value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: value = value[1:-1] if "\x00" not in value and "\n" not in value and "\r" not in value: values[key] = value return values @dataclass(frozen=True, repr=False) class OCISettings: auth_type: str config_file: str profile: str compartment_id: str def load_oci_settings() -> OCISettings: """Resolve OCI GenAI settings from safe .env keys and environment.""" values = read_allowed_dotenv() for key in ALLOWED_OCI_SETTINGS: value = os.environ.get(key) if isinstance(value, str) and value.strip(): values[key] = value.strip() auth_type = values.get("OCI_AUTH_TYPE", "config_file").strip().casefold() auth_type = auth_type.replace("-", "_") if auth_type in {"api_key", "config", "config_file"}: auth_type = "config_file" elif auth_type not in {"instance_principal", "resource_principal"}: raise ValueError("unsupported OCI authentication mode") compartment_id = values.get("OCI_GENAI_COMPARTMENT_ID", "").strip() if not _COMPARTMENT_ID.fullmatch(compartment_id): raise ValueError("OCI Generative AI compartment is not configured") return OCISettings( auth_type=auth_type, config_file=values.get("OCI_CONFIG_FILE", "~/.oci/config").strip(), profile=values.get("OCI_PROFILE", "DEFAULT").strip() or "DEFAULT", compartment_id=compartment_id, ) class OCICompletionClient: """Minimal OCI GenericChatRequest adapter.""" def __init__( self, settings: OCISettings, model_id: str, region: str, endpoint: str, ) -> None: try: import oci from oci.generative_ai_inference import GenerativeAiInferenceClient except (ImportError, AttributeError): raise RuntimeError("OCI SDK is unavailable") from None kwargs: Dict[str, object] = {} if settings.auth_type == "config_file": try: config = oci.config.from_file( file_location=os.path.expandvars( os.path.expanduser(settings.config_file) ), profile_name=settings.profile, ) except Exception: raise RuntimeError("OCI SDK configuration is unavailable") from None config["region"] = region elif settings.auth_type == "instance_principal": try: signer = oci.auth.signers.InstancePrincipalsSecurityTokenSigner() except Exception: raise RuntimeError("OCI signer is unavailable") from None config = {"region": region} kwargs["signer"] = signer else: try: signer = oci.auth.signers.get_resource_principals_signer() except Exception: raise RuntimeError("OCI signer is unavailable") from None config = {"region": region} kwargs["signer"] = signer if not config.get("region"): raise RuntimeError("OCI region is unavailable") kwargs["service_endpoint"] = endpoint try: self._client = GenerativeAiInferenceClient(config, **kwargs) except Exception: raise RuntimeError("OCI Generative AI client is unavailable") from None self._compartment_id = settings.compartment_id self._model_id = model_id def complete( self, system_prompt: str, user_prompt: str, response_schema: Mapping[str, object], max_tokens: int, temperature: Optional[float], ) -> str: try: from oci.generative_ai_inference.models import ( ChatDetails, GenericChatRequest, JsonSchemaResponseFormat, OnDemandServingMode, ResponseJsonSchema, SystemMessage, TextContent, UserMessage, ) schema = ResponseJsonSchema( name="oci_genai_json_response", description="Strict JSON response", schema=dict(response_schema), is_strict=True, ) request_options: Dict[str, object] = { "api_format": "GENERIC", "messages": [ SystemMessage(content=[TextContent(text=system_prompt)]), UserMessage(content=[TextContent(text=user_prompt)]), ], "max_completion_tokens": max_tokens, "is_stream": False, "response_format": JsonSchemaResponseFormat(json_schema=schema), } if temperature is not None: request_options["temperature"] = temperature request = GenericChatRequest(**request_options) details = ChatDetails( compartment_id=self._compartment_id, serving_mode=OnDemandServingMode(model_id=self._model_id), chat_request=request, ) response = self._client.chat(details) data = getattr(response, "data", None) chat_response = getattr(data, "chat_response", None) choices = getattr(chat_response, "choices", None) if not isinstance(choices, Sequence) or not choices: raise RuntimeError("OCI response has no choice") message = getattr(choices[0], "message", None) content = getattr(message, "content", None) if not isinstance(content, Sequence) or isinstance( content, (str, bytes, bytearray) ) or not content: raise RuntimeError("OCI response has no content") text = getattr(content[0], "text", None) if not isinstance(text, str): raise RuntimeError("OCI response content is invalid") return text except Exception: raise RuntimeError("OCI GenAI completion call failed") from None def build_oci_genai_completion_client( model_id: str, region: str, endpoint: str, ) -> CompletionClient: """Build a completion client for one validated model route.""" return _cached_oci_genai_completion_client( load_oci_settings(), model_id, region, endpoint, ) @lru_cache(maxsize=16) def _cached_oci_genai_completion_client( settings: OCISettings, model_id: str, region: str, endpoint: str, ) -> CompletionClient: """Reuse OCI GenAI clients within one Python process.""" return OCICompletionClient(settings, model_id, region, endpoint) def temperature_for_model_key(model_key: object) -> Optional[float]: """Return provider-compatible temperature for one registered model key.""" key = str(model_key or "").strip().lower() if key.startswith("gpt"): return None return 0.1 def temperature_for_model_profile(profile: object) -> Optional[float]: return temperature_for_model_key(getattr(profile, "model_key", profile)) __all__ = [ "ALLOWED_OCI_SETTINGS", "CompletionClient", "OCICompletionClient", "OCISettings", "build_oci_genai_completion_client", "load_oci_settings", "read_allowed_dotenv", "temperature_for_model_key", "temperature_for_model_profile", ]