From afc0e671e6b5a717a7499279fa58944f89c9f1d5 Mon Sep 17 00:00:00 2001 From: akash-vijay-kv Date: Thu, 5 Mar 2026 10:11:42 +0530 Subject: [PATCH] [NET-333] feat: Add SDK support for prompt management feature --- netra/__init__.py | 11 +++- netra/prompts/__init__.py | 3 + netra/prompts/api.py | 40 +++++++++++ netra/prompts/client.py | 135 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 188 insertions(+), 1 deletion(-) create mode 100644 netra/prompts/__init__.py create mode 100644 netra/prompts/api.py create mode 100644 netra/prompts/client.py diff --git a/netra/__init__.py b/netra/__init__.py index 6989f8f..033f973 100644 --- a/netra/__init__.py +++ b/netra/__init__.py @@ -13,6 +13,7 @@ from netra.instrumentation import init_instrumentations from netra.instrumentation.instruments import NetraInstruments from netra.logging_utils import configure_package_logging +from netra.prompts import Prompts from netra.session_manager import ConversationType, SessionManager from netra.simulation import Simulation from netra.span_wrapper import ActionModel, SpanType, SpanWrapper, UsageModel @@ -23,6 +24,7 @@ "Netra", "UsageModel", "ActionModel", + "Prompts", ] logger = logging.getLogger(__name__) @@ -133,6 +135,13 @@ def init( logger.warning("Failed to initialize dashboard client: %s", e, exc_info=True) cls.dashboard = None # type:ignore[attr-defined] + # Initialize prompts client and expose as class attribute + try: + cls.prompts = Prompts(cfg) # type:ignore[attr-defined] + except Exception as e: + logger.warning("Failed to initialize prompts client: %s", e, exc_info=True) + cls.prompts = None # type:ignore[attr-defined] + # Initialize simulation client and expose as class attribute try: cls.simulation = Simulation(cfg) # type:ignore[attr-defined] @@ -313,4 +322,4 @@ def start_span( return SpanWrapper(name, attributes, module_name, as_type=as_type) -__all__ = ["Netra", "UsageModel", "ActionModel", "SpanType", "EvaluationScore"] +__all__ = ["Netra", "UsageModel", "ActionModel", "SpanType", "EvaluationScore", "Prompts"] diff --git a/netra/prompts/__init__.py b/netra/prompts/__init__.py new file mode 100644 index 0000000..33e400a --- /dev/null +++ b/netra/prompts/__init__.py @@ -0,0 +1,3 @@ +from netra.prompts.api import Prompts + +__all__ = ["Prompts"] diff --git a/netra/prompts/api.py b/netra/prompts/api.py new file mode 100644 index 0000000..c85ed71 --- /dev/null +++ b/netra/prompts/api.py @@ -0,0 +1,40 @@ +import logging +from typing import Any + +from netra.config import Config +from netra.prompts.client import PromptsHttpClient + +logger = logging.getLogger(__name__) + + +class Prompts: + """ + Public entry-point exposed as Netra.prompts + """ + + def __init__(self, cfg: Config) -> None: + """ + Initialize the Prompts client. + + Args: + cfg: Configuration object containing API key and base URL + """ + self._config = cfg + self._client = PromptsHttpClient(cfg) + + def get_prompt(self, name: str, label: str = "production") -> Any: + """ + Fetch a prompt version by name and label. + + Args: + name: Name of the prompt + label: Label of the prompt version (default: "production") + + Returns: + Prompt version data or empty dict if not found + """ + if not name: + logger.error("netra.prompts: name is required to fetch a prompt") + return None + + return self._client.get_prompt_version(prompt_name=name, label=label) diff --git a/netra/prompts/client.py b/netra/prompts/client.py new file mode 100644 index 0000000..f23de36 --- /dev/null +++ b/netra/prompts/client.py @@ -0,0 +1,135 @@ +import logging +import os +from typing import Any, Dict, Optional + +import httpx + +from netra.config import Config + +logger = logging.getLogger(__name__) + + +class PromptsHttpClient: + """ + Internal HTTP client for prompts APIs. + """ + + def __init__(self, config: Config) -> None: + """ + Initialize the prompts HTTP client. + + Args: + config: Configuration object containing API key and base URL + """ + self._client: Optional[httpx.Client] = self._create_client(config) + + def _create_client(self, config: Config) -> Optional[httpx.Client]: + """ + Create and configure the HTTP client. + + Args: + config: Configuration object containing API key and base URL + + Returns: + Configured HTTP client or None if initialization fails + """ + endpoint = (config.otlp_endpoint or "").strip() + if not endpoint: + logger.error("netra.prompts: NETRA_OTLP_ENDPOINT is required for prompts APIs") + return None + + base_url = self._resolve_base_url(endpoint) + headers = self._build_headers(config) + timeout = self._get_timeout() + + try: + return httpx.Client(base_url=base_url, headers=headers, timeout=timeout) + except Exception as exc: + logger.error("netra.prompts: Failed to initialize prompts HTTP client: %s", exc) + return None + + def _resolve_base_url(self, endpoint: str) -> str: + """ + Resolve the base URL by removing /telemetry suffix if present. + + Args: + endpoint: The endpoint URL + + Returns: + Resolved base URL + """ + base_url = endpoint.rstrip("/") + if base_url.endswith("/telemetry"): + base_url = base_url[: -len("/telemetry")] + return base_url + + def _build_headers(self, config: Config) -> Dict[str, str]: + """ + Build HTTP headers for API requests. + + Args: + config: Configuration object containing API key and base URL + + Returns: + Dictionary of HTTP headers + """ + headers: Dict[str, str] = dict(config.headers or {}) + api_key = config.api_key + if api_key: + headers["x-api-key"] = api_key + return headers + + def _get_timeout(self) -> float: + """ + Get the timeout value from environment variable or use default. + + Returns: + Timeout value in seconds + """ + timeout_env = os.getenv("NETRA_PROMPTS_TIMEOUT") + if not timeout_env: + return 10.0 + try: + return float(timeout_env) + except ValueError: + logger.warning( + "netra.prompts: Invalid NETRA_PROMPTS_TIMEOUT value '%s', using default 10.0", + timeout_env, + ) + return 10.0 + + def get_prompt_version(self, prompt_name: str, label: str) -> Any: + """ + Fetch a prompt version by name and label. + + Args: + prompt_name: Name of the prompt + label: Label of the prompt version + + Returns: + Prompt version data or empty dict if not found + """ + if not self._client: + logger.error( + "netra.prompts: Prompts client is not initialized; cannot fetch prompt version for '%s'", + prompt_name, + ) + return {} + + try: + url = "/sdk/prompts/version" + payload: Dict[str, Any] = {"promptName": prompt_name, "label": label} + response = self._client.post(url, json=payload) + response.raise_for_status() + data = response.json() + if isinstance(data, dict) and "data" in data: + return data.get("data", {}) + return data + except Exception as exc: + logger.error( + "netra.prompts: Failed to fetch prompt version for '%s' (label=%s): %s", + prompt_name, + label, + exc, + ) + return {}