feat: add evaluations module scaffold, credentials, LD API client, result types

packages/client/src/launchdarkly_ai_server/evaluations/__init__.py

@@ -0,0 +1,28 @@
+"""Run LaunchDarkly evaluations from your own environment."""
+
+from .api import (
+    DEFAULT_BASE_URI,
+    EvaluationsError,
+    HttpResponse,
+    LDApiClient,
+    LDApiError,
+    Transport,
+    urllib_transport,
+)
+from .module import EvaluationsModule, init_evaluations
+from .types import EvalRunResult, RunSummary, Usage
+
+__all__ = [
+    "DEFAULT_BASE_URI",
+    "EvalRunResult",
+    "EvaluationsError",
+    "EvaluationsModule",
+    "HttpResponse",
+    "LDApiClient",
+    "LDApiError",
+    "RunSummary",
+    "Transport",
+    "Usage",
+    "init_evaluations",
+    "urllib_transport",
+]

packages/client/src/launchdarkly_ai_server/evaluations/api.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass, field
+from typing import Any, Protocol
+
+DEFAULT_BASE_URI = "https://app.launchdarkly.com"
+
+
+class EvaluationsError(Exception):
+    """Base error for the evaluations harness."""
+
+
+class LDApiError(EvaluationsError):
+    """A non-2xx response from the LaunchDarkly API."""
+
+    def __init__(self, status: int, method: str, path: str, body: str) -> None:
+        super().__init__(
+            f"LaunchDarkly API {method} {path} failed with {status}: {body}"
+        )
+        self.status = status
+        self.method = method
+        self.path = path
+        self.body = body
+
+
+@dataclass
+class HttpResponse:
+    status: int
+    body: str
+    headers: dict[str, str] = field(default_factory=dict)
+
+
+class Transport(Protocol):
+    """Seam the API client sends requests through; replaced in tests."""
+
+    def __call__(
+        self,
+        method: str,
+        url: str,
+        headers: dict[str, str],
+        body: bytes | None,
+        timeout: float,
+    ) -> HttpResponse: ...
+
+
+def urllib_transport(
+    method: str,
+    url: str,
+    headers: dict[str, str],
+    body: bytes | None,
+    timeout: float,
+) -> HttpResponse:
+    request = urllib.request.Request(url, data=body, headers=headers, method=method)
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            return HttpResponse(
+                status=response.status,
+                body=response.read().decode("utf-8"),
+                headers={k.lower(): v for k, v in response.headers.items()},
+            )
+    except urllib.error.HTTPError as error:
+        return HttpResponse(
+            status=error.code,
+            body=error.read().decode("utf-8"),
+            headers={k.lower(): v for k, v in error.headers.items()},
+        )
+
+
+class LDApiClient:
+    """
+    Minimal client for the LaunchDarkly public ``/api/v2`` surface used by the
+    evaluations harness. Every request carries the API access token; the base
+    URI is overridable for non-default instances.
+    """
+
+    def __init__(
+        self,
+        api_token: str,
+        base_uri: str = DEFAULT_BASE_URI,
+        transport: Transport = urllib_transport,
+        timeout: float = 30.0,
+    ) -> None:
+        self.api_token = api_token
+        self.base_uri = base_uri.rstrip("/")
+        self._transport = transport
+        self._timeout = timeout
+
+    def url_for(self, path: str, params: dict[str, Any] | None = None) -> str:
+        url = f"{self.base_uri}/api/v2/{path.lstrip('/')}"
+        if params:
+            query = {k: str(v) for k, v in params.items() if v is not None}
+            if query:
+                url = f"{url}?{urllib.parse.urlencode(query)}"
+        return url
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        body: Any = None,
+        params: dict[str, Any] | None = None,
+    ) -> Any:
+        headers = {
+            "Authorization": self.api_token,
+            "Accept": "application/json",
+            "User-Agent": "launchdarkly-ai-evaluations-python",
+        }
+        payload: bytes | None = None
+        if body is not None:
+            headers["Content-Type"] = "application/json"
+            payload = json.dumps(body).encode("utf-8")
+
+        response = self._transport(
+            method, self.url_for(path, params), headers, payload, self._timeout
+        )
+        if response.status < 200 or response.status >= 300:
+            raise LDApiError(response.status, method, path, response.body)
+        if not response.body:
+            return None
+        return json.loads(response.body)
+
+    def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
+        return self.request("GET", path, params=params)
+
+    def post(self, path: str, body: Any = None) -> Any:
+        return self.request("POST", path, body=body)

packages/client/src/launchdarkly_ai_server/evaluations/module.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import logging
+import os
+
+from .api import (
+    DEFAULT_BASE_URI,
+    EvaluationsError,
+    LDApiClient,
+    Transport,
+    urllib_transport,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _env(name: str) -> str | None:
+    """Read an env var, treating blank/whitespace-only values as unset."""
+    value = os.environ.get(name, "").strip()
+    return value if value else None
+
+
+class EvaluationsModule:
+    """
+    Entry point for running LaunchDarkly evaluations from code. Holds the
+    resolved credentials and the LaunchDarkly API client; ``run()`` arrives with
+    the harness.
+    """
+
+    def __init__(self, api_client: LDApiClient, sdk_key: str | None = None) -> None:
+        self._api = api_client
+        self._sdk_key = sdk_key
+
+    @property
+    def api(self) -> LDApiClient:
+        return self._api
+
+    @property
+    def sdk_key(self) -> str | None:
+        """SDK key used for observability traces; ``None`` disables tracing."""
+        return self._sdk_key
+
+
+def init_evaluations(
+    api_token: str | None = None,
+    sdk_key: str | None = None,
+    base_uri: str | None = None,
+    transport: Transport = urllib_transport,
+) -> EvaluationsModule:
+    """
+    Resolves credentials and builds the evaluations module.
+
+    ``api_token`` (``LD_API_TOKEN``) authenticates every ``/api/v2`` call and is
+    required — a missing token raises before any network I/O rather than
+    surfacing as an opaque 401 mid-run. ``sdk_key`` (``LD_SDK_KEY``) is optional
+    and only makes handler calls emit observability traces. Both credentials
+    must point at the same project.
+    """
+    token = api_token or _env("LD_API_TOKEN")
+    if not token:
+        raise EvaluationsError(
+            "No LaunchDarkly API access token provided. Set the LD_API_TOKEN "
+            "environment variable or pass api_token to init_evaluations()."
+        )
+
+    resolved_sdk_key = sdk_key or _env("LD_SDK_KEY")
+    if not resolved_sdk_key:
+        logger.info(
+            "No LaunchDarkly SDK key provided; evaluation runs will not emit traces."
+        )
+
+    api_client = LDApiClient(
+        api_token=token,
+        base_uri=base_uri or _env("LD_BASE_URI") or DEFAULT_BASE_URI,
+        transport=transport,
+    )
+    return EvaluationsModule(api_client=api_client, sdk_key=resolved_sdk_key)

packages/client/src/launchdarkly_ai_server/evaluations/types.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass
+class Usage:
+    """
+    Token counts for a single generation, in the ingest wire shape. Handler
+    results carry this dict verbatim, so nothing on the eval path adapts it.
+    """
+
+    input_tokens: int
+    output_tokens: int
+
+    def to_wire(self) -> dict[str, int]:
+        return {
+            "input_tokens": self.input_tokens,
+            "output_tokens": self.output_tokens,
+        }
+
+    @classmethod
+    def from_wire(cls, data: dict[str, Any]) -> Usage:
+        return cls(
+            input_tokens=int(data.get("input_tokens") or 0),
+            output_tokens=int(data.get("output_tokens") or 0),
+        )
+
+
+@dataclass
+class RunSummary:
+    """Row counts for a finished evaluation run."""
+
+    total_rows: int = 0
+    passed_rows: int = 0
+    failed_rows: int = 0
+    error_rows: int = 0
+
+    @classmethod
+    def from_wire(cls, data: dict[str, Any] | None) -> RunSummary:
+        data = data or {}
+        return cls(
+            total_rows=int(data.get("total_rows") or 0),
+            passed_rows=int(data.get("passed_rows") or 0),
+            failed_rows=int(data.get("failed_rows") or 0),
+            error_rows=int(data.get("error_rows") or 0),
+        )
+
+
+@dataclass
+class EvalRunResult:
+    """The verdict of an evaluation run, as computed and stored by LaunchDarkly."""
+
+    passed: bool
+    url: str
+    run_id: str
+    summary: RunSummary