feat(api): add client bootstrap and service facade

py/src/braintrust/api/__init__.py

@@ -1,5 +1,11 @@
-"""Braintrust API client package."""
+"""Public Braintrust API client package."""
 
+from ._routing import EndpointRouter, RequestTarget
+from ._service import ClientContext
+from .attachments import AttachmentsAPI
+from .auth import AuthAPI, LoginResult, OrganizationInfo
+from .client import BraintrustClient
+from .datasets import DatasetsAPI
 from .errors import (
  BraintrustAPIError,
  BraintrustHTTPError,
@@ -8,16 +14,35 @@
  BraintrustTransportError,
  BraintrustTransportRetryExhaustedError,
 )
+from .experiments import ExperimentsAPI
+from .functions import FunctionsAPI
 from .policies import RetryMode, RetryPolicy
+from .projects import ProjectsAPI
+from .prompts import PromptsAPI
+from .queries import QueriesAPI
 
 
 __all__ = [
+ "AttachmentsAPI",
+ "AuthAPI",
  "BraintrustAPIError",
+ "BraintrustClient",
  "BraintrustHTTPError",
  "BraintrustResponseError",
  "BraintrustRetryExhaustedError",
  "BraintrustTransportError",
  "BraintrustTransportRetryExhaustedError",
+ "ClientContext",
+ "DatasetsAPI",
+ "EndpointRouter",
+ "ExperimentsAPI",
+ "FunctionsAPI",
+ "LoginResult",
+ "OrganizationInfo",
+ "ProjectsAPI",
+ "PromptsAPI",
+ "QueriesAPI",
+ "RequestTarget",
  "RetryMode",
  "RetryPolicy",
 ]

py/src/braintrust/api/_routing.py

@@ -0,0 +1,69 @@
+"""Endpoint routing for Braintrust API requests."""
+
+import enum
+from dataclasses import dataclass
+
+from ..util import _urljoin
+
+
+_V1_PROXY_SUFFIX = "/v1/proxy"
+
+
+class RequestTarget(enum.Enum):
+ """A logical Braintrust request destination."""
+
+ APP = "app"
+ API = "api"
+ PROXY = "proxy"
+
+
+def normalize_proxy_url(proxy_url: str) -> str:
+ """Normalize a Universal Proxy URL to the API host used by SDK routes."""
+
+ if proxy_url.endswith(_V1_PROXY_SUFFIX):
+ return proxy_url[: -len(_V1_PROXY_SUFFIX)]
+ return proxy_url
+
+
+@dataclass
+class EndpointRouter:
+ """Resolve logical Braintrust targets without changing their configured origins."""
+
+ app_url: str
+ api_url: str | None = None
+ proxy_url: str | None = None
+ is_universal_api: bool = False
+
+ def configure(
+ self,
+ *,
+ api_url: str | None,
+ proxy_url: str | None,
+ is_universal_api: bool = False,
+ ) -> None:
+ """Apply URLs discovered during authentication."""
+
+ self.api_url = api_url
+ self.proxy_url = proxy_url
+ self.is_universal_api = is_universal_api
+
+ def base_url(self, target: RequestTarget) -> str:
+ """Return the configured origin for ``target``."""
+
+ if target is RequestTarget.APP:
+ return self.app_url
+ if target is RequestTarget.API:
+ if not self.api_url:
+ raise RuntimeError("API URL is unavailable before organization discovery")
+ return self.api_url
+ if target is RequestTarget.PROXY:
+ base_url = self.proxy_url or self.api_url
+ if not base_url:
+ raise RuntimeError("Proxy URL is unavailable before organization discovery")
+ return normalize_proxy_url(base_url)
+ raise ValueError(f"Unknown request target: {target!r}")
+
+ def resolve(self, target: RequestTarget, path: str) -> str:
+ """Resolve ``path`` against the origin for ``target``."""
+
+ return _urljoin(self.base_url(target), path)

py/src/braintrust/api/_service.py

@@ -0,0 +1,51 @@
+"""Shared resource service primitives."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import Transport
+
+
+@dataclass(frozen=True)
+class ClientContext:
+ """Organization and credential context shared by resource services."""
+
+ org_id: str
+ org_name: str
+
+
+class ResourceAPI:
+ """Base class for synchronous resource services."""
+
+ def __init__(
+ self,
+ transport: Transport,
+ router: EndpointRouter,
+ context: ClientContext,
+ api_key: str,
+ ):
+ self._transport = transport
+ self._router = router
+ self._context = context
+ self._api_key = api_key
+
+ def _request_json(
+ self,
+ target: RequestTarget,
+ method: str,
+ path: str,
+ *,
+ headers: Mapping[str, str] | None = None,
+ **kwargs: Any,
+ ) -> Any:
+ request_headers = {"Authorization": f"Bearer {self._api_key}"}
+ if headers:
+ request_headers.update(headers)
+ return self._transport.request_json(
+ method,
+ self._router.resolve(target, path),
+ headers=request_headers,
+ **kwargs,
+ )

py/src/braintrust/api/_transport.py

@@ -1,6 +1,7 @@
 """Legacy and policy-aware HTTP transport primitives for the Braintrust SDK."""
 
 import datetime
+import http.cookiejar
 import logging
 import sys
 import time
@@ -28,6 +29,11 @@
 logger = logging.getLogger(__name__)
 
 
+class _RejectCookiesPolicy(http.cookiejar.DefaultCookiePolicy):
+ def set_ok(self, cookie: Any, request: Any) -> bool:
+ return False
+
+
 class RetryRequestExceptionsAdapter(HTTPAdapter):
  """An HTTP adapter that automatically retries requests on connection exceptions.
 
@@ -189,13 +195,16 @@ def __init__(
  session: requests.Session | None = None,
  adapter: HTTPAdapter | None = None,
  enable_sdk_retries: bool | None = None,
+ persist_cookies: bool = True,
  sleep: Callable[[float], None] = time.sleep,
  monotonic: Callable[[], float] = time.monotonic,
  wall_clock: Callable[[], float] = time.time,
  ):
  custom_transport = session is not None or adapter is not None
  self._owns_session = session is None
  self.session = session if session is not None else requests.Session()
+ if not persist_cookies and self._owns_session:
+ self.session.cookies.set_policy(_RejectCookiesPolicy())
  self._sdk_retries_enabled = not custom_transport if enable_sdk_retries is None else enable_sdk_retries
  if adapter is not None:
  self.session.mount("http://", adapter)

py/src/braintrust/api/attachments.py

@@ -0,0 +1,10 @@
+"""Attachment metadata API service."""
+
+from ._service import ResourceAPI
+
+
+class AttachmentsAPI(ResourceAPI):
+ """Synchronous attachment metadata operations.
+
+ Signed object-storage traffic remains outside the routed transport.
+ """

py/src/braintrust/api/auth.py

@@ -0,0 +1,122 @@
+"""Authentication and organization discovery service."""
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Any
+
+from ..env import BraintrustEnv
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import HTTPConnection, Transport
+from .policies import RetryMode
+
+
+@dataclass(frozen=True)
+class OrganizationInfo:
+ """Organization routing information returned by API-key login."""
+
+ id: str
+ name: str
+ api_url: str | None
+ proxy_url: str | None
+ realtime_url: str | None
+ is_universal_api: bool
+ git_metadata: Mapping[str, Any] | None
+ raw: Mapping[str, Any]
+
+ @classmethod
+ def from_dict(cls, value: Mapping[str, Any]) -> "OrganizationInfo":
+ """Parse an additive login response while retaining unknown fields."""
+
+ org_id = value.get("id")
+ name = value.get("name")
+ if not isinstance(org_id, str) or not isinstance(name, str):
+ raise ValueError("Organization login data must include string id and name fields")
+
+ def optional_string(field: str) -> str | None:
+ result = value.get(field)
+ return result if isinstance(result, str) and result else None
+
+ git_metadata = value.get("git_metadata")
+ if not isinstance(git_metadata, Mapping):
+ git_metadata = None
+
+ return cls(
+ id=org_id,
+ name=name,
+ api_url=optional_string("api_url"),
+ proxy_url=optional_string("proxy_url"),
+ realtime_url=optional_string("realtime_url"),
+ is_universal_api=bool(value.get("is_universal_api", False)),
+ git_metadata=MappingProxyType(dict(git_metadata)) if git_metadata is not None else None,
+ raw=MappingProxyType(dict(value)),
+ )
+
+
+@dataclass(frozen=True)
+class LoginResult:
+ """Selected organization and the complete login response."""
+
+ organization: OrganizationInfo
+ response: Mapping[str, Any]
+
+
+class AuthAPI:
+ """Authenticate an API key and configure an endpoint router."""
+
+ def __init__(self, transport: Transport, router: EndpointRouter):
+ self._transport = transport
+ self._router = router
+
+ def login(
+ self,
+ api_key: str,
+ *,
+ org_name: str | None = None,
+ api_url: str | None = None,
+ proxy_url: str | None = None,
+ ) -> LoginResult:
+ """Log in, select an organization, and apply routing override precedence."""
+
+ api_key = HTTPConnection.sanitize_token(api_key)
+ response = self._transport.request_json(
+ "POST",
+ self._router.resolve(RequestTarget.APP, "/api/apikey/login"),
+ headers={"Authorization": f"Bearer {api_key}"},
+ retry_mode=RetryMode.SAFE_READ,
+ )
+ if not isinstance(response, Mapping):
+ raise ValueError("API-key login returned a non-object response")
+ raw_orgs = response.get("org_info")
+ if not isinstance(raw_orgs, Sequence) or isinstance(raw_orgs, (str, bytes)):
+ raise ValueError("API-key login response did not include an organization list")
+
+ organizations = [OrganizationInfo.from_dict(org) for org in raw_orgs if isinstance(org, Mapping)]
+ organization = self._select_organization(organizations, org_name)
+
+ resolved_api_url = api_url or BraintrustEnv.API_URL.get(organization.api_url)
+ resolved_proxy_url = proxy_url or BraintrustEnv.PROXY_URL.get(organization.proxy_url)
+ if not resolved_api_url:
+ if org_name:
+ raise ValueError(
+ f"Unable to log into organization '{org_name}'."
+ " Are you sure this credential is scoped to the organization?"
+ )
+ raise ValueError("Unable to log into any organization with the provided credential.")
+
+ self._router.configure(
+ api_url=resolved_api_url,
+ proxy_url=resolved_proxy_url,
+ is_universal_api=organization.is_universal_api,
+ )
+ return LoginResult(organization=organization, response=MappingProxyType(dict(response)))
+
+ @staticmethod
+ def _select_organization(organizations: Sequence[OrganizationInfo], org_name: str | None) -> OrganizationInfo:
+ if not organizations:
+ raise ValueError("This user is not part of any organizations.")
+ for organization in organizations:
+ if org_name is None or organization.name == org_name:
+ return organization
+ choices = ", ".join(organization.name for organization in organizations)
+ raise ValueError(f"Organization {org_name} not found. Must be one of {choices}")

py/src/braintrust/api/__init__.py

@@ -1,5 +1,11 @@
-"""Braintrust API client package."""
+"""Public Braintrust API client package."""
 
+from ._routing import EndpointRouter, RequestTarget
+from ._service import ClientContext
+from .attachments import AttachmentsAPI
+from .auth import AuthAPI, LoginResult, OrganizationInfo
+from .client import BraintrustClient
+from .datasets import DatasetsAPI
 from .errors import (
  BraintrustAPIError,
  BraintrustHTTPError,
@@ -8,16 +14,35 @@
  BraintrustTransportError,
  BraintrustTransportRetryExhaustedError,
 )
+from .experiments import ExperimentsAPI
+from .functions import FunctionsAPI
 from .policies import RetryMode, RetryPolicy
+from .projects import ProjectsAPI
+from .prompts import PromptsAPI
+from .queries import QueriesAPI
 
 
 __all__ = [
+ "AttachmentsAPI",
+ "AuthAPI",
  "BraintrustAPIError",
+ "BraintrustClient",
  "BraintrustHTTPError",
  "BraintrustResponseError",
  "BraintrustRetryExhaustedError",
  "BraintrustTransportError",
  "BraintrustTransportRetryExhaustedError",
+ "ClientContext",
+ "DatasetsAPI",
+ "EndpointRouter",
+ "ExperimentsAPI",
+ "FunctionsAPI",
+ "LoginResult",
+ "OrganizationInfo",
+ "ProjectsAPI",
+ "PromptsAPI",
+ "QueriesAPI",
+ "RequestTarget",
  "RetryMode",
  "RetryPolicy",
 ]

py/src/braintrust/api/_routing.py

@@ -0,0 +1,69 @@
+"""Endpoint routing for Braintrust API requests."""
+
+import enum
+from dataclasses import dataclass
+
+from ..util import _urljoin
+
+
+_V1_PROXY_SUFFIX = "/v1/proxy"
+
+
+class RequestTarget(enum.Enum):
+ """A logical Braintrust request destination."""
+
+ APP = "app"
+ API = "api"
+ PROXY = "proxy"
+
+
+def normalize_proxy_url(proxy_url: str) -> str:
+ """Normalize a Universal Proxy URL to the API host used by SDK routes."""
+
+ if proxy_url.endswith(_V1_PROXY_SUFFIX):
+ return proxy_url[: -len(_V1_PROXY_SUFFIX)]
+ return proxy_url
+
+
+@dataclass
+class EndpointRouter:
+ """Resolve logical Braintrust targets without changing their configured origins."""
+
+ app_url: str
+ api_url: str | None = None
+ proxy_url: str | None = None
+ is_universal_api: bool = False
+
+ def configure(
+ self,
+ *,
+ api_url: str | None,
+ proxy_url: str | None,
+ is_universal_api: bool = False,
+ ) -> None:
+ """Apply URLs discovered during authentication."""
+
+ self.api_url = api_url
+ self.proxy_url = proxy_url
+ self.is_universal_api = is_universal_api
+
+ def base_url(self, target: RequestTarget) -> str:
+ """Return the configured origin for ``target``."""
+
+ if target is RequestTarget.APP:
+ return self.app_url
+ if target is RequestTarget.API:
+ if not self.api_url:
+ raise RuntimeError("API URL is unavailable before organization discovery")
+ return self.api_url
+ if target is RequestTarget.PROXY:
+ base_url = self.proxy_url or self.api_url
+ if not base_url:
+ raise RuntimeError("Proxy URL is unavailable before organization discovery")
+ return normalize_proxy_url(base_url)
+ raise ValueError(f"Unknown request target: {target!r}")
+
+ def resolve(self, target: RequestTarget, path: str) -> str:
+ """Resolve ``path`` against the origin for ``target``."""
+
+ return _urljoin(self.base_url(target), path)

py/src/braintrust/api/_service.py

@@ -0,0 +1,51 @@
+"""Shared resource service primitives."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import Transport
+
+
+@dataclass(frozen=True)
+class ClientContext:
+ """Organization and credential context shared by resource services."""
+
+ org_id: str
+ org_name: str
+
+
+class ResourceAPI:
+ """Base class for synchronous resource services."""
+
+ def __init__(
+ self,
+ transport: Transport,
+ router: EndpointRouter,
+ context: ClientContext,
+ api_key: str,
+ ):
+ self._transport = transport
+ self._router = router
+ self._context = context
+ self._api_key = api_key
+
+ def _request_json(
+ self,
+ target: RequestTarget,
+ method: str,
+ path: str,
+ *,
+ headers: Mapping[str, str] | None = None,
+ **kwargs: Any,
+ ) -> Any:
+ request_headers = {"Authorization": f"Bearer {self._api_key}"}
+ if headers:
+ request_headers.update(headers)
+ return self._transport.request_json(
+ method,
+ self._router.resolve(target, path),
+ headers=request_headers,
+ **kwargs,
+ )

py/src/braintrust/api/_transport.py

@@ -1,6 +1,7 @@
 """Legacy and policy-aware HTTP transport primitives for the Braintrust SDK."""
 
 import datetime
+import http.cookiejar
 import logging
 import sys
 import time
@@ -28,6 +29,11 @@
 logger = logging.getLogger(__name__)
 
 
+class _RejectCookiesPolicy(http.cookiejar.DefaultCookiePolicy):
+ def set_ok(self, cookie: Any, request: Any) -> bool:
+ return False
+
+
 class RetryRequestExceptionsAdapter(HTTPAdapter):
  """An HTTP adapter that automatically retries requests on connection exceptions.
 
@@ -189,13 +195,16 @@ def __init__(
  session: requests.Session | None = None,
  adapter: HTTPAdapter | None = None,
  enable_sdk_retries: bool | None = None,
+ persist_cookies: bool = True,
  sleep: Callable[[float], None] = time.sleep,
  monotonic: Callable[[], float] = time.monotonic,
  wall_clock: Callable[[], float] = time.time,
  ):
  custom_transport = session is not None or adapter is not None
  self._owns_session = session is None
  self.session = session if session is not None else requests.Session()
+ if not persist_cookies and self._owns_session:
+ self.session.cookies.set_policy(_RejectCookiesPolicy())
  self._sdk_retries_enabled = not custom_transport if enable_sdk_retries is None else enable_sdk_retries
  if adapter is not None:
  self.session.mount("http://", adapter)

py/src/braintrust/api/attachments.py

@@ -0,0 +1,10 @@
+"""Attachment metadata API service."""
+
+from ._service import ResourceAPI
+
+
+class AttachmentsAPI(ResourceAPI):
+ """Synchronous attachment metadata operations.
+
+ Signed object-storage traffic remains outside the routed transport.
+ """

py/src/braintrust/api/auth.py

@@ -0,0 +1,122 @@
+"""Authentication and organization discovery service."""
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Any
+
+from ..env import BraintrustEnv
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import HTTPConnection, Transport
+from .policies import RetryMode
+
+
+@dataclass(frozen=True)
+class OrganizationInfo:
+ """Organization routing information returned by API-key login."""
+
+ id: str
+ name: str
+ api_url: str | None
+ proxy_url: str | None
+ realtime_url: str | None
+ is_universal_api: bool
+ git_metadata: Mapping[str, Any] | None
+ raw: Mapping[str, Any]
+
+ @classmethod
+ def from_dict(cls, value: Mapping[str, Any]) -> "OrganizationInfo":
+ """Parse an additive login response while retaining unknown fields."""
+
+ org_id = value.get("id")
+ name = value.get("name")
+ if not isinstance(org_id, str) or not isinstance(name, str):
+ raise ValueError("Organization login data must include string id and name fields")
+
+ def optional_string(field: str) -> str | None:
+ result = value.get(field)
+ return result if isinstance(result, str) and result else None
+
+ git_metadata = value.get("git_metadata")
+ if not isinstance(git_metadata, Mapping):
+ git_metadata = None
+
+ return cls(
+ id=org_id,
+ name=name,
+ api_url=optional_string("api_url"),
+ proxy_url=optional_string("proxy_url"),
+ realtime_url=optional_string("realtime_url"),
+ is_universal_api=bool(value.get("is_universal_api", False)),
+ git_metadata=MappingProxyType(dict(git_metadata)) if git_metadata is not None else None,
+ raw=MappingProxyType(dict(value)),
+ )
+
+
+@dataclass(frozen=True)
+class LoginResult:
+ """Selected organization and the complete login response."""
+
+ organization: OrganizationInfo
+ response: Mapping[str, Any]
+
+
+class AuthAPI:
+ """Authenticate an API key and configure an endpoint router."""
+
+ def __init__(self, transport: Transport, router: EndpointRouter):
+ self._transport = transport
+ self._router = router
+
+ def login(
+ self,
+ api_key: str,
+ *,
+ org_name: str | None = None,
+ api_url: str | None = None,
+ proxy_url: str | None = None,
+ ) -> LoginResult:
+ """Log in, select an organization, and apply routing override precedence."""
+
+ api_key = HTTPConnection.sanitize_token(api_key)
+ response = self._transport.request_json(
+ "POST",
+ self._router.resolve(RequestTarget.APP, "/api/apikey/login"),
+ headers={"Authorization": f"Bearer {api_key}"},
+ retry_mode=RetryMode.SAFE_READ,
+ )
+ if not isinstance(response, Mapping):
+ raise ValueError("API-key login returned a non-object response")
+ raw_orgs = response.get("org_info")
+ if not isinstance(raw_orgs, Sequence) or isinstance(raw_orgs, (str, bytes)):
+ raise ValueError("API-key login response did not include an organization list")
+
+ organizations = [OrganizationInfo.from_dict(org) for org in raw_orgs if isinstance(org, Mapping)]
+ organization = self._select_organization(organizations, org_name)
+
+ resolved_api_url = api_url or BraintrustEnv.API_URL.get(organization.api_url)
+ resolved_proxy_url = proxy_url or BraintrustEnv.PROXY_URL.get(organization.proxy_url)
+ if not resolved_api_url:
+ if org_name:
+ raise ValueError(
+ f"Unable to log into organization '{org_name}'."
+ " Are you sure this credential is scoped to the organization?"
+ )
+ raise ValueError("Unable to log into any organization with the provided credential.")
+
+ self._router.configure(
+ api_url=resolved_api_url,
+ proxy_url=resolved_proxy_url,
+ is_universal_api=organization.is_universal_api,
+ )
+ return LoginResult(organization=organization, response=MappingProxyType(dict(response)))
+
+ @staticmethod
+ def _select_organization(organizations: Sequence[OrganizationInfo], org_name: str | None) -> OrganizationInfo:
+ if not organizations:
+ raise ValueError("This user is not part of any organizations.")
+ for organization in organizations:
+ if org_name is None or organization.name == org_name:
+ return organization
+ choices = ", ".join(organization.name for organization in organizations)
+ raise ValueError(f"Organization {org_name} not found. Must be one of {choices}")

py/src/braintrust/api/__init__.py

@@ -1,5 +1,11 @@
-"""Braintrust API client package."""
+"""Public Braintrust API client package."""
 
+from ._routing import EndpointRouter, RequestTarget
+from ._service import ClientContext
+from .attachments import AttachmentsAPI
+from .auth import AuthAPI, LoginResult, OrganizationInfo
+from .client import BraintrustClient
+from .datasets import DatasetsAPI
 from .errors import (
  BraintrustAPIError,
  BraintrustHTTPError,
@@ -8,16 +14,35 @@
  BraintrustTransportError,
  BraintrustTransportRetryExhaustedError,
 )
+from .experiments import ExperimentsAPI
+from .functions import FunctionsAPI
 from .policies import RetryMode, RetryPolicy
+from .projects import ProjectsAPI
+from .prompts import PromptsAPI
+from .queries import QueriesAPI
 
 
 __all__ = [
+ "AttachmentsAPI",
+ "AuthAPI",
  "BraintrustAPIError",
+ "BraintrustClient",
  "BraintrustHTTPError",
  "BraintrustResponseError",
  "BraintrustRetryExhaustedError",
  "BraintrustTransportError",
  "BraintrustTransportRetryExhaustedError",
+ "ClientContext",
+ "DatasetsAPI",
+ "EndpointRouter",
+ "ExperimentsAPI",
+ "FunctionsAPI",
+ "LoginResult",
+ "OrganizationInfo",
+ "ProjectsAPI",
+ "PromptsAPI",
+ "QueriesAPI",
+ "RequestTarget",
  "RetryMode",
  "RetryPolicy",
 ]

py/src/braintrust/api/_routing.py

@@ -0,0 +1,69 @@
+"""Endpoint routing for Braintrust API requests."""
+
+import enum
+from dataclasses import dataclass
+
+from ..util import _urljoin
+
+
+_V1_PROXY_SUFFIX = "/v1/proxy"
+
+
+class RequestTarget(enum.Enum):
+ """A logical Braintrust request destination."""
+
+ APP = "app"
+ API = "api"
+ PROXY = "proxy"
+
+
+def normalize_proxy_url(proxy_url: str) -> str:
+ """Normalize a Universal Proxy URL to the API host used by SDK routes."""
+
+ if proxy_url.endswith(_V1_PROXY_SUFFIX):
+ return proxy_url[: -len(_V1_PROXY_SUFFIX)]
+ return proxy_url
+
+
+@dataclass
+class EndpointRouter:
+ """Resolve logical Braintrust targets without changing their configured origins."""
+
+ app_url: str
+ api_url: str | None = None
+ proxy_url: str | None = None
+ is_universal_api: bool = False
+
+ def configure(
+ self,
+ *,
+ api_url: str | None,
+ proxy_url: str | None,
+ is_universal_api: bool = False,
+ ) -> None:
+ """Apply URLs discovered during authentication."""
+
+ self.api_url = api_url
+ self.proxy_url = proxy_url
+ self.is_universal_api = is_universal_api
+
+ def base_url(self, target: RequestTarget) -> str:
+ """Return the configured origin for ``target``."""
+
+ if target is RequestTarget.APP:
+ return self.app_url
+ if target is RequestTarget.API:
+ if not self.api_url:
+ raise RuntimeError("API URL is unavailable before organization discovery")
+ return self.api_url
+ if target is RequestTarget.PROXY:
+ base_url = self.proxy_url or self.api_url
+ if not base_url:
+ raise RuntimeError("Proxy URL is unavailable before organization discovery")
+ return normalize_proxy_url(base_url)
+ raise ValueError(f"Unknown request target: {target!r}")
+
+ def resolve(self, target: RequestTarget, path: str) -> str:
+ """Resolve ``path`` against the origin for ``target``."""
+
+ return _urljoin(self.base_url(target), path)

py/src/braintrust/api/_service.py

@@ -0,0 +1,51 @@
+"""Shared resource service primitives."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import Transport
+
+
+@dataclass(frozen=True)
+class ClientContext:
+ """Organization and credential context shared by resource services."""
+
+ org_id: str
+ org_name: str
+
+
+class ResourceAPI:
+ """Base class for synchronous resource services."""
+
+ def __init__(
+ self,
+ transport: Transport,
+ router: EndpointRouter,
+ context: ClientContext,
+ api_key: str,
+ ):
+ self._transport = transport
+ self._router = router
+ self._context = context
+ self._api_key = api_key
+
+ def _request_json(
+ self,
+ target: RequestTarget,
+ method: str,
+ path: str,
+ *,
+ headers: Mapping[str, str] | None = None,
+ **kwargs: Any,
+ ) -> Any:
+ request_headers = {"Authorization": f"Bearer {self._api_key}"}
+ if headers:
+ request_headers.update(headers)
+ return self._transport.request_json(
+ method,
+ self._router.resolve(target, path),
+ headers=request_headers,
+ **kwargs,
+ )

py/src/braintrust/api/_transport.py

@@ -1,6 +1,7 @@
 """Legacy and policy-aware HTTP transport primitives for the Braintrust SDK."""
 
 import datetime
+import http.cookiejar
 import logging
 import sys
 import time
@@ -28,6 +29,11 @@
 logger = logging.getLogger(__name__)
 
 
+class _RejectCookiesPolicy(http.cookiejar.DefaultCookiePolicy):
+ def set_ok(self, cookie: Any, request: Any) -> bool:
+ return False
+
+
 class RetryRequestExceptionsAdapter(HTTPAdapter):
  """An HTTP adapter that automatically retries requests on connection exceptions.
 
@@ -189,13 +195,16 @@ def __init__(
  session: requests.Session | None = None,
  adapter: HTTPAdapter | None = None,
  enable_sdk_retries: bool | None = None,
+ persist_cookies: bool = True,
  sleep: Callable[[float], None] = time.sleep,
  monotonic: Callable[[], float] = time.monotonic,
  wall_clock: Callable[[], float] = time.time,
  ):
  custom_transport = session is not None or adapter is not None
  self._owns_session = session is None
  self.session = session if session is not None else requests.Session()
+ if not persist_cookies and self._owns_session:
+ self.session.cookies.set_policy(_RejectCookiesPolicy())
  self._sdk_retries_enabled = not custom_transport if enable_sdk_retries is None else enable_sdk_retries
  if adapter is not None:
  self.session.mount("http://", adapter)

py/src/braintrust/api/attachments.py

@@ -0,0 +1,10 @@
+"""Attachment metadata API service."""
+
+from ._service import ResourceAPI
+
+
+class AttachmentsAPI(ResourceAPI):
+ """Synchronous attachment metadata operations.
+
+ Signed object-storage traffic remains outside the routed transport.
+ """

py/src/braintrust/api/auth.py

@@ -0,0 +1,122 @@
+"""Authentication and organization discovery service."""
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Any
+
+from ..env import BraintrustEnv
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import HTTPConnection, Transport
+from .policies import RetryMode
+
+
+@dataclass(frozen=True)
+class OrganizationInfo:
+ """Organization routing information returned by API-key login."""
+
+ id: str
+ name: str
+ api_url: str | None
+ proxy_url: str | None
+ realtime_url: str | None
+ is_universal_api: bool
+ git_metadata: Mapping[str, Any] | None
+ raw: Mapping[str, Any]
+
+ @classmethod
+ def from_dict(cls, value: Mapping[str, Any]) -> "OrganizationInfo":
+ """Parse an additive login response while retaining unknown fields."""
+
+ org_id = value.get("id")
+ name = value.get("name")
+ if not isinstance(org_id, str) or not isinstance(name, str):
+ raise ValueError("Organization login data must include string id and name fields")
+
+ def optional_string(field: str) -> str | None:
+ result = value.get(field)
+ return result if isinstance(result, str) and result else None
+
+ git_metadata = value.get("git_metadata")
+ if not isinstance(git_metadata, Mapping):
+ git_metadata = None
+
+ return cls(
+ id=org_id,
+ name=name,
+ api_url=optional_string("api_url"),
+ proxy_url=optional_string("proxy_url"),
+ realtime_url=optional_string("realtime_url"),
+ is_universal_api=bool(value.get("is_universal_api", False)),
+ git_metadata=MappingProxyType(dict(git_metadata)) if git_metadata is not None else None,
+ raw=MappingProxyType(dict(value)),
+ )
+
+
+@dataclass(frozen=True)
+class LoginResult:
+ """Selected organization and the complete login response."""
+
+ organization: OrganizationInfo
+ response: Mapping[str, Any]
+
+
+class AuthAPI:
+ """Authenticate an API key and configure an endpoint router."""
+
+ def __init__(self, transport: Transport, router: EndpointRouter):
+ self._transport = transport
+ self._router = router
+
+ def login(
+ self,
+ api_key: str,
+ *,
+ org_name: str | None = None,
+ api_url: str | None = None,
+ proxy_url: str | None = None,
+ ) -> LoginResult:
+ """Log in, select an organization, and apply routing override precedence."""
+
+ api_key = HTTPConnection.sanitize_token(api_key)
+ response = self._transport.request_json(
+ "POST",
+ self._router.resolve(RequestTarget.APP, "/api/apikey/login"),
+ headers={"Authorization": f"Bearer {api_key}"},
+ retry_mode=RetryMode.SAFE_READ,
+ )
+ if not isinstance(response, Mapping):
+ raise ValueError("API-key login returned a non-object response")
+ raw_orgs = response.get("org_info")
+ if not isinstance(raw_orgs, Sequence) or isinstance(raw_orgs, (str, bytes)):
+ raise ValueError("API-key login response did not include an organization list")
+
+ organizations = [OrganizationInfo.from_dict(org) for org in raw_orgs if isinstance(org, Mapping)]
+ organization = self._select_organization(organizations, org_name)
+
+ resolved_api_url = api_url or BraintrustEnv.API_URL.get(organization.api_url)
+ resolved_proxy_url = proxy_url or BraintrustEnv.PROXY_URL.get(organization.proxy_url)
+ if not resolved_api_url:
+ if org_name:
+ raise ValueError(
+ f"Unable to log into organization '{org_name}'."
+ " Are you sure this credential is scoped to the organization?"
+ )
+ raise ValueError("Unable to log into any organization with the provided credential.")
+
+ self._router.configure(
+ api_url=resolved_api_url,
+ proxy_url=resolved_proxy_url,
+ is_universal_api=organization.is_universal_api,
+ )
+ return LoginResult(organization=organization, response=MappingProxyType(dict(response)))
+
+ @staticmethod
+ def _select_organization(organizations: Sequence[OrganizationInfo], org_name: str | None) -> OrganizationInfo:
+ if not organizations:
+ raise ValueError("This user is not part of any organizations.")
+ for organization in organizations:
+ if org_name is None or organization.name == org_name:
+ return organization
+ choices = ", ".join(organization.name for organization in organizations)
+ raise ValueError(f"Organization {org_name} not found. Must be one of {choices}")

py/src/braintrust/api/__init__.py

@@ -1,5 +1,11 @@
-"""Braintrust API client package."""
+"""Public Braintrust API client package."""
 
+from ._routing import EndpointRouter, RequestTarget
+from ._service import ClientContext
+from .attachments import AttachmentsAPI
+from .auth import AuthAPI, LoginResult, OrganizationInfo
+from .client import BraintrustClient
+from .datasets import DatasetsAPI
 from .errors import (
  BraintrustAPIError,
  BraintrustHTTPError,
@@ -8,16 +14,35 @@
  BraintrustTransportError,
  BraintrustTransportRetryExhaustedError,
 )
+from .experiments import ExperimentsAPI
+from .functions import FunctionsAPI
 from .policies import RetryMode, RetryPolicy
+from .projects import ProjectsAPI
+from .prompts import PromptsAPI
+from .queries import QueriesAPI
 
 
 __all__ = [
+ "AttachmentsAPI",
+ "AuthAPI",
  "BraintrustAPIError",
+ "BraintrustClient",
  "BraintrustHTTPError",
  "BraintrustResponseError",
  "BraintrustRetryExhaustedError",
  "BraintrustTransportError",
  "BraintrustTransportRetryExhaustedError",
+ "ClientContext",
+ "DatasetsAPI",
+ "EndpointRouter",
+ "ExperimentsAPI",
+ "FunctionsAPI",
+ "LoginResult",
+ "OrganizationInfo",
+ "ProjectsAPI",
+ "PromptsAPI",
+ "QueriesAPI",
+ "RequestTarget",
  "RetryMode",
  "RetryPolicy",
 ]

py/src/braintrust/api/_routing.py

@@ -0,0 +1,69 @@
+"""Endpoint routing for Braintrust API requests."""
+
+import enum
+from dataclasses import dataclass
+
+from ..util import _urljoin
+
+
+_V1_PROXY_SUFFIX = "/v1/proxy"
+
+
+class RequestTarget(enum.Enum):
+ """A logical Braintrust request destination."""
+
+ APP = "app"
+ API = "api"
+ PROXY = "proxy"
+
+
+def normalize_proxy_url(proxy_url: str) -> str:
+ """Normalize a Universal Proxy URL to the API host used by SDK routes."""
+
+ if proxy_url.endswith(_V1_PROXY_SUFFIX):
+ return proxy_url[: -len(_V1_PROXY_SUFFIX)]
+ return proxy_url
+
+
+@dataclass
+class EndpointRouter:
+ """Resolve logical Braintrust targets without changing their configured origins."""
+
+ app_url: str
+ api_url: str | None = None
+ proxy_url: str | None = None
+ is_universal_api: bool = False
+
+ def configure(
+ self,
+ *,
+ api_url: str | None,
+ proxy_url: str | None,
+ is_universal_api: bool = False,
+ ) -> None:
+ """Apply URLs discovered during authentication."""
+
+ self.api_url = api_url
+ self.proxy_url = proxy_url
+ self.is_universal_api = is_universal_api
+
+ def base_url(self, target: RequestTarget) -> str:
+ """Return the configured origin for ``target``."""
+
+ if target is RequestTarget.APP:
+ return self.app_url
+ if target is RequestTarget.API:
+ if not self.api_url:
+ raise RuntimeError("API URL is unavailable before organization discovery")
+ return self.api_url
+ if target is RequestTarget.PROXY:
+ base_url = self.proxy_url or self.api_url
+ if not base_url:
+ raise RuntimeError("Proxy URL is unavailable before organization discovery")
+ return normalize_proxy_url(base_url)
+ raise ValueError(f"Unknown request target: {target!r}")
+
+ def resolve(self, target: RequestTarget, path: str) -> str:
+ """Resolve ``path`` against the origin for ``target``."""
+
+ return _urljoin(self.base_url(target), path)

py/src/braintrust/api/_service.py

@@ -0,0 +1,51 @@
+"""Shared resource service primitives."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import Transport
+
+
+@dataclass(frozen=True)
+class ClientContext:
+ """Organization and credential context shared by resource services."""
+
+ org_id: str
+ org_name: str
+
+
+class ResourceAPI:
+ """Base class for synchronous resource services."""
+
+ def __init__(
+ self,
+ transport: Transport,
+ router: EndpointRouter,
+ context: ClientContext,
+ api_key: str,
+ ):
+ self._transport = transport
+ self._router = router
+ self._context = context
+ self._api_key = api_key
+
+ def _request_json(
+ self,
+ target: RequestTarget,
+ method: str,
+ path: str,
+ *,
+ headers: Mapping[str, str] | None = None,
+ **kwargs: Any,
+ ) -> Any:
+ request_headers = {"Authorization": f"Bearer {self._api_key}"}
+ if headers:
+ request_headers.update(headers)
+ return self._transport.request_json(
+ method,
+ self._router.resolve(target, path),
+ headers=request_headers,
+ **kwargs,
+ )

py/src/braintrust/api/_transport.py

@@ -1,6 +1,7 @@
 """Legacy and policy-aware HTTP transport primitives for the Braintrust SDK."""
 
 import datetime
+import http.cookiejar
 import logging
 import sys
 import time
@@ -28,6 +29,11 @@
 logger = logging.getLogger(__name__)
 
 
+class _RejectCookiesPolicy(http.cookiejar.DefaultCookiePolicy):
+ def set_ok(self, cookie: Any, request: Any) -> bool:
+ return False
+
+
 class RetryRequestExceptionsAdapter(HTTPAdapter):
  """An HTTP adapter that automatically retries requests on connection exceptions.
 
@@ -189,13 +195,16 @@ def __init__(
  session: requests.Session | None = None,
  adapter: HTTPAdapter | None = None,
  enable_sdk_retries: bool | None = None,
+ persist_cookies: bool = True,
  sleep: Callable[[float], None] = time.sleep,
  monotonic: Callable[[], float] = time.monotonic,
  wall_clock: Callable[[], float] = time.time,
  ):
  custom_transport = session is not None or adapter is not None
  self._owns_session = session is None
  self.session = session if session is not None else requests.Session()
+ if not persist_cookies and self._owns_session:
+ self.session.cookies.set_policy(_RejectCookiesPolicy())
  self._sdk_retries_enabled = not custom_transport if enable_sdk_retries is None else enable_sdk_retries
  if adapter is not None:
  self.session.mount("http://", adapter)

py/src/braintrust/api/attachments.py

@@ -0,0 +1,10 @@
+"""Attachment metadata API service."""
+
+from ._service import ResourceAPI
+
+
+class AttachmentsAPI(ResourceAPI):
+ """Synchronous attachment metadata operations.
+
+ Signed object-storage traffic remains outside the routed transport.
+ """

py/src/braintrust/api/auth.py

@@ -0,0 +1,122 @@
+"""Authentication and organization discovery service."""
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Any
+
+from ..env import BraintrustEnv
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import HTTPConnection, Transport
+from .policies import RetryMode
+
+
+@dataclass(frozen=True)
+class OrganizationInfo:
+ """Organization routing information returned by API-key login."""
+
+ id: str
+ name: str
+ api_url: str | None
+ proxy_url: str | None
+ realtime_url: str | None
+ is_universal_api: bool
+ git_metadata: Mapping[str, Any] | None
+ raw: Mapping[str, Any]
+
+ @classmethod
+ def from_dict(cls, value: Mapping[str, Any]) -> "OrganizationInfo":
+ """Parse an additive login response while retaining unknown fields."""
+
+ org_id = value.get("id")
+ name = value.get("name")
+ if not isinstance(org_id, str) or not isinstance(name, str):
+ raise ValueError("Organization login data must include string id and name fields")
+
+ def optional_string(field: str) -> str | None:
+ result = value.get(field)
+ return result if isinstance(result, str) and result else None
+
+ git_metadata = value.get("git_metadata")
+ if not isinstance(git_metadata, Mapping):
+ git_metadata = None
+
+ return cls(
+ id=org_id,
+ name=name,
+ api_url=optional_string("api_url"),
+ proxy_url=optional_string("proxy_url"),
+ realtime_url=optional_string("realtime_url"),
+ is_universal_api=bool(value.get("is_universal_api", False)),
+ git_metadata=MappingProxyType(dict(git_metadata)) if git_metadata is not None else None,
+ raw=MappingProxyType(dict(value)),
+ )
+
+
+@dataclass(frozen=True)
+class LoginResult:
+ """Selected organization and the complete login response."""
+
+ organization: OrganizationInfo
+ response: Mapping[str, Any]
+
+
+class AuthAPI:
+ """Authenticate an API key and configure an endpoint router."""
+
+ def __init__(self, transport: Transport, router: EndpointRouter):
+ self._transport = transport
+ self._router = router
+
+ def login(
+ self,
+ api_key: str,
+ *,
+ org_name: str | None = None,
+ api_url: str | None = None,
+ proxy_url: str | None = None,
+ ) -> LoginResult:
+ """Log in, select an organization, and apply routing override precedence."""
+
+ api_key = HTTPConnection.sanitize_token(api_key)
+ response = self._transport.request_json(
+ "POST",
+ self._router.resolve(RequestTarget.APP, "/api/apikey/login"),
+ headers={"Authorization": f"Bearer {api_key}"},
+ retry_mode=RetryMode.SAFE_READ,
+ )
+ if not isinstance(response, Mapping):
+ raise ValueError("API-key login returned a non-object response")
+ raw_orgs = response.get("org_info")
+ if not isinstance(raw_orgs, Sequence) or isinstance(raw_orgs, (str, bytes)):
+ raise ValueError("API-key login response did not include an organization list")
+
+ organizations = [OrganizationInfo.from_dict(org) for org in raw_orgs if isinstance(org, Mapping)]
+ organization = self._select_organization(organizations, org_name)
+
+ resolved_api_url = api_url or BraintrustEnv.API_URL.get(organization.api_url)
+ resolved_proxy_url = proxy_url or BraintrustEnv.PROXY_URL.get(organization.proxy_url)
+ if not resolved_api_url:
+ if org_name:
+ raise ValueError(
+ f"Unable to log into organization '{org_name}'."
+ " Are you sure this credential is scoped to the organization?"
+ )
+ raise ValueError("Unable to log into any organization with the provided credential.")
+
+ self._router.configure(
+ api_url=resolved_api_url,
+ proxy_url=resolved_proxy_url,
+ is_universal_api=organization.is_universal_api,
+ )
+ return LoginResult(organization=organization, response=MappingProxyType(dict(response)))
+
+ @staticmethod
+ def _select_organization(organizations: Sequence[OrganizationInfo], org_name: str | None) -> OrganizationInfo:
+ if not organizations:
+ raise ValueError("This user is not part of any organizations.")
+ for organization in organizations:
+ if org_name is None or organization.name == org_name:
+ return organization
+ choices = ", ".join(organization.name for organization in organizations)
+ raise ValueError(f"Organization {org_name} not found. Must be one of {choices}")

py/src/braintrust/api/__init__.py

@@ -1,5 +1,11 @@
-"""Braintrust API client package."""
+"""Public Braintrust API client package."""
 
+from ._routing import EndpointRouter, RequestTarget
+from ._service import ClientContext
+from .attachments import AttachmentsAPI
+from .auth import AuthAPI, LoginResult, OrganizationInfo
+from .client import BraintrustClient
+from .datasets import DatasetsAPI
 from .errors import (
  BraintrustAPIError,
  BraintrustHTTPError,
@@ -8,16 +14,35 @@
  BraintrustTransportError,
  BraintrustTransportRetryExhaustedError,
 )
+from .experiments import ExperimentsAPI
+from .functions import FunctionsAPI
 from .policies import RetryMode, RetryPolicy
+from .projects import ProjectsAPI
+from .prompts import PromptsAPI
+from .queries import QueriesAPI
 
 
 __all__ = [
+ "AttachmentsAPI",
+ "AuthAPI",
  "BraintrustAPIError",
+ "BraintrustClient",
  "BraintrustHTTPError",
  "BraintrustResponseError",
  "BraintrustRetryExhaustedError",
  "BraintrustTransportError",
  "BraintrustTransportRetryExhaustedError",
+ "ClientContext",
+ "DatasetsAPI",
+ "EndpointRouter",
+ "ExperimentsAPI",
+ "FunctionsAPI",
+ "LoginResult",
+ "OrganizationInfo",
+ "ProjectsAPI",
+ "PromptsAPI",
+ "QueriesAPI",
+ "RequestTarget",
  "RetryMode",
  "RetryPolicy",
 ]

py/src/braintrust/api/_routing.py

@@ -0,0 +1,69 @@
+"""Endpoint routing for Braintrust API requests."""
+
+import enum
+from dataclasses import dataclass
+
+from ..util import _urljoin
+
+
+_V1_PROXY_SUFFIX = "/v1/proxy"
+
+
+class RequestTarget(enum.Enum):
+ """A logical Braintrust request destination."""
+
+ APP = "app"
+ API = "api"
+ PROXY = "proxy"
+
+
+def normalize_proxy_url(proxy_url: str) -> str:
+ """Normalize a Universal Proxy URL to the API host used by SDK routes."""
+
+ if proxy_url.endswith(_V1_PROXY_SUFFIX):
+ return proxy_url[: -len(_V1_PROXY_SUFFIX)]
+ return proxy_url
+
+
+@dataclass
+class EndpointRouter:
+ """Resolve logical Braintrust targets without changing their configured origins."""
+
+ app_url: str
+ api_url: str | None = None
+ proxy_url: str | None = None
+ is_universal_api: bool = False
+
+ def configure(
+ self,
+ *,
+ api_url: str | None,
+ proxy_url: str | None,
+ is_universal_api: bool = False,
+ ) -> None:
+ """Apply URLs discovered during authentication."""
+
+ self.api_url = api_url
+ self.proxy_url = proxy_url
+ self.is_universal_api = is_universal_api
+
+ def base_url(self, target: RequestTarget) -> str:
+ """Return the configured origin for ``target``."""
+
+ if target is RequestTarget.APP:
+ return self.app_url
+ if target is RequestTarget.API:
+ if not self.api_url:
+ raise RuntimeError("API URL is unavailable before organization discovery")
+ return self.api_url
+ if target is RequestTarget.PROXY:
+ base_url = self.proxy_url or self.api_url
+ if not base_url:
+ raise RuntimeError("Proxy URL is unavailable before organization discovery")
+ return normalize_proxy_url(base_url)
+ raise ValueError(f"Unknown request target: {target!r}")
+
+ def resolve(self, target: RequestTarget, path: str) -> str:
+ """Resolve ``path`` against the origin for ``target``."""
+
+ return _urljoin(self.base_url(target), path)

py/src/braintrust/api/_service.py

@@ -0,0 +1,51 @@
+"""Shared resource service primitives."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import Transport
+
+
+@dataclass(frozen=True)
+class ClientContext:
+ """Organization and credential context shared by resource services."""
+
+ org_id: str
+ org_name: str
+
+
+class ResourceAPI:
+ """Base class for synchronous resource services."""
+
+ def __init__(
+ self,
+ transport: Transport,
+ router: EndpointRouter,
+ context: ClientContext,
+ api_key: str,
+ ):
+ self._transport = transport
+ self._router = router
+ self._context = context
+ self._api_key = api_key
+
+ def _request_json(
+ self,
+ target: RequestTarget,
+ method: str,
+ path: str,
+ *,
+ headers: Mapping[str, str] | None = None,
+ **kwargs: Any,
+ ) -> Any:
+ request_headers = {"Authorization": f"Bearer {self._api_key}"}
+ if headers:
+ request_headers.update(headers)
+ return self._transport.request_json(
+ method,
+ self._router.resolve(target, path),
+ headers=request_headers,
+ **kwargs,
+ )

py/src/braintrust/api/_transport.py

@@ -1,6 +1,7 @@
 """Legacy and policy-aware HTTP transport primitives for the Braintrust SDK."""
 
 import datetime
+import http.cookiejar
 import logging
 import sys
 import time
@@ -28,6 +29,11 @@
 logger = logging.getLogger(__name__)
 
 
+class _RejectCookiesPolicy(http.cookiejar.DefaultCookiePolicy):
+ def set_ok(self, cookie: Any, request: Any) -> bool:
+ return False
+
+
 class RetryRequestExceptionsAdapter(HTTPAdapter):
  """An HTTP adapter that automatically retries requests on connection exceptions.
 
@@ -189,13 +195,16 @@ def __init__(
  session: requests.Session | None = None,
  adapter: HTTPAdapter | None = None,
  enable_sdk_retries: bool | None = None,
+ persist_cookies: bool = True,
  sleep: Callable[[float], None] = time.sleep,
  monotonic: Callable[[], float] = time.monotonic,
  wall_clock: Callable[[], float] = time.time,
  ):
  custom_transport = session is not None or adapter is not None
  self._owns_session = session is None
  self.session = session if session is not None else requests.Session()
+ if not persist_cookies and self._owns_session:
+ self.session.cookies.set_policy(_RejectCookiesPolicy())
  self._sdk_retries_enabled = not custom_transport if enable_sdk_retries is None else enable_sdk_retries
  if adapter is not None:
  self.session.mount("http://", adapter)

py/src/braintrust/api/attachments.py

@@ -0,0 +1,10 @@
+"""Attachment metadata API service."""
+
+from ._service import ResourceAPI
+
+
+class AttachmentsAPI(ResourceAPI):
+ """Synchronous attachment metadata operations.
+
+ Signed object-storage traffic remains outside the routed transport.
+ """

py/src/braintrust/api/auth.py

@@ -0,0 +1,122 @@
+"""Authentication and organization discovery service."""
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Any
+
+from ..env import BraintrustEnv
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import HTTPConnection, Transport
+from .policies import RetryMode
+
+
+@dataclass(frozen=True)
+class OrganizationInfo:
+ """Organization routing information returned by API-key login."""
+
+ id: str
+ name: str
+ api_url: str | None
+ proxy_url: str | None
+ realtime_url: str | None
+ is_universal_api: bool
+ git_metadata: Mapping[str, Any] | None
+ raw: Mapping[str, Any]
+
+ @classmethod
+ def from_dict(cls, value: Mapping[str, Any]) -> "OrganizationInfo":
+ """Parse an additive login response while retaining unknown fields."""
+
+ org_id = value.get("id")
+ name = value.get("name")
+ if not isinstance(org_id, str) or not isinstance(name, str):
+ raise ValueError("Organization login data must include string id and name fields")
+
+ def optional_string(field: str) -> str | None:
+ result = value.get(field)
+ return result if isinstance(result, str) and result else None
+
+ git_metadata = value.get("git_metadata")
+ if not isinstance(git_metadata, Mapping):
+ git_metadata = None
+
+ return cls(
+ id=org_id,
+ name=name,
+ api_url=optional_string("api_url"),
+ proxy_url=optional_string("proxy_url"),
+ realtime_url=optional_string("realtime_url"),
+ is_universal_api=bool(value.get("is_universal_api", False)),
+ git_metadata=MappingProxyType(dict(git_metadata)) if git_metadata is not None else None,
+ raw=MappingProxyType(dict(value)),
+ )
+
+
+@dataclass(frozen=True)
+class LoginResult:
+ """Selected organization and the complete login response."""
+
+ organization: OrganizationInfo
+ response: Mapping[str, Any]
+
+
+class AuthAPI:
+ """Authenticate an API key and configure an endpoint router."""
+
+ def __init__(self, transport: Transport, router: EndpointRouter):
+ self._transport = transport
+ self._router = router
+
+ def login(
+ self,
+ api_key: str,
+ *,
+ org_name: str | None = None,
+ api_url: str | None = None,
+ proxy_url: str | None = None,
+ ) -> LoginResult:
+ """Log in, select an organization, and apply routing override precedence."""
+
+ api_key = HTTPConnection.sanitize_token(api_key)
+ response = self._transport.request_json(
+ "POST",
+ self._router.resolve(RequestTarget.APP, "/api/apikey/login"),
+ headers={"Authorization": f"Bearer {api_key}"},
+ retry_mode=RetryMode.SAFE_READ,
+ )
+ if not isinstance(response, Mapping):
+ raise ValueError("API-key login returned a non-object response")
+ raw_orgs = response.get("org_info")
+ if not isinstance(raw_orgs, Sequence) or isinstance(raw_orgs, (str, bytes)):
+ raise ValueError("API-key login response did not include an organization list")
+
+ organizations = [OrganizationInfo.from_dict(org) for org in raw_orgs if isinstance(org, Mapping)]
+ organization = self._select_organization(organizations, org_name)
+
+ resolved_api_url = api_url or BraintrustEnv.API_URL.get(organization.api_url)
+ resolved_proxy_url = proxy_url or BraintrustEnv.PROXY_URL.get(organization.proxy_url)
+ if not resolved_api_url:
+ if org_name:
+ raise ValueError(
+ f"Unable to log into organization '{org_name}'."
+ " Are you sure this credential is scoped to the organization?"
+ )
+ raise ValueError("Unable to log into any organization with the provided credential.")
+
+ self._router.configure(
+ api_url=resolved_api_url,
+ proxy_url=resolved_proxy_url,
+ is_universal_api=organization.is_universal_api,
+ )
+ return LoginResult(organization=organization, response=MappingProxyType(dict(response)))
+
+ @staticmethod
+ def _select_organization(organizations: Sequence[OrganizationInfo], org_name: str | None) -> OrganizationInfo:
+ if not organizations:
+ raise ValueError("This user is not part of any organizations.")
+ for organization in organizations:
+ if org_name is None or organization.name == org_name:
+ return organization
+ choices = ", ".join(organization.name for organization in organizations)
+ raise ValueError(f"Organization {org_name} not found. Must be one of {choices}")

py/src/braintrust/api/__init__.py

@@ -1,5 +1,11 @@
-"""Braintrust API client package."""
+"""Public Braintrust API client package."""
 
+from ._routing import EndpointRouter, RequestTarget
+from ._service import ClientContext
+from .attachments import AttachmentsAPI
+from .auth import AuthAPI, LoginResult, OrganizationInfo
+from .client import BraintrustClient
+from .datasets import DatasetsAPI
 from .errors import (
  BraintrustAPIError,
  BraintrustHTTPError,
@@ -8,16 +14,35 @@
  BraintrustTransportError,
  BraintrustTransportRetryExhaustedError,
 )
+from .experiments import ExperimentsAPI
+from .functions import FunctionsAPI
 from .policies import RetryMode, RetryPolicy
+from .projects import ProjectsAPI
+from .prompts import PromptsAPI
+from .queries import QueriesAPI
 
 
 __all__ = [
+ "AttachmentsAPI",
+ "AuthAPI",
  "BraintrustAPIError",
+ "BraintrustClient",
  "BraintrustHTTPError",
  "BraintrustResponseError",
  "BraintrustRetryExhaustedError",
  "BraintrustTransportError",
  "BraintrustTransportRetryExhaustedError",
+ "ClientContext",
+ "DatasetsAPI",
+ "EndpointRouter",
+ "ExperimentsAPI",
+ "FunctionsAPI",
+ "LoginResult",
+ "OrganizationInfo",
+ "ProjectsAPI",
+ "PromptsAPI",
+ "QueriesAPI",
+ "RequestTarget",
  "RetryMode",
  "RetryPolicy",
 ]

py/src/braintrust/api/_routing.py

@@ -0,0 +1,69 @@
+"""Endpoint routing for Braintrust API requests."""
+
+import enum
+from dataclasses import dataclass
+
+from ..util import _urljoin
+
+
+_V1_PROXY_SUFFIX = "/v1/proxy"
+
+
+class RequestTarget(enum.Enum):
+ """A logical Braintrust request destination."""
+
+ APP = "app"
+ API = "api"
+ PROXY = "proxy"
+
+
+def normalize_proxy_url(proxy_url: str) -> str:
+ """Normalize a Universal Proxy URL to the API host used by SDK routes."""
+
+ if proxy_url.endswith(_V1_PROXY_SUFFIX):
+ return proxy_url[: -len(_V1_PROXY_SUFFIX)]
+ return proxy_url
+
+
+@dataclass
+class EndpointRouter:
+ """Resolve logical Braintrust targets without changing their configured origins."""
+
+ app_url: str
+ api_url: str | None = None
+ proxy_url: str | None = None
+ is_universal_api: bool = False
+
+ def configure(
+ self,
+ *,
+ api_url: str | None,
+ proxy_url: str | None,
+ is_universal_api: bool = False,
+ ) -> None:
+ """Apply URLs discovered during authentication."""
+
+ self.api_url = api_url
+ self.proxy_url = proxy_url
+ self.is_universal_api = is_universal_api
+
+ def base_url(self, target: RequestTarget) -> str:
+ """Return the configured origin for ``target``."""
+
+ if target is RequestTarget.APP:
+ return self.app_url
+ if target is RequestTarget.API:
+ if not self.api_url:
+ raise RuntimeError("API URL is unavailable before organization discovery")
+ return self.api_url
+ if target is RequestTarget.PROXY:
+ base_url = self.proxy_url or self.api_url
+ if not base_url:
+ raise RuntimeError("Proxy URL is unavailable before organization discovery")
+ return normalize_proxy_url(base_url)
+ raise ValueError(f"Unknown request target: {target!r}")
+
+ def resolve(self, target: RequestTarget, path: str) -> str:
+ """Resolve ``path`` against the origin for ``target``."""
+
+ return _urljoin(self.base_url(target), path)

py/src/braintrust/api/_service.py

@@ -0,0 +1,51 @@
+"""Shared resource service primitives."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import Transport
+
+
+@dataclass(frozen=True)
+class ClientContext:
+ """Organization and credential context shared by resource services."""
+
+ org_id: str
+ org_name: str
+
+
+class ResourceAPI:
+ """Base class for synchronous resource services."""
+
+ def __init__(
+ self,
+ transport: Transport,
+ router: EndpointRouter,
+ context: ClientContext,
+ api_key: str,
+ ):
+ self._transport = transport
+ self._router = router
+ self._context = context
+ self._api_key = api_key
+
+ def _request_json(
+ self,
+ target: RequestTarget,
+ method: str,
+ path: str,
+ *,
+ headers: Mapping[str, str] | None = None,
+ **kwargs: Any,
+ ) -> Any:
+ request_headers = {"Authorization": f"Bearer {self._api_key}"}
+ if headers:
+ request_headers.update(headers)
+ return self._transport.request_json(
+ method,
+ self._router.resolve(target, path),
+ headers=request_headers,
+ **kwargs,
+ )

py/src/braintrust/api/_transport.py

@@ -1,6 +1,7 @@
 """Legacy and policy-aware HTTP transport primitives for the Braintrust SDK."""
 
 import datetime
+import http.cookiejar
 import logging
 import sys
 import time
@@ -28,6 +29,11 @@
 logger = logging.getLogger(__name__)
 
 
+class _RejectCookiesPolicy(http.cookiejar.DefaultCookiePolicy):
+ def set_ok(self, cookie: Any, request: Any) -> bool:
+ return False
+
+
 class RetryRequestExceptionsAdapter(HTTPAdapter):
  """An HTTP adapter that automatically retries requests on connection exceptions.
 
@@ -189,13 +195,16 @@ def __init__(
  session: requests.Session | None = None,
  adapter: HTTPAdapter | None = None,
  enable_sdk_retries: bool | None = None,
+ persist_cookies: bool = True,
  sleep: Callable[[float], None] = time.sleep,
  monotonic: Callable[[], float] = time.monotonic,
  wall_clock: Callable[[], float] = time.time,
  ):
  custom_transport = session is not None or adapter is not None
  self._owns_session = session is None
  self.session = session if session is not None else requests.Session()
+ if not persist_cookies and self._owns_session:
+ self.session.cookies.set_policy(_RejectCookiesPolicy())
  self._sdk_retries_enabled = not custom_transport if enable_sdk_retries is None else enable_sdk_retries
  if adapter is not None:
  self.session.mount("http://", adapter)

py/src/braintrust/api/attachments.py

@@ -0,0 +1,10 @@
+"""Attachment metadata API service."""
+
+from ._service import ResourceAPI
+
+
+class AttachmentsAPI(ResourceAPI):
+ """Synchronous attachment metadata operations.
+
+ Signed object-storage traffic remains outside the routed transport.
+ """

py/src/braintrust/api/auth.py

@@ -0,0 +1,122 @@
+"""Authentication and organization discovery service."""
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Any
+
+from ..env import BraintrustEnv
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import HTTPConnection, Transport
+from .policies import RetryMode
+
+
+@dataclass(frozen=True)
+class OrganizationInfo:
+ """Organization routing information returned by API-key login."""
+
+ id: str
+ name: str
+ api_url: str | None
+ proxy_url: str | None
+ realtime_url: str | None
+ is_universal_api: bool
+ git_metadata: Mapping[str, Any] | None
+ raw: Mapping[str, Any]
+
+ @classmethod
+ def from_dict(cls, value: Mapping[str, Any]) -> "OrganizationInfo":
+ """Parse an additive login response while retaining unknown fields."""
+
+ org_id = value.get("id")
+ name = value.get("name")
+ if not isinstance(org_id, str) or not isinstance(name, str):
+ raise ValueError("Organization login data must include string id and name fields")
+
+ def optional_string(field: str) -> str | None:
+ result = value.get(field)
+ return result if isinstance(result, str) and result else None
+
+ git_metadata = value.get("git_metadata")
+ if not isinstance(git_metadata, Mapping):
+ git_metadata = None
+
+ return cls(
+ id=org_id,
+ name=name,
+ api_url=optional_string("api_url"),
+ proxy_url=optional_string("proxy_url"),
+ realtime_url=optional_string("realtime_url"),
+ is_universal_api=bool(value.get("is_universal_api", False)),
+ git_metadata=MappingProxyType(dict(git_metadata)) if git_metadata is not None else None,
+ raw=MappingProxyType(dict(value)),
+ )
+
+
+@dataclass(frozen=True)
+class LoginResult:
+ """Selected organization and the complete login response."""
+
+ organization: OrganizationInfo
+ response: Mapping[str, Any]
+
+
+class AuthAPI:
+ """Authenticate an API key and configure an endpoint router."""
+
+ def __init__(self, transport: Transport, router: EndpointRouter):
+ self._transport = transport
+ self._router = router
+
+ def login(
+ self,
+ api_key: str,
+ *,
+ org_name: str | None = None,
+ api_url: str | None = None,
+ proxy_url: str | None = None,
+ ) -> LoginResult:
+ """Log in, select an organization, and apply routing override precedence."""
+
+ api_key = HTTPConnection.sanitize_token(api_key)
+ response = self._transport.request_json(
+ "POST",
+ self._router.resolve(RequestTarget.APP, "/api/apikey/login"),
+ headers={"Authorization": f"Bearer {api_key}"},
+ retry_mode=RetryMode.SAFE_READ,
+ )
+ if not isinstance(response, Mapping):
+ raise ValueError("API-key login returned a non-object response")
+ raw_orgs = response.get("org_info")
+ if not isinstance(raw_orgs, Sequence) or isinstance(raw_orgs, (str, bytes)):
+ raise ValueError("API-key login response did not include an organization list")
+
+ organizations = [OrganizationInfo.from_dict(org) for org in raw_orgs if isinstance(org, Mapping)]
+ organization = self._select_organization(organizations, org_name)
+
+ resolved_api_url = api_url or BraintrustEnv.API_URL.get(organization.api_url)
+ resolved_proxy_url = proxy_url or BraintrustEnv.PROXY_URL.get(organization.proxy_url)
+ if not resolved_api_url:
+ if org_name:
+ raise ValueError(
+ f"Unable to log into organization '{org_name}'."
+ " Are you sure this credential is scoped to the organization?"
+ )
+ raise ValueError("Unable to log into any organization with the provided credential.")
+
+ self._router.configure(
+ api_url=resolved_api_url,
+ proxy_url=resolved_proxy_url,
+ is_universal_api=organization.is_universal_api,
+ )
+ return LoginResult(organization=organization, response=MappingProxyType(dict(response)))
+
+ @staticmethod
+ def _select_organization(organizations: Sequence[OrganizationInfo], org_name: str | None) -> OrganizationInfo:
+ if not organizations:
+ raise ValueError("This user is not part of any organizations.")
+ for organization in organizations:
+ if org_name is None or organization.name == org_name:
+ return organization
+ choices = ", ".join(organization.name for organization in organizations)
+ raise ValueError(f"Organization {org_name} not found. Must be one of {choices}")

py/src/braintrust/api/__init__.py

@@ -1,5 +1,11 @@
-"""Braintrust API client package."""
+"""Public Braintrust API client package."""
 
+from ._routing import EndpointRouter, RequestTarget
+from ._service import ClientContext
+from .attachments import AttachmentsAPI
+from .auth import AuthAPI, LoginResult, OrganizationInfo
+from .client import BraintrustClient
+from .datasets import DatasetsAPI
 from .errors import (
  BraintrustAPIError,
  BraintrustHTTPError,
@@ -8,16 +14,35 @@
  BraintrustTransportError,
  BraintrustTransportRetryExhaustedError,
 )
+from .experiments import ExperimentsAPI
+from .functions import FunctionsAPI
 from .policies import RetryMode, RetryPolicy
+from .projects import ProjectsAPI
+from .prompts import PromptsAPI
+from .queries import QueriesAPI
 
 
 __all__ = [
+ "AttachmentsAPI",
+ "AuthAPI",
  "BraintrustAPIError",
+ "BraintrustClient",
  "BraintrustHTTPError",
  "BraintrustResponseError",
  "BraintrustRetryExhaustedError",
  "BraintrustTransportError",
  "BraintrustTransportRetryExhaustedError",
+ "ClientContext",
+ "DatasetsAPI",
+ "EndpointRouter",
+ "ExperimentsAPI",
+ "FunctionsAPI",
+ "LoginResult",
+ "OrganizationInfo",
+ "ProjectsAPI",
+ "PromptsAPI",
+ "QueriesAPI",
+ "RequestTarget",
  "RetryMode",
  "RetryPolicy",
 ]

py/src/braintrust/api/_routing.py

@@ -0,0 +1,69 @@
+"""Endpoint routing for Braintrust API requests."""
+
+import enum
+from dataclasses import dataclass
+
+from ..util import _urljoin
+
+
+_V1_PROXY_SUFFIX = "/v1/proxy"
+
+
+class RequestTarget(enum.Enum):
+ """A logical Braintrust request destination."""
+
+ APP = "app"
+ API = "api"
+ PROXY = "proxy"
+
+
+def normalize_proxy_url(proxy_url: str) -> str:
+ """Normalize a Universal Proxy URL to the API host used by SDK routes."""
+
+ if proxy_url.endswith(_V1_PROXY_SUFFIX):
+ return proxy_url[: -len(_V1_PROXY_SUFFIX)]
+ return proxy_url
+
+
+@dataclass
+class EndpointRouter:
+ """Resolve logical Braintrust targets without changing their configured origins."""
+
+ app_url: str
+ api_url: str | None = None
+ proxy_url: str | None = None
+ is_universal_api: bool = False
+
+ def configure(
+ self,
+ *,
+ api_url: str | None,
+ proxy_url: str | None,
+ is_universal_api: bool = False,
+ ) -> None:
+ """Apply URLs discovered during authentication."""
+
+ self.api_url = api_url
+ self.proxy_url = proxy_url
+ self.is_universal_api = is_universal_api
+
+ def base_url(self, target: RequestTarget) -> str:
+ """Return the configured origin for ``target``."""
+
+ if target is RequestTarget.APP:
+ return self.app_url
+ if target is RequestTarget.API:
+ if not self.api_url:
+ raise RuntimeError("API URL is unavailable before organization discovery")
+ return self.api_url
+ if target is RequestTarget.PROXY:
+ base_url = self.proxy_url or self.api_url
+ if not base_url:
+ raise RuntimeError("Proxy URL is unavailable before organization discovery")
+ return normalize_proxy_url(base_url)
+ raise ValueError(f"Unknown request target: {target!r}")
+
+ def resolve(self, target: RequestTarget, path: str) -> str:
+ """Resolve ``path`` against the origin for ``target``."""
+
+ return _urljoin(self.base_url(target), path)

py/src/braintrust/api/_service.py

@@ -0,0 +1,51 @@
+"""Shared resource service primitives."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import Transport
+
+
+@dataclass(frozen=True)
+class ClientContext:
+ """Organization and credential context shared by resource services."""
+
+ org_id: str
+ org_name: str
+
+
+class ResourceAPI:
+ """Base class for synchronous resource services."""
+
+ def __init__(
+ self,
+ transport: Transport,
+ router: EndpointRouter,
+ context: ClientContext,
+ api_key: str,
+ ):
+ self._transport = transport
+ self._router = router
+ self._context = context
+ self._api_key = api_key
+
+ def _request_json(
+ self,
+ target: RequestTarget,
+ method: str,
+ path: str,
+ *,
+ headers: Mapping[str, str] | None = None,
+ **kwargs: Any,
+ ) -> Any:
+ request_headers = {"Authorization": f"Bearer {self._api_key}"}
+ if headers:
+ request_headers.update(headers)
+ return self._transport.request_json(
+ method,
+ self._router.resolve(target, path),
+ headers=request_headers,
+ **kwargs,
+ )

py/src/braintrust/api/_transport.py

@@ -1,6 +1,7 @@
 """Legacy and policy-aware HTTP transport primitives for the Braintrust SDK."""
 
 import datetime
+import http.cookiejar
 import logging
 import sys
 import time
@@ -28,6 +29,11 @@
 logger = logging.getLogger(__name__)
 
 
+class _RejectCookiesPolicy(http.cookiejar.DefaultCookiePolicy):
+ def set_ok(self, cookie: Any, request: Any) -> bool:
+ return False
+
+
 class RetryRequestExceptionsAdapter(HTTPAdapter):
  """An HTTP adapter that automatically retries requests on connection exceptions.
 
@@ -189,13 +195,16 @@ def __init__(
  session: requests.Session | None = None,
  adapter: HTTPAdapter | None = None,
  enable_sdk_retries: bool | None = None,
+ persist_cookies: bool = True,
  sleep: Callable[[float], None] = time.sleep,
  monotonic: Callable[[], float] = time.monotonic,
  wall_clock: Callable[[], float] = time.time,
  ):
  custom_transport = session is not None or adapter is not None
  self._owns_session = session is None
  self.session = session if session is not None else requests.Session()
+ if not persist_cookies and self._owns_session:
+ self.session.cookies.set_policy(_RejectCookiesPolicy())
  self._sdk_retries_enabled = not custom_transport if enable_sdk_retries is None else enable_sdk_retries
  if adapter is not None:
  self.session.mount("http://", adapter)

py/src/braintrust/api/attachments.py

@@ -0,0 +1,10 @@
+"""Attachment metadata API service."""
+
+from ._service import ResourceAPI
+
+
+class AttachmentsAPI(ResourceAPI):
+ """Synchronous attachment metadata operations.
+
+ Signed object-storage traffic remains outside the routed transport.
+ """

py/src/braintrust/api/auth.py

@@ -0,0 +1,122 @@
+"""Authentication and organization discovery service."""
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Any
+
+from ..env import BraintrustEnv
+from ._routing import EndpointRouter, RequestTarget
+from ._transport import HTTPConnection, Transport
+from .policies import RetryMode
+
+
+@dataclass(frozen=True)
+class OrganizationInfo:
+ """Organization routing information returned by API-key login."""
+
+ id: str
+ name: str
+ api_url: str | None
+ proxy_url: str | None
+ realtime_url: str | None
+ is_universal_api: bool
+ git_metadata: Mapping[str, Any] | None
+ raw: Mapping[str, Any]
+
+ @classmethod
+ def from_dict(cls, value: Mapping[str, Any]) -> "OrganizationInfo":
+ """Parse an additive login response while retaining unknown fields."""
+
+ org_id = value.get("id")
+ name = value.get("name")
+ if not isinstance(org_id, str) or not isinstance(name, str):
+ raise ValueError("Organization login data must include string id and name fields")
+
+ def optional_string(field: str) -> str | None:
+ result = value.get(field)
+ return result if isinstance(result, str) and result else None
+
+ git_metadata = value.get("git_metadata")
+ if not isinstance(git_metadata, Mapping):
+ git_metadata = None
+
+ return cls(
+ id=org_id,
+ name=name,
+ api_url=optional_string("api_url"),
+ proxy_url=optional_string("proxy_url"),
+ realtime_url=optional_string("realtime_url"),
+ is_universal_api=bool(value.get("is_universal_api", False)),
+ git_metadata=MappingProxyType(dict(git_metadata)) if git_metadata is not None else None,
+ raw=MappingProxyType(dict(value)),
+ )
+
+
+@dataclass(frozen=True)
+class LoginResult:
+ """Selected organization and the complete login response."""
+
+ organization: OrganizationInfo
+ response: Mapping[str, Any]
+
+
+class AuthAPI:
+ """Authenticate an API key and configure an endpoint router."""
+
+ def __init__(self, transport: Transport, router: EndpointRouter):
+ self._transport = transport
+ self._router = router
+
+ def login(
+ self,
+ api_key: str,
+ *,
+ org_name: str | None = None,
+ api_url: str | None = None,
+ proxy_url: str | None = None,
+ ) -> LoginResult:
+ """Log in, select an organization, and apply routing override precedence."""
+
+ api_key = HTTPConnection.sanitize_token(api_key)
+ response = self._transport.request_json(
+ "POST",
+ self._router.resolve(RequestTarget.APP, "/api/apikey/login"),
+ headers={"Authorization": f"Bearer {api_key}"},
+ retry_mode=RetryMode.SAFE_READ,
+ )
+ if not isinstance(response, Mapping):
+ raise ValueError("API-key login returned a non-object response")
+ raw_orgs = response.get("org_info")
+ if not isinstance(raw_orgs, Sequence) or isinstance(raw_orgs, (str, bytes)):
+ raise ValueError("API-key login response did not include an organization list")
+
+ organizations = [OrganizationInfo.from_dict(org) for org in raw_orgs if isinstance(org, Mapping)]
+ organization = self._select_organization(organizations, org_name)
+
+ resolved_api_url = api_url or BraintrustEnv.API_URL.get(organization.api_url)
+ resolved_proxy_url = proxy_url or BraintrustEnv.PROXY_URL.get(organization.proxy_url)
+ if not resolved_api_url:
+ if org_name:
+ raise ValueError(
+ f"Unable to log into organization '{org_name}'."
+ " Are you sure this credential is scoped to the organization?"
+ )
+ raise ValueError("Unable to log into any organization with the provided credential.")
+
+ self._router.configure(
+ api_url=resolved_api_url,
+ proxy_url=resolved_proxy_url,
+ is_universal_api=organization.is_universal_api,
+ )
+ return LoginResult(organization=organization, response=MappingProxyType(dict(response)))
+
+ @staticmethod
+ def _select_organization(organizations: Sequence[OrganizationInfo], org_name: str | None) -> OrganizationInfo:
+ if not organizations:
+ raise ValueError("This user is not part of any organizations.")
+ for organization in organizations:
+ if org_name is None or organization.name == org_name:
+ return organization
+ choices = ", ".join(organization.name for organization in organizations)
+ raise ValueError(f"Organization {org_name} not found. Must be one of {choices}")