mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-21 17:25:26 -04:00
feat(code): add LangSmith base URL to /auth (#4228)
Configure the LangSmith tracing region/endpoint (US, EU, or a custom URL) directly from `/auth` and `dcode auth set langsmith --base-url`. --- Following #4193 (LangSmith tracing in `/auth`), this lets users configure the LangSmith **endpoint** through `/auth` instead of exporting `LANGSMITH_ENDPOINT` by hand — motivated by EU-region users at `https://eu.api.smith.langchain.com`. The stored LangSmith credential's `base_url` is now applied to `LANGSMITH_ENDPOINT` (clearing the `LANGCHAIN_ENDPOINT` alternate) with the same precedence as the stored project: an explicit env value stays authoritative at startup, and the immediate `/auth` save replaces it. - The `/auth` prompt gains a region selector (`RadioSet`): **United States (default)** / **Europe** / **Custom…**, where Custom reveals a free-text URL field for self-hosted/proxy endpoints. - `dcode auth set` gains `--base-url`, accepting the `us`/`eu` shorthands or a full URL (a non-interactive flag can't render a radio). - Both surfaces validate that a custom endpoint is an `http(s)` URL, so a key is never paired with a non-HTTP, attacker-shaped endpoint. Region aliases normalize to canonical URLs on input, so `auth.json` always stores a real URL (US stores blank → SaaS default). At startup a stored key without an endpoint never clears an existing env `LANGSMITH_ENDPOINT`, so self-hosted setups keep working. Made by [Open SWE](https://openswe.vercel.app/agents/4f2446ed-36cc-d8e7-a098-3df2bbcf650d) --------- Co-authored-by: Mason Daugherty <61371264+mdrxy@users.noreply.github.com> Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com> Co-authored-by: Mason Daugherty <github@mdrxy.com> Co-authored-by: Mason Daugherty <mason@langchain.dev>
This commit is contained in:
@@ -108,6 +108,16 @@ def setup_auth_parser(
|
||||
default=None,
|
||||
help="With `set langsmith`, set a custom LangSmith project name",
|
||||
)
|
||||
set_parser.add_argument(
|
||||
"--base-url",
|
||||
dest="base_url",
|
||||
metavar="URL",
|
||||
default=None,
|
||||
help=(
|
||||
"Pair an endpoint with the key. For `set langsmith`, accepts the "
|
||||
"`us`/`eu` region shorthands or a full URL"
|
||||
),
|
||||
)
|
||||
set_parser.add_argument(
|
||||
"-h",
|
||||
"--help",
|
||||
@@ -171,6 +181,7 @@ def run_auth_command(args: argparse.Namespace) -> int:
|
||||
args.provider,
|
||||
from_env=args.from_env,
|
||||
project=getattr(args, "project", None),
|
||||
base_url=getattr(args, "base_url", None),
|
||||
)
|
||||
if command in {"remove", "rm", "delete"}:
|
||||
return _run_remove(args.provider)
|
||||
@@ -358,12 +369,19 @@ def _run_status(provider: str | None) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
def _run_set(provider: str, *, from_env: str | None, project: str | None) -> int:
|
||||
def _run_set(
|
||||
provider: str,
|
||||
*,
|
||||
from_env: str | None,
|
||||
project: str | None,
|
||||
base_url: str | None,
|
||||
) -> int:
|
||||
"""Store an API key for `provider`, reading it from env or stdin.
|
||||
|
||||
Returns:
|
||||
Process exit code (`0` on success, `1` on a recoverable input error).
|
||||
"""
|
||||
from deepagents_code.config import is_http_url, normalize_langsmith_endpoint
|
||||
from deepagents_code.model_config import (
|
||||
CODEX_PROVIDER,
|
||||
LANGSMITH_SERVICE,
|
||||
@@ -385,6 +403,25 @@ def _run_set(provider: str, *, from_env: str | None, project: str | None) -> int
|
||||
)
|
||||
return 1
|
||||
|
||||
# Resolve --base-url before reading the store: a malformed value is a clean
|
||||
# input error. `None` means "not passed" (preserve the stored endpoint); an
|
||||
# explicit empty string clears it. LangSmith accepts the `us`/`eu` region
|
||||
# shorthands; any non-empty value must be an http(s) URL so the key is never
|
||||
# paired with a non-HTTP endpoint.
|
||||
base_url_override: str | None = None
|
||||
if base_url is not None:
|
||||
base_url_override = (
|
||||
normalize_langsmith_endpoint(base_url)
|
||||
if is_langsmith(provider)
|
||||
else base_url.strip()
|
||||
)
|
||||
if base_url_override and not is_http_url(base_url_override):
|
||||
print( # noqa: T201
|
||||
"Error: --base-url must be an http(s) URL.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 1
|
||||
|
||||
import os
|
||||
|
||||
from deepagents_code import auth_store
|
||||
@@ -419,16 +456,22 @@ def _run_set(provider: str, *, from_env: str | None, project: str | None) -> int
|
||||
return 1
|
||||
|
||||
try:
|
||||
# Preserve a previously stored project unless `--project` overrides it
|
||||
# (an explicit empty value clears it). Resolved inside the try so a
|
||||
# corrupt store surfaces as a clean error, not a traceback.
|
||||
# Preserve a previously stored project/endpoint unless `--project` /
|
||||
# `--base-url` overrides it (an explicit empty value clears it). Resolved
|
||||
# inside the try so a corrupt store surfaces as a clean error, not a
|
||||
# traceback.
|
||||
stored_project = (
|
||||
project if project is not None else auth_store.get_stored_project(provider)
|
||||
)
|
||||
stored_base_url = (
|
||||
base_url_override
|
||||
if base_url is not None
|
||||
else auth_store.get_stored_base_url(provider)
|
||||
)
|
||||
outcome = auth_store.set_stored_key(
|
||||
provider,
|
||||
key,
|
||||
base_url=auth_store.get_stored_base_url(provider),
|
||||
base_url=stored_base_url,
|
||||
project=stored_project,
|
||||
)
|
||||
except (ValueError, RuntimeError) as exc:
|
||||
|
||||
@@ -379,6 +379,65 @@ _TRACING_API_KEY_ENV_VARS = ("LANGSMITH_API_KEY", "LANGCHAIN_API_KEY")
|
||||
_TRACING_ENDPOINT_ENV_VARS = ("LANGSMITH_ENDPOINT", "LANGCHAIN_ENDPOINT")
|
||||
"""Env vars that point tracing at a non-default (self-hosted/proxied) endpoint."""
|
||||
|
||||
LANGSMITH_US_ENDPOINT = "https://api.smith.langchain.com"
|
||||
"""Canonical LangSmith SaaS endpoint for the US region (the SDK default)."""
|
||||
|
||||
LANGSMITH_EU_ENDPOINT = "https://eu.api.smith.langchain.com"
|
||||
"""Canonical LangSmith SaaS endpoint for the EU region."""
|
||||
|
||||
|
||||
def normalize_langsmith_endpoint(value: str) -> str:
|
||||
"""Resolve a LangSmith endpoint shorthand to its canonical URL.
|
||||
|
||||
Maps the case-insensitive region aliases `us`/`eu` to the LangSmith SaaS
|
||||
endpoints so the CLI `--base-url` flag and the TUI `/auth` prompt share one
|
||||
decode. Any other non-empty value is returned stripped and unchanged (a
|
||||
self-hosted or proxied URL); empty input returns an empty string.
|
||||
|
||||
Args:
|
||||
value: A region alias, a full endpoint URL, or an empty string.
|
||||
|
||||
Returns:
|
||||
The canonical endpoint URL, the stripped literal value, or `""`.
|
||||
"""
|
||||
cleaned = value.strip()
|
||||
if not cleaned:
|
||||
return ""
|
||||
alias = cleaned.lower()
|
||||
if alias == "us":
|
||||
return LANGSMITH_US_ENDPOINT
|
||||
if alias == "eu":
|
||||
return LANGSMITH_EU_ENDPOINT
|
||||
return cleaned
|
||||
|
||||
|
||||
def is_http_url(value: str) -> bool:
|
||||
"""Return whether `value` is a non-empty `http`/`https` URL with a host.
|
||||
|
||||
Guards the LangSmith endpoint so a stored API key is never paired with a
|
||||
non-HTTP, malformed, or schemeless value that could route trace ingestion
|
||||
(and the key) somewhere unintended.
|
||||
|
||||
Args:
|
||||
value: Candidate endpoint URL.
|
||||
|
||||
Returns:
|
||||
`True` when `value` parses as an `http`/`https` URL with a network
|
||||
location that contains no whitespace.
|
||||
"""
|
||||
try:
|
||||
parsed = urlparse(value)
|
||||
except ValueError:
|
||||
return False
|
||||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||||
return False
|
||||
# A real host never contains whitespace, but `urlparse` keeps an internal
|
||||
# space in the netloc (e.g. "exa mple.com"). Such a value would be stored,
|
||||
# written to `LANGSMITH_ENDPOINT`, and its traces may then be dropped at
|
||||
# ingest. Reject it loudly at save time.
|
||||
return not any(char.isspace() for char in parsed.netloc)
|
||||
|
||||
|
||||
_TRACING_RUNS_ENDPOINTS_ENV_VARS = (
|
||||
"LANGSMITH_RUNS_ENDPOINTS",
|
||||
"LANGCHAIN_RUNS_ENDPOINTS",
|
||||
@@ -650,7 +709,9 @@ def _apply_stored_langsmith_tracing(*, replace_project: bool = False) -> None:
|
||||
which bootstrap bridges to `LANGSMITH_TRACING`) is honored and tracing stays
|
||||
off, so the stored key can be paused without deleting it. A custom stored
|
||||
project is applied to `LANGSMITH_PROJECT` when the user has not set one,
|
||||
unless `replace_project` is set for the immediate `/auth` save path.
|
||||
unless `replace_project` is set for the immediate `/auth` save path. A stored
|
||||
endpoint (e.g. the EU region) is applied to `LANGSMITH_ENDPOINT` with the
|
||||
same precedence via `_apply_stored_langsmith_endpoint`.
|
||||
|
||||
No-op when no LangSmith key is stored, so a key supplied only through the
|
||||
environment keeps the prior behavior (tracing stays off unless a flag is
|
||||
@@ -688,6 +749,8 @@ def _apply_stored_langsmith_tracing(*, replace_project: bool = False) -> None:
|
||||
# (tracing stays off unless a flag is set).
|
||||
if entry is None or entry["type"] != "api_key" or not entry["key"]:
|
||||
return
|
||||
if _stored_langsmith_key_is_suppressed(entry["key"]):
|
||||
return
|
||||
|
||||
# The key was bridged onto LANGSMITH_API_KEY by
|
||||
# `apply_stored_service_credentials`. Decide whether to enable tracing.
|
||||
@@ -704,6 +767,10 @@ def _apply_stored_langsmith_tracing(*, replace_project: bool = False) -> None:
|
||||
if not any(flag is True for flag in flags):
|
||||
os.environ["LANGSMITH_TRACING"] = "true"
|
||||
|
||||
_apply_stored_langsmith_endpoint(
|
||||
entry.get("base_url") or None, replace=replace_project
|
||||
)
|
||||
|
||||
project = entry.get("project") or None
|
||||
if replace_project:
|
||||
if project:
|
||||
@@ -715,6 +782,64 @@ def _apply_stored_langsmith_tracing(*, replace_project: bool = False) -> None:
|
||||
os.environ["LANGSMITH_PROJECT"] = project
|
||||
|
||||
|
||||
def _stored_langsmith_key_is_suppressed(stored_key: str) -> bool:
|
||||
"""Return whether an env override keeps `stored_key` from taking effect."""
|
||||
prefixed_names = [f"DEEPAGENTS_CODE_{name}" for name in _TRACING_API_KEY_ENV_VARS]
|
||||
prefixed_values = [
|
||||
os.environ.get(name) or None for name in prefixed_names if name in os.environ
|
||||
]
|
||||
if prefixed_values:
|
||||
return any(value != stored_key for value in prefixed_values)
|
||||
env_key = os.environ.get("LANGSMITH_API_KEY")
|
||||
return bool(env_key and env_key != stored_key)
|
||||
|
||||
|
||||
def _apply_stored_langsmith_endpoint(endpoint: str | None, *, replace: bool) -> None:
|
||||
"""Apply a `/auth`-stored LangSmith endpoint to `LANGSMITH_ENDPOINT`.
|
||||
|
||||
Writes a stored endpoint to the canonical `LANGSMITH_ENDPOINT` and clears the
|
||||
`LANGCHAIN_ENDPOINT` alternate so the SDK can't read a stale value through it.
|
||||
Precedence mirrors the stored project:
|
||||
|
||||
- `replace` (the immediate `/auth` save): the stored endpoint replaces the
|
||||
current value, and a blank endpoint (the US default) clears both names so
|
||||
ingestion falls back to the LangSmith SaaS default.
|
||||
- Startup (`replace=False`): a non-empty `LANGSMITH_ENDPOINT`/`LANGCHAIN_ENDPOINT`
|
||||
already in the environment stays authoritative, so a stored endpoint is
|
||||
applied only when neither is set. A stored credential without an endpoint
|
||||
never clears an existing env value (self-hosted setups keep working).
|
||||
|
||||
Like a stored key, a stored endpoint is trusted by *presence*, not
|
||||
reachability: this never connects to it. A wrong-but-well-formed endpoint (a
|
||||
typo'd or dead host) is applied anyway, and its traces may then be dropped at
|
||||
ingest. `is_http_url` rejects the obviously malformed cases at save time, but
|
||||
if traces never appear the stored endpoint is worth re-checking via `/auth`
|
||||
alongside the key.
|
||||
|
||||
Args:
|
||||
endpoint: The stored endpoint URL, or `None` when none is stored.
|
||||
replace: Whether the stored value should overwrite the current
|
||||
environment (the immediate `/auth` save path).
|
||||
"""
|
||||
canonical, alternate = _TRACING_ENDPOINT_ENV_VARS
|
||||
if replace:
|
||||
if endpoint:
|
||||
os.environ[canonical] = endpoint
|
||||
else:
|
||||
os.environ.pop(canonical, None)
|
||||
os.environ.pop(alternate, None)
|
||||
return
|
||||
if not endpoint:
|
||||
return
|
||||
if any(os.environ.get(var) for var in _TRACING_ENDPOINT_ENV_VARS):
|
||||
return
|
||||
os.environ[canonical] = endpoint
|
||||
# Past the guard above both endpoint vars are falsy, so this only clears an
|
||||
# empty-string `LANGCHAIN_ENDPOINT`; it keeps canonical as the one name the
|
||||
# SDK reads and mirrors the `replace` branch's alternate-clearing.
|
||||
os.environ.pop(alternate, None)
|
||||
|
||||
|
||||
def _ensure_bootstrap() -> None:
|
||||
"""Run one-time bootstrap: dotenv loading and `LANGSMITH_PROJECT` override.
|
||||
|
||||
|
||||
@@ -710,6 +710,10 @@ def show_auth_help() -> None:
|
||||
console.print("[bold]Options:[/bold]", style=theme.PRIMARY)
|
||||
console.print(" --from-env VAR With `set`, copy the key from env var VAR")
|
||||
console.print(" --project NAME With `set langsmith`, set the trace project")
|
||||
console.print(
|
||||
" --base-url URL With `set`, pair an endpoint with the key "
|
||||
"(langsmith accepts us|eu)"
|
||||
)
|
||||
console.print(" -h, --help Show this help message")
|
||||
console.print()
|
||||
console.print(
|
||||
@@ -726,6 +730,7 @@ def show_auth_help() -> None:
|
||||
console.print(
|
||||
" echo $LANGSMITH_API_KEY | dcode auth set langsmith --project my-app"
|
||||
)
|
||||
console.print(" echo $LANGSMITH_API_KEY | dcode auth set langsmith --base-url eu")
|
||||
console.print(" dcode auth status anthropic")
|
||||
console.print(" dcode auth remove anthropic")
|
||||
console.print(" dcode auth path")
|
||||
|
||||
@@ -20,7 +20,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
import os
|
||||
from enum import StrEnum
|
||||
from typing import TYPE_CHECKING, ClassVar
|
||||
from typing import TYPE_CHECKING, ClassVar, NamedTuple
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from textual.binding import Binding, BindingType
|
||||
@@ -30,7 +30,7 @@ from textual.content import Content
|
||||
from textual.message import Message
|
||||
from textual.screen import ModalScreen
|
||||
from textual.style import Style as TStyle
|
||||
from textual.widgets import Input, OptionList, Static
|
||||
from textual.widgets import Input, OptionList, RadioButton, RadioSet, Static
|
||||
from textual.widgets.option_list import Option, OptionDoesNotExist
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -42,9 +42,13 @@ if TYPE_CHECKING:
|
||||
from deepagents_code import auth_store, theme
|
||||
from deepagents_code.auth_display import format_auth_badge
|
||||
from deepagents_code.config import (
|
||||
LANGSMITH_EU_ENDPOINT,
|
||||
LANGSMITH_US_ENDPOINT,
|
||||
apply_stored_langsmith_auth,
|
||||
get_glyphs,
|
||||
is_ascii_mode,
|
||||
is_http_url,
|
||||
normalize_langsmith_endpoint,
|
||||
)
|
||||
from deepagents_code.model_config import (
|
||||
CODEX_PROVIDER,
|
||||
@@ -77,6 +81,81 @@ CONFIGURATION_DOCS_URL = (
|
||||
)
|
||||
|
||||
|
||||
class Region(StrEnum):
|
||||
"""The closed set of LangSmith region selections in the `/auth` prompt."""
|
||||
|
||||
US = "us"
|
||||
EU = "eu"
|
||||
CUSTOM = "custom"
|
||||
|
||||
|
||||
class _RegionSpec(NamedTuple):
|
||||
"""One region row: its enum, radio widget id, SaaS endpoint, and label.
|
||||
|
||||
The single source of truth for the region <-> radio-id <-> endpoint mapping,
|
||||
so `_REGION_BY_RADIO_ID`, `compose`'s radio buttons, `_region_for_endpoint`,
|
||||
and `_resolve_langsmith_endpoint` can't drift apart. `endpoint` is the
|
||||
canonical URL stored for a fixed-SaaS region (`""` for the US default, which
|
||||
clears the stored endpoint); `Region.CUSTOM` carries `""` here because the
|
||||
user supplies its endpoint at prompt time, so it is handled explicitly.
|
||||
"""
|
||||
|
||||
region: Region
|
||||
radio_id: str
|
||||
endpoint: str
|
||||
label: str
|
||||
|
||||
|
||||
_REGION_SPECS: tuple[_RegionSpec, ...] = (
|
||||
_RegionSpec(Region.US, "auth-region-us", "", "United States (default)"),
|
||||
_RegionSpec(Region.EU, "auth-region-eu", LANGSMITH_EU_ENDPOINT, "Europe"),
|
||||
_RegionSpec(
|
||||
Region.CUSTOM, "auth-region-custom", "", "Custom (self-hosted / proxy)"
|
||||
),
|
||||
)
|
||||
|
||||
_REGION_BY_RADIO_ID: dict[str, Region] = {
|
||||
spec.radio_id: spec.region for spec in _REGION_SPECS
|
||||
}
|
||||
"""Map each region radio's widget id to its region, avoiding string-munging."""
|
||||
|
||||
_ENDPOINT_BY_REGION: dict[Region, str] = {
|
||||
spec.region: spec.endpoint for spec in _REGION_SPECS
|
||||
}
|
||||
"""Canonical endpoint stored for each fixed-SaaS region (Custom is dynamic)."""
|
||||
|
||||
|
||||
class ResolvedEndpoint(NamedTuple):
|
||||
"""Outcome of resolving the region selector to an endpoint to persist.
|
||||
|
||||
`endpoint` is the canonical URL to store (empty for the US SaaS default);
|
||||
`error` is a user-facing message when a Custom URL is missing or malformed,
|
||||
in which case `endpoint` is empty and nothing should be saved.
|
||||
"""
|
||||
|
||||
endpoint: str
|
||||
error: str | None
|
||||
|
||||
|
||||
def _region_for_endpoint(base_url: str) -> Region:
|
||||
"""Map a stored LangSmith endpoint to its `/auth` region selection.
|
||||
|
||||
Args:
|
||||
base_url: The stored endpoint, or an empty string for the SaaS default.
|
||||
|
||||
Returns:
|
||||
`Region.US` (blank, or the canonical US SaaS URL that the CLI
|
||||
`--base-url us` shorthand stores), `Region.EU` (the EU SaaS URL), or
|
||||
`Region.CUSTOM` (any other endpoint, e.g. self-hosted).
|
||||
"""
|
||||
if not base_url or base_url == LANGSMITH_US_ENDPOINT:
|
||||
return Region.US
|
||||
for spec in _REGION_SPECS:
|
||||
if spec.endpoint and base_url == spec.endpoint:
|
||||
return spec.region
|
||||
return Region.CUSTOM
|
||||
|
||||
|
||||
PROVIDER_DISPLAY_NAMES: dict[str, str] = {
|
||||
"anthropic": "Anthropic",
|
||||
"azure_openai": "Azure OpenAI",
|
||||
@@ -500,6 +579,21 @@ class AuthPromptScreen(ModalScreen[AuthResult]):
|
||||
margin-bottom: 1;
|
||||
}
|
||||
|
||||
AuthPromptScreen #auth-prompt-project-hint {
|
||||
margin-top: 1;
|
||||
}
|
||||
|
||||
AuthPromptScreen #auth-prompt-region {
|
||||
height: auto;
|
||||
width: 100%;
|
||||
margin-bottom: 1;
|
||||
border: solid $primary-lighten-2;
|
||||
}
|
||||
|
||||
AuthPromptScreen #auth-prompt-region:focus-within {
|
||||
border: solid $primary;
|
||||
}
|
||||
|
||||
AuthPromptScreen #auth-prompt-input,
|
||||
AuthPromptScreen #auth-prompt-base-url {
|
||||
margin-bottom: 1;
|
||||
@@ -556,9 +650,9 @@ class AuthPromptScreen(ModalScreen[AuthResult]):
|
||||
self._allow_empty_submit = allow_empty_submit
|
||||
self._input_placeholder = input_placeholder
|
||||
self._submit_label = submit_label
|
||||
# LangSmith is configured as a tracing service: it has no base-URL
|
||||
# override but does carry an optional project name, and saving a key
|
||||
# turns tracing on.
|
||||
# LangSmith is configured as a tracing service: it carries an optional
|
||||
# project name and an endpoint chosen from a region selector (US/EU SaaS
|
||||
# or a custom self-hosted URL), and saving a key turns tracing on.
|
||||
self._is_langsmith = is_langsmith(provider)
|
||||
# Resolve the current credential source and probe the store, but never
|
||||
# let a corrupt `auth.json`/config crash the screen at construction
|
||||
@@ -577,6 +671,7 @@ class AuthPromptScreen(ModalScreen[AuthResult]):
|
||||
self._advanced_visible = bool(
|
||||
self._existing_base_url or self._existing_project
|
||||
)
|
||||
self._region: Region = _region_for_endpoint(self._existing_base_url)
|
||||
self._store_warning: str | None = None
|
||||
except RuntimeError as exc:
|
||||
logger.warning(
|
||||
@@ -590,9 +685,26 @@ class AuthPromptScreen(ModalScreen[AuthResult]):
|
||||
self._existing_base_url = ""
|
||||
self._existing_project = ""
|
||||
self._advanced_visible = False
|
||||
self._region = Region.US
|
||||
self._store_warning = (
|
||||
f"Credential file is unreadable ({exc}). Saving here will overwrite it."
|
||||
)
|
||||
# Surface when an environment endpoint is set: at startup an existing
|
||||
# `LANGSMITH_ENDPOINT`/`LANGCHAIN_ENDPOINT` takes precedence over the
|
||||
# stored region, so without this note the radio could show one region
|
||||
# while traces route somewhere else. (The note fires on presence, not on
|
||||
# divergence — it may show even when the env value matches the stored
|
||||
# region.) Saving here applies the selection (the save path replaces the
|
||||
# env value), so word the note around that.
|
||||
self._endpoint_env_notice: str | None = None
|
||||
if self._is_langsmith and any(
|
||||
os.environ.get(var) for var in ("LANGSMITH_ENDPOINT", "LANGCHAIN_ENDPOINT")
|
||||
):
|
||||
self._endpoint_env_notice = (
|
||||
"An endpoint is set in your environment "
|
||||
"(LANGSMITH_ENDPOINT/LANGCHAIN_ENDPOINT) and takes precedence at "
|
||||
"startup; saving a region here applies your selection instead."
|
||||
)
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
"""Compose the prompt.
|
||||
@@ -751,6 +863,55 @@ class AuthPromptScreen(ModalScreen[AuthResult]):
|
||||
key_meta.display = self._advanced_visible
|
||||
yield key_meta
|
||||
if self._is_langsmith:
|
||||
if self._endpoint_env_notice:
|
||||
yield Static(
|
||||
Content.from_markup("$msg", msg=self._endpoint_env_notice),
|
||||
classes="auth-prompt-meta",
|
||||
id="auth-prompt-endpoint-env-notice",
|
||||
)
|
||||
region_label = Static(
|
||||
Content.from_markup("[bold]LangSmith region[/bold]"),
|
||||
classes="auth-prompt-meta",
|
||||
id="auth-prompt-region-label",
|
||||
)
|
||||
region_label.display = self._advanced_visible
|
||||
yield region_label
|
||||
region_set = RadioSet(
|
||||
*(
|
||||
RadioButton(
|
||||
spec.label,
|
||||
value=self._region == spec.region,
|
||||
id=spec.radio_id,
|
||||
)
|
||||
for spec in _REGION_SPECS
|
||||
),
|
||||
id="auth-prompt-region",
|
||||
)
|
||||
region_set.display = self._advanced_visible
|
||||
yield region_set
|
||||
custom_visible = (
|
||||
self._advanced_visible and self._region == Region.CUSTOM
|
||||
)
|
||||
base_url_input = Input(
|
||||
value=(
|
||||
self._existing_base_url if self._region == Region.CUSTOM else ""
|
||||
),
|
||||
placeholder="https://my-langsmith.example.com",
|
||||
id="auth-prompt-base-url",
|
||||
)
|
||||
base_url_input.display = custom_visible
|
||||
yield base_url_input
|
||||
base_url_hint_widget = Static(
|
||||
Content.from_markup(
|
||||
"Point tracing at a self-hosted or proxied LangSmith. "
|
||||
"Sets [bold]LANGSMITH_ENDPOINT[/bold]; must be an "
|
||||
"http(s) URL."
|
||||
),
|
||||
classes="auth-prompt-meta",
|
||||
id="auth-prompt-base-url-hint",
|
||||
)
|
||||
base_url_hint_widget.display = custom_visible
|
||||
yield base_url_hint_widget
|
||||
project_label = Static(
|
||||
Content.from_markup("[bold]Project name[/bold]"),
|
||||
classes="auth-prompt-meta",
|
||||
@@ -953,6 +1114,8 @@ class AuthPromptScreen(ModalScreen[AuthResult]):
|
||||
self._advanced_visible = not self._advanced_visible
|
||||
for selector in (
|
||||
"#auth-prompt-key-meta",
|
||||
"#auth-prompt-region-label",
|
||||
"#auth-prompt-region",
|
||||
"#auth-prompt-base-url-label",
|
||||
"#auth-prompt-base-url",
|
||||
"#auth-prompt-base-url-hint",
|
||||
@@ -962,12 +1125,36 @@ class AuthPromptScreen(ModalScreen[AuthResult]):
|
||||
):
|
||||
for widget in self.query(selector):
|
||||
widget.display = self._advanced_visible
|
||||
# For LangSmith the custom URL field only shows under the Custom region,
|
||||
# so re-apply its region-gated visibility after the blanket toggle above.
|
||||
if self._is_langsmith:
|
||||
self._refresh_langsmith_custom_visibility()
|
||||
self.query_one("#auth-prompt-advanced-toggle", Static).update(
|
||||
self._build_advanced_toggle_label()
|
||||
)
|
||||
if not self._advanced_visible:
|
||||
self.query_one("#auth-prompt-input", Input).focus()
|
||||
|
||||
def _refresh_langsmith_custom_visibility(self) -> None:
|
||||
"""Show the custom URL field only when Advanced is open and region is Custom."""
|
||||
custom_visible = self._advanced_visible and self._region == Region.CUSTOM
|
||||
for selector in ("#auth-prompt-base-url", "#auth-prompt-base-url-hint"):
|
||||
for widget in self.query(selector):
|
||||
widget.display = custom_visible
|
||||
|
||||
def on_radio_set_changed(self, event: RadioSet.Changed) -> None:
|
||||
"""Track the chosen LangSmith region and reveal the custom URL field."""
|
||||
event.stop()
|
||||
region = _REGION_BY_RADIO_ID.get(event.pressed.id or "")
|
||||
if region is None:
|
||||
# An unknown id (e.g. a renamed radio) must not silently degrade to a
|
||||
# region — leave the current selection unchanged rather than guess.
|
||||
return
|
||||
self._region = region
|
||||
self._refresh_langsmith_custom_visibility()
|
||||
if self._region == Region.CUSTOM:
|
||||
self.query_one("#auth-prompt-base-url", Input).focus()
|
||||
|
||||
def on_click(self, event: Click) -> None:
|
||||
"""Open style-embedded hyperlinks or toggle Advanced."""
|
||||
widget = event.widget
|
||||
@@ -1002,6 +1189,34 @@ class AuthPromptScreen(ModalScreen[AuthResult]):
|
||||
colors = theme.get_theme_colors(self)
|
||||
container.styles.border = ("ascii", colors.success)
|
||||
|
||||
def _resolve_langsmith_endpoint(self) -> ResolvedEndpoint:
|
||||
"""Resolve the endpoint to store from the LangSmith region selector.
|
||||
|
||||
Returns:
|
||||
A `ResolvedEndpoint`: the canonical endpoint to persist (empty for
|
||||
the US SaaS default), and a non-`None` error when a custom URL is
|
||||
missing or malformed (so an explicit `Custom` selection never
|
||||
silently routes the key to the US SaaS default, and the key is
|
||||
never paired with a non-http(s) endpoint).
|
||||
"""
|
||||
if self._region != Region.CUSTOM:
|
||||
# US and EU resolve to their fixed SaaS endpoints from the table
|
||||
# (US -> "" clears the stored endpoint back to the SaaS default).
|
||||
return ResolvedEndpoint(_ENDPOINT_BY_REGION[self._region], None)
|
||||
raw = self.query_one("#auth-prompt-base-url", Input).value.strip()
|
||||
if not raw:
|
||||
# `Custom` is a deliberate non-default choice; a blank field must not
|
||||
# silently fall back to the US SaaS default (which would reroute the
|
||||
# key and traces to the very endpoint a self-hosted user is avoiding).
|
||||
missing_url_error = (
|
||||
"Custom endpoint URL is required (or choose United States/Europe)."
|
||||
)
|
||||
return ResolvedEndpoint("", missing_url_error)
|
||||
endpoint = normalize_langsmith_endpoint(raw)
|
||||
if not is_http_url(endpoint):
|
||||
return ResolvedEndpoint("", "Custom endpoint must be an http(s) URL.")
|
||||
return ResolvedEndpoint(endpoint, None)
|
||||
|
||||
def on_input_submitted(self, event: Input.Submitted) -> None:
|
||||
"""Validate, persist, and dismiss.
|
||||
|
||||
@@ -1012,10 +1227,27 @@ class AuthPromptScreen(ModalScreen[AuthResult]):
|
||||
event.stop()
|
||||
cleaned = self.query_one("#auth-prompt-input", Input).value.strip()
|
||||
if self._is_langsmith:
|
||||
base_url = ""
|
||||
project = self.query_one("#auth-prompt-project", Input).value.strip()
|
||||
resolved = self._resolve_langsmith_endpoint()
|
||||
if resolved.error:
|
||||
self._show_error(resolved.error)
|
||||
return
|
||||
base_url = resolved.endpoint
|
||||
else:
|
||||
base_url = self.query_one("#auth-prompt-base-url", Input).value.strip()
|
||||
# Match the CLI `--base-url` guard: never pair the key with a
|
||||
# non-http(s) endpoint. Validate only a *changed* value, though: the
|
||||
# field is pre-filled with the stored base URL, so rotating just the
|
||||
# key must not be blocked by a legacy non-http(s) endpoint (the CLI
|
||||
# never validated base URLs before this feature). A new or edited
|
||||
# value must still be http(s).
|
||||
if (
|
||||
base_url
|
||||
and base_url != self._existing_base_url
|
||||
and not is_http_url(base_url)
|
||||
):
|
||||
self._show_error("Base URL must be an http(s) URL.")
|
||||
return
|
||||
project = ""
|
||||
if not cleaned:
|
||||
if self._allow_empty_submit:
|
||||
|
||||
@@ -220,6 +220,162 @@ class TestSet:
|
||||
assert auth_store.get_stored_key("anthropic") is None
|
||||
assert "--project is only valid for langsmith" in capsys.readouterr().err
|
||||
|
||||
def test_set_langsmith_base_url_eu_alias(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""`--base-url eu` resolves the shorthand to the canonical EU URL."""
|
||||
from deepagents_code.config import LANGSMITH_EU_ENDPOINT
|
||||
|
||||
monkeypatch.setattr(sys, "stdin", io.StringIO("lsv2_test\n"))
|
||||
code = run_auth_command(
|
||||
_ns(
|
||||
auth_command="set",
|
||||
provider="langsmith",
|
||||
from_env=None,
|
||||
project=None,
|
||||
base_url="eu",
|
||||
)
|
||||
)
|
||||
assert code == 0
|
||||
assert auth_store.get_stored_base_url("langsmith") == LANGSMITH_EU_ENDPOINT
|
||||
|
||||
def test_set_langsmith_base_url_full_url(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""A full custom endpoint URL is stored verbatim."""
|
||||
monkeypatch.setattr(sys, "stdin", io.StringIO("lsv2_test\n"))
|
||||
code = run_auth_command(
|
||||
_ns(
|
||||
auth_command="set",
|
||||
provider="langsmith",
|
||||
from_env=None,
|
||||
project=None,
|
||||
base_url="https://langsmith.internal.example.com",
|
||||
)
|
||||
)
|
||||
assert code == 0
|
||||
assert (
|
||||
auth_store.get_stored_base_url("langsmith")
|
||||
== "https://langsmith.internal.example.com"
|
||||
)
|
||||
|
||||
def test_set_base_url_invalid_scheme_rejected(
|
||||
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
"""A non-http(s) `--base-url` exits non-zero and stores nothing."""
|
||||
monkeypatch.setattr(sys, "stdin", io.StringIO("lsv2_test\n"))
|
||||
code = run_auth_command(
|
||||
_ns(
|
||||
auth_command="set",
|
||||
provider="langsmith",
|
||||
from_env=None,
|
||||
project=None,
|
||||
base_url="ftp://nope.example.com",
|
||||
)
|
||||
)
|
||||
assert code == 1
|
||||
assert auth_store.get_stored_key("langsmith") is None
|
||||
assert "--base-url must be an http(s) URL" in capsys.readouterr().err
|
||||
|
||||
def test_set_base_url_malformed_ipv6_rejected_cleanly(
|
||||
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
"""A malformed IPv6 `--base-url` is a clean input error, not a traceback."""
|
||||
# `urlparse("http://[::1")` raises ValueError; the guard must catch it and
|
||||
# exit 1 with the standard message rather than crash the CLI.
|
||||
monkeypatch.setattr(sys, "stdin", io.StringIO("lsv2_test\n"))
|
||||
code = run_auth_command(
|
||||
_ns(
|
||||
auth_command="set",
|
||||
provider="langsmith",
|
||||
from_env=None,
|
||||
project=None,
|
||||
base_url="http://[::1",
|
||||
)
|
||||
)
|
||||
assert code == 1
|
||||
assert auth_store.get_stored_key("langsmith") is None
|
||||
assert "--base-url must be an http(s) URL" in capsys.readouterr().err
|
||||
|
||||
def test_set_empty_base_url_clears_existing(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""An explicit empty `--base-url` clears a previously stored endpoint."""
|
||||
from deepagents_code.config import LANGSMITH_EU_ENDPOINT
|
||||
|
||||
auth_store.set_stored_key("langsmith", "old", base_url=LANGSMITH_EU_ENDPOINT)
|
||||
monkeypatch.setattr(sys, "stdin", io.StringIO("new\n"))
|
||||
code = run_auth_command(
|
||||
_ns(
|
||||
auth_command="set",
|
||||
provider="langsmith",
|
||||
from_env=None,
|
||||
project=None,
|
||||
base_url="",
|
||||
)
|
||||
)
|
||||
assert code == 0
|
||||
assert auth_store.get_stored_key("langsmith") == "new"
|
||||
assert auth_store.get_stored_base_url("langsmith") is None
|
||||
|
||||
def test_set_base_url_none_preserves_existing(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Rotating the key without `--base-url` keeps the stored endpoint."""
|
||||
from deepagents_code.config import LANGSMITH_EU_ENDPOINT
|
||||
|
||||
auth_store.set_stored_key("langsmith", "old", base_url=LANGSMITH_EU_ENDPOINT)
|
||||
monkeypatch.setattr(sys, "stdin", io.StringIO("new\n"))
|
||||
code = run_auth_command(
|
||||
_ns(
|
||||
auth_command="set",
|
||||
provider="langsmith",
|
||||
from_env=None,
|
||||
project=None,
|
||||
base_url=None,
|
||||
)
|
||||
)
|
||||
assert code == 0
|
||||
assert auth_store.get_stored_base_url("langsmith") == LANGSMITH_EU_ENDPOINT
|
||||
|
||||
def test_set_non_langsmith_base_url_stored(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""A valid `--base-url` is stored for a non-LangSmith provider."""
|
||||
monkeypatch.setattr(sys, "stdin", io.StringIO("sk-openai-abc\n"))
|
||||
code = run_auth_command(
|
||||
_ns(
|
||||
auth_command="set",
|
||||
provider="openai",
|
||||
from_env=None,
|
||||
project=None,
|
||||
base_url="https://proxy.internal.example.com/v1",
|
||||
)
|
||||
)
|
||||
assert code == 0
|
||||
assert (
|
||||
auth_store.get_stored_base_url("openai")
|
||||
== "https://proxy.internal.example.com/v1"
|
||||
)
|
||||
|
||||
def test_set_non_langsmith_base_url_rejects_region_alias(
|
||||
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
"""Region aliases are LangSmith-only: `eu` stays a literal (invalid) URL."""
|
||||
monkeypatch.setattr(sys, "stdin", io.StringIO("sk-openai-abc\n"))
|
||||
code = run_auth_command(
|
||||
_ns(
|
||||
auth_command="set",
|
||||
provider="openai",
|
||||
from_env=None,
|
||||
project=None,
|
||||
base_url="eu",
|
||||
)
|
||||
)
|
||||
assert code == 1
|
||||
assert auth_store.get_stored_key("openai") is None
|
||||
assert "--base-url must be an http(s) URL" in capsys.readouterr().err
|
||||
|
||||
def test_set_from_unset_env_fails(
|
||||
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
|
||||
@@ -10,17 +10,20 @@ from typing import TYPE_CHECKING, Any, cast
|
||||
import pytest
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.containers import Container
|
||||
from textual.widgets import Input, OptionList, Static
|
||||
from textual.widgets import Input, OptionList, RadioButton, RadioSet, Static
|
||||
|
||||
from deepagents_code import auth_store, model_config
|
||||
from deepagents_code.config import get_glyphs
|
||||
from deepagents_code.widgets.auth import (
|
||||
_ENDPOINT_BY_REGION,
|
||||
PROVIDER_API_KEY_URLS,
|
||||
PROVIDER_DISPLAY_NAMES,
|
||||
AuthManagerScreen,
|
||||
AuthPromptScreen,
|
||||
AuthResult,
|
||||
Region,
|
||||
_is_safe_acquisition_url,
|
||||
_region_for_endpoint,
|
||||
provider_display_name,
|
||||
provider_short_name,
|
||||
)
|
||||
@@ -662,6 +665,270 @@ api_key_url = "javascript:alert(1)"
|
||||
assert os.environ["LANGSMITH_API_KEY"] == "lsv2_live"
|
||||
assert os.environ["LANGSMITH_TRACING"] == "true"
|
||||
|
||||
async def test_langsmith_prompt_renders_region_selector(self) -> None:
|
||||
"""The LangSmith prompt shows a region selector defaulting to US."""
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
await pilot.press("f2")
|
||||
await pilot.pause()
|
||||
region = app.screen.query_one("#auth-prompt-region", RadioSet)
|
||||
assert region.display is True
|
||||
us = app.screen.query_one("#auth-region-us", RadioButton)
|
||||
assert us.value is True
|
||||
# The custom URL field stays hidden until Custom is chosen.
|
||||
assert app.screen.query_one("#auth-prompt-base-url", Input).display is False
|
||||
|
||||
async def test_langsmith_eu_region_saves_eu_endpoint(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Choosing the Europe region stores the canonical EU endpoint."""
|
||||
from deepagents_code.config import LANGSMITH_EU_ENDPOINT
|
||||
|
||||
monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
await pilot.press("f2")
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-region-eu", RadioButton).value = True
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-prompt-input", Input).value = "lsv2_live"
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
assert app.prompt_result is AuthResult.SAVED
|
||||
assert auth_store.get_stored_base_url("langsmith") == LANGSMITH_EU_ENDPOINT
|
||||
|
||||
async def test_langsmith_custom_region_reveals_input_and_validates(self) -> None:
|
||||
"""Custom reveals the URL field and rejects a non-http(s) value."""
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
await pilot.press("f2")
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-region-custom", RadioButton).value = True
|
||||
await pilot.pause()
|
||||
custom = app.screen.query_one("#auth-prompt-base-url", Input)
|
||||
assert custom.display is True
|
||||
app.screen.query_one("#auth-prompt-input", Input).value = "lsv2_live"
|
||||
custom.value = "ftp://nope.example.com"
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
assert app.prompt_dismissed is False
|
||||
error = app.screen.query_one("#auth-prompt-error", Static)
|
||||
assert "http(s)" in str(error.render())
|
||||
assert auth_store.get_stored_key("langsmith") is None
|
||||
# A valid URL now saves the pair.
|
||||
custom.value = "https://langsmith.internal.example.com"
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
assert app.prompt_result is AuthResult.SAVED
|
||||
assert (
|
||||
auth_store.get_stored_base_url("langsmith")
|
||||
== "https://langsmith.internal.example.com"
|
||||
)
|
||||
|
||||
async def test_langsmith_existing_eu_endpoint_preselects_region(self) -> None:
|
||||
"""Reopening with a stored EU endpoint preselects the Europe region."""
|
||||
from deepagents_code.config import LANGSMITH_EU_ENDPOINT
|
||||
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_existing", base_url=LANGSMITH_EU_ENDPOINT
|
||||
)
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
eu = app.screen.query_one("#auth-region-eu", RadioButton)
|
||||
assert eu.value is True
|
||||
|
||||
async def test_langsmith_existing_custom_endpoint_preselects_and_prefills(
|
||||
self,
|
||||
) -> None:
|
||||
"""Reopening a stored self-hosted URL preselects Custom and prefills it."""
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_existing", base_url="https://self.example.com"
|
||||
)
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
custom = app.screen.query_one("#auth-region-custom", RadioButton)
|
||||
assert custom.value is True
|
||||
field = app.screen.query_one("#auth-prompt-base-url", Input)
|
||||
assert field.display is True
|
||||
assert field.value == "https://self.example.com"
|
||||
|
||||
async def test_langsmith_cli_stored_us_url_preselects_us(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""A CLI-stored literal US URL reopens as United States, not Custom."""
|
||||
from deepagents_code.config import LANGSMITH_US_ENDPOINT
|
||||
|
||||
monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_existing", base_url=LANGSMITH_US_ENDPOINT
|
||||
)
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
us = app.screen.query_one("#auth-region-us", RadioButton)
|
||||
assert us.value is True
|
||||
assert app.screen.query_one("#auth-prompt-base-url", Input).display is False
|
||||
|
||||
async def test_langsmith_custom_empty_url_blocks_save(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Custom with a blank URL errors instead of silently defaulting to US."""
|
||||
monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
await pilot.press("f2")
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-region-custom", RadioButton).value = True
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-prompt-input", Input).value = "lsv2_live"
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
assert app.prompt_dismissed is False
|
||||
error = app.screen.query_one("#auth-prompt-error", Static)
|
||||
assert "required" in str(error.render())
|
||||
assert auth_store.get_stored_key("langsmith") is None
|
||||
|
||||
async def test_langsmith_region_switch_discards_typed_custom_url(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Typing a custom URL then switching to US saves the US default, not it."""
|
||||
monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
await pilot.press("f2")
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-region-custom", RadioButton).value = True
|
||||
await pilot.pause()
|
||||
app.screen.query_one(
|
||||
"#auth-prompt-base-url", Input
|
||||
).value = "https://typed.example.com"
|
||||
app.screen.query_one("#auth-region-us", RadioButton).value = True
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-prompt-input", Input).value = "lsv2_live"
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
assert app.prompt_result is AuthResult.SAVED
|
||||
assert auth_store.get_stored_base_url("langsmith") is None
|
||||
|
||||
async def test_langsmith_region_switch_custom_to_eu_discards_typed_url(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Typing a custom URL then switching to Europe saves the EU endpoint."""
|
||||
from deepagents_code.config import LANGSMITH_EU_ENDPOINT
|
||||
|
||||
monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
await pilot.press("f2")
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-region-custom", RadioButton).value = True
|
||||
await pilot.pause()
|
||||
app.screen.query_one(
|
||||
"#auth-prompt-base-url", Input
|
||||
).value = "https://typed.example.com"
|
||||
app.screen.query_one("#auth-region-eu", RadioButton).value = True
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-prompt-input", Input).value = "lsv2_live"
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
assert app.prompt_result is AuthResult.SAVED
|
||||
assert auth_store.get_stored_base_url("langsmith") == LANGSMITH_EU_ENDPOINT
|
||||
|
||||
async def test_langsmith_unknown_radio_id_leaves_region_unchanged(self) -> None:
|
||||
"""A `Changed` from an unmapped radio id is a no-op, not a silent US reset."""
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
await pilot.press("f2")
|
||||
await pilot.pause()
|
||||
screen = cast("AuthPromptScreen", app.screen)
|
||||
screen.query_one("#auth-region-eu", RadioButton).value = True
|
||||
await pilot.pause()
|
||||
region_set = screen.query_one("#auth-prompt-region", RadioSet)
|
||||
stray = RadioButton("x", id="auth-region-bogus")
|
||||
# Synthesize a Changed from a foreign (renamed/unknown) radio id.
|
||||
screen.on_radio_set_changed(RadioSet.Changed(region_set, stray))
|
||||
assert screen._region == Region.EU # unchanged, not degraded to us
|
||||
|
||||
async def test_langsmith_env_notice_via_alternate_var(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""The precedence notice also fires for a lone LANGCHAIN_ENDPOINT."""
|
||||
monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
|
||||
monkeypatch.setenv("LANGCHAIN_ENDPOINT", "https://from-alt-env.example.com")
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
assert app.screen.query("#auth-prompt-endpoint-env-notice")
|
||||
|
||||
async def test_langsmith_no_env_endpoint_hides_precedence_notice(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""With no env endpoint the notice is absent (no spurious warning)."""
|
||||
monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
|
||||
monkeypatch.delenv("LANGCHAIN_ENDPOINT", raising=False)
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
assert not app.screen.query("#auth-prompt-endpoint-env-notice")
|
||||
|
||||
async def test_non_langsmith_invalid_base_url_blocks_save(self) -> None:
|
||||
"""A newly typed non-http(s) base URL is rejected for a non-LangSmith key."""
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("openai", "OPENAI_API_KEY")
|
||||
await pilot.pause()
|
||||
await pilot.press("f2")
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-prompt-input", Input).value = "sk-x"
|
||||
app.screen.query_one(
|
||||
"#auth-prompt-base-url", Input
|
||||
).value = "ftp://nope.example.com"
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
assert app.prompt_dismissed is False
|
||||
error = app.screen.query_one("#auth-prompt-error", Static)
|
||||
assert "http(s)" in str(error.render())
|
||||
assert auth_store.get_stored_key("openai") is None
|
||||
|
||||
async def test_non_langsmith_preexisting_invalid_base_url_allows_key_reroll(
|
||||
self,
|
||||
) -> None:
|
||||
"""A legacy non-http(s) base URL, left unchanged, never blocks a key update."""
|
||||
# A schemeless value could only get stored before this feature validated;
|
||||
# rotating just the key must not force the user to fix it.
|
||||
auth_store.set_stored_key("openai", "sk-old", base_url="localhost:8000")
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("openai", "OPENAI_API_KEY")
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#auth-prompt-input", Input).value = "sk-new"
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
assert app.prompt_result is AuthResult.SAVED
|
||||
assert auth_store.get_stored_key("openai") == "sk-new"
|
||||
assert auth_store.get_stored_base_url("openai") == "localhost:8000"
|
||||
|
||||
async def test_base_url_round_trips_on_submit(self) -> None:
|
||||
"""A base URL typed alongside the key is persisted as the pair."""
|
||||
app = _AuthHostApp()
|
||||
@@ -744,14 +1011,15 @@ api_key_url = "javascript:alert(1)"
|
||||
assert auth_store.get_stored_key("langsmith") == "lsv2_test"
|
||||
assert auth_store.get_stored_project("langsmith") == "my-app"
|
||||
|
||||
async def test_langsmith_prompt_has_no_base_url_field(self) -> None:
|
||||
"""The LangSmith prompt swaps the base-URL field for the project field."""
|
||||
async def test_langsmith_prompt_hides_custom_url_field_by_default(self) -> None:
|
||||
"""LangSmith shows the project field; the custom URL field is hidden at US."""
|
||||
app = _AuthHostApp()
|
||||
async with app.run_test() as pilot:
|
||||
app.show_prompt("langsmith", "LANGSMITH_API_KEY")
|
||||
await pilot.pause()
|
||||
assert not app.screen.query("#auth-prompt-base-url")
|
||||
assert app.screen.query("#auth-prompt-project")
|
||||
# The base-URL field exists (for Custom) but stays hidden under US.
|
||||
assert app.screen.query_one("#auth-prompt-base-url", Input).display is False
|
||||
|
||||
async def test_empty_submit_shows_error_and_does_not_dismiss(self) -> None:
|
||||
"""Empty input renders an inline error instead of dismissing."""
|
||||
@@ -2189,3 +2457,20 @@ class TestCodexAuthInManager:
|
||||
break
|
||||
assert results == [True]
|
||||
assert any("plus" in note for note in notices)
|
||||
|
||||
|
||||
class TestRegionEndpointMapping:
|
||||
"""Round-trip tests for the region <-> endpoint single-source table."""
|
||||
|
||||
def test_fixed_saas_regions_round_trip(self) -> None:
|
||||
"""Each fixed-SaaS region maps to an endpoint that maps back to it."""
|
||||
from deepagents_code.config import LANGSMITH_US_ENDPOINT
|
||||
|
||||
for region in (Region.US, Region.EU):
|
||||
assert _region_for_endpoint(_ENDPOINT_BY_REGION[region]) == region
|
||||
# The US alias URL (what `--base-url us` persists) also classifies as US.
|
||||
assert _region_for_endpoint(LANGSMITH_US_ENDPOINT) == Region.US
|
||||
|
||||
def test_self_hosted_endpoint_is_custom(self) -> None:
|
||||
"""Any endpoint outside the SaaS set classifies as Custom."""
|
||||
assert _region_for_endpoint("https://self.example.com") == Region.CUSTOM
|
||||
|
||||
@@ -15,6 +15,8 @@ from deepagents_code._env_vars import SERVER_ENV_PREFIX
|
||||
from deepagents_code._version import __version__
|
||||
from deepagents_code.config import (
|
||||
CLI_MAX_RETRIES_KEY,
|
||||
LANGSMITH_EU_ENDPOINT,
|
||||
LANGSMITH_US_ENDPOINT,
|
||||
RECOMMENDED_SAFE_SHELL_COMMANDS,
|
||||
SHELL_ALLOW_ALL,
|
||||
LangSmithApiError,
|
||||
@@ -41,8 +43,10 @@ from deepagents_code.config import (
|
||||
fetch_langsmith_project_url,
|
||||
fetch_langsmith_project_url_or_raise,
|
||||
get_langsmith_project_name,
|
||||
is_http_url,
|
||||
is_langsmith_redaction_enabled,
|
||||
newline_shortcut,
|
||||
normalize_langsmith_endpoint,
|
||||
parse_shell_allow_list,
|
||||
reset_langsmith_url_cache,
|
||||
settings,
|
||||
@@ -2696,6 +2700,215 @@ class TestApplyStoredLangSmithTracing:
|
||||
assert "LANGSMITH_TRACING" not in os.environ
|
||||
assert any("may be corrupt" in r.getMessage() for r in caplog.records)
|
||||
|
||||
def test_applies_stored_endpoint(
|
||||
self,
|
||||
fake_state_dir: Path, # noqa: ARG002
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A stored endpoint is applied to LANGSMITH_ENDPOINT at startup."""
|
||||
import os
|
||||
|
||||
from deepagents_code import auth_store
|
||||
|
||||
for var in ("LANGSMITH_TRACING", "LANGSMITH_ENDPOINT", "LANGCHAIN_ENDPOINT"):
|
||||
monkeypatch.delenv(var, raising=False)
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_test", base_url=LANGSMITH_EU_ENDPOINT
|
||||
)
|
||||
_apply_stored_langsmith_tracing()
|
||||
assert os.environ["LANGSMITH_ENDPOINT"] == LANGSMITH_EU_ENDPOINT
|
||||
assert "LANGCHAIN_ENDPOINT" not in os.environ
|
||||
|
||||
def test_stored_endpoint_does_not_override_env(
|
||||
self,
|
||||
fake_state_dir: Path, # noqa: ARG002
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""An explicit LANGSMITH_ENDPOINT wins over the stored endpoint at startup."""
|
||||
import os
|
||||
|
||||
from deepagents_code import auth_store
|
||||
|
||||
monkeypatch.delenv("LANGSMITH_TRACING", raising=False)
|
||||
monkeypatch.delenv("LANGCHAIN_ENDPOINT", raising=False)
|
||||
monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://from-env.example.com")
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_test", base_url=LANGSMITH_EU_ENDPOINT
|
||||
)
|
||||
_apply_stored_langsmith_tracing()
|
||||
assert os.environ["LANGSMITH_ENDPOINT"] == "https://from-env.example.com"
|
||||
|
||||
def test_no_stored_endpoint_leaves_env_endpoint_untouched(
|
||||
self,
|
||||
fake_state_dir: Path, # noqa: ARG002
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A stored key without an endpoint never clears an existing env endpoint."""
|
||||
import os
|
||||
|
||||
from deepagents_code import auth_store
|
||||
|
||||
monkeypatch.delenv("LANGSMITH_TRACING", raising=False)
|
||||
monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://self-hosted.example.com")
|
||||
auth_store.set_stored_key("langsmith", "lsv2_test")
|
||||
_apply_stored_langsmith_tracing()
|
||||
assert os.environ["LANGSMITH_ENDPOINT"] == "https://self-hosted.example.com"
|
||||
|
||||
def test_alternate_env_endpoint_blocks_stored_endpoint(
|
||||
self,
|
||||
fake_state_dir: Path, # noqa: ARG002
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A lone LANGCHAIN_ENDPOINT (the alternate) also wins over a stored one."""
|
||||
import os
|
||||
|
||||
from deepagents_code import auth_store
|
||||
|
||||
monkeypatch.delenv("LANGSMITH_TRACING", raising=False)
|
||||
monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
|
||||
monkeypatch.setenv("LANGCHAIN_ENDPOINT", "https://alt-env.example.com")
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_test", base_url=LANGSMITH_EU_ENDPOINT
|
||||
)
|
||||
_apply_stored_langsmith_tracing()
|
||||
# The stored endpoint must not be applied, and the alternate is preserved.
|
||||
assert "LANGSMITH_ENDPOINT" not in os.environ
|
||||
assert os.environ["LANGCHAIN_ENDPOINT"] == "https://alt-env.example.com"
|
||||
|
||||
def test_replace_applies_stored_endpoint_over_existing_env(
|
||||
self,
|
||||
fake_state_dir: Path, # noqa: ARG002
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Immediate `/auth` save replaces the endpoint and clears the alternate."""
|
||||
import os
|
||||
|
||||
from deepagents_code import auth_store
|
||||
|
||||
monkeypatch.delenv("LANGSMITH_TRACING", raising=False)
|
||||
monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://old.example.com")
|
||||
monkeypatch.setenv("LANGCHAIN_ENDPOINT", "https://old-alt.example.com")
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_test", base_url=LANGSMITH_EU_ENDPOINT
|
||||
)
|
||||
_apply_stored_langsmith_tracing(replace_project=True)
|
||||
assert os.environ["LANGSMITH_ENDPOINT"] == LANGSMITH_EU_ENDPOINT
|
||||
assert "LANGCHAIN_ENDPOINT" not in os.environ
|
||||
|
||||
def test_replace_clears_endpoint_when_blank(
|
||||
self,
|
||||
fake_state_dir: Path, # noqa: ARG002
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Immediate `/auth` save clears the endpoint when no base URL is stored."""
|
||||
import os
|
||||
|
||||
from deepagents_code import auth_store
|
||||
|
||||
monkeypatch.delenv("LANGSMITH_TRACING", raising=False)
|
||||
monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://old.example.com")
|
||||
auth_store.set_stored_key("langsmith", "lsv2_test")
|
||||
_apply_stored_langsmith_tracing(replace_project=True)
|
||||
assert "LANGSMITH_ENDPOINT" not in os.environ
|
||||
|
||||
def test_disabled_tracing_leaves_endpoint_unset(
|
||||
self,
|
||||
fake_state_dir: Path, # noqa: ARG002
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A paused key (tracing off) never applies its stored endpoint.
|
||||
|
||||
The endpoint is applied only after the enable decision, so a deliberate
|
||||
opt-out must not leak a stored endpoint into the process env.
|
||||
"""
|
||||
import os
|
||||
|
||||
from deepagents_code import auth_store
|
||||
|
||||
monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
|
||||
monkeypatch.delenv("LANGCHAIN_ENDPOINT", raising=False)
|
||||
monkeypatch.setenv("LANGSMITH_TRACING", "false")
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_test", base_url=LANGSMITH_EU_ENDPOINT
|
||||
)
|
||||
_apply_stored_langsmith_tracing()
|
||||
assert "LANGSMITH_ENDPOINT" not in os.environ
|
||||
|
||||
def test_prefixed_key_override_leaves_stored_endpoint_unset(
|
||||
self,
|
||||
fake_state_dir: Path, # noqa: ARG002
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A session-scoped key is not paired with a stale stored endpoint."""
|
||||
import os
|
||||
|
||||
from deepagents_code import auth_store
|
||||
|
||||
for var in ("LANGSMITH_ENDPOINT", "LANGCHAIN_ENDPOINT"):
|
||||
monkeypatch.delenv(var, raising=False)
|
||||
monkeypatch.setenv("DEEPAGENTS_CODE_LANGSMITH_API_KEY", "lsv2_prefixed")
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_stored", base_url=LANGSMITH_EU_ENDPOINT
|
||||
)
|
||||
_apply_stored_langsmith_tracing()
|
||||
assert "LANGSMITH_ENDPOINT" not in os.environ
|
||||
assert "LANGCHAIN_ENDPOINT" not in os.environ
|
||||
|
||||
def test_matching_prefixed_key_allows_stored_endpoint(
|
||||
self,
|
||||
fake_state_dir: Path, # noqa: ARG002
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A scoped key with the stored value still uses the stored endpoint."""
|
||||
import os
|
||||
|
||||
from deepagents_code import auth_store
|
||||
|
||||
for var in ("LANGSMITH_ENDPOINT", "LANGCHAIN_ENDPOINT"):
|
||||
monkeypatch.delenv(var, raising=False)
|
||||
monkeypatch.setenv("DEEPAGENTS_CODE_LANGSMITH_API_KEY", "lsv2_stored")
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_stored", base_url=LANGSMITH_EU_ENDPOINT
|
||||
)
|
||||
_apply_stored_langsmith_tracing()
|
||||
assert os.environ["LANGSMITH_ENDPOINT"] == LANGSMITH_EU_ENDPOINT
|
||||
assert "LANGCHAIN_ENDPOINT" not in os.environ
|
||||
|
||||
|
||||
class TestNormalizeLangSmithEndpoint:
|
||||
"""Tests for normalize_langsmith_endpoint() and is_http_url()."""
|
||||
|
||||
def test_us_alias(self) -> None:
|
||||
"""`us` (any case) resolves to the canonical US endpoint."""
|
||||
assert normalize_langsmith_endpoint("us") == LANGSMITH_US_ENDPOINT
|
||||
assert normalize_langsmith_endpoint(" US ") == LANGSMITH_US_ENDPOINT
|
||||
|
||||
def test_eu_alias(self) -> None:
|
||||
"""`eu` (any case) resolves to the canonical EU endpoint."""
|
||||
assert normalize_langsmith_endpoint("eu") == LANGSMITH_EU_ENDPOINT
|
||||
assert normalize_langsmith_endpoint("Eu") == LANGSMITH_EU_ENDPOINT
|
||||
|
||||
def test_full_url_passthrough(self) -> None:
|
||||
"""A full URL is returned stripped and unchanged."""
|
||||
url = "https://langsmith.internal.example.com"
|
||||
assert normalize_langsmith_endpoint(f" {url} ") == url
|
||||
|
||||
def test_empty(self) -> None:
|
||||
"""Blank input returns an empty string."""
|
||||
assert normalize_langsmith_endpoint(" ") == ""
|
||||
|
||||
def test_is_http_url(self) -> None:
|
||||
"""Only http(s) URLs with a host pass validation."""
|
||||
assert is_http_url(LANGSMITH_EU_ENDPOINT) is True
|
||||
assert is_http_url("http://localhost:1984") is True
|
||||
assert is_http_url("ftp://example.com") is False
|
||||
assert is_http_url("not-a-url") is False
|
||||
assert is_http_url("https://") is False
|
||||
assert is_http_url("http://[::1") is False
|
||||
# A host with internal whitespace is malformed: rejecting it at save time
|
||||
# stops a wrong endpoint from being stored and silently dropping traces.
|
||||
assert is_http_url("https://exa mple.com") is False
|
||||
|
||||
|
||||
class TestGetTracingStatus:
|
||||
"""Tests for get_tracing_status()."""
|
||||
@@ -5333,8 +5546,12 @@ class TestLazyModuleAttributes:
|
||||
monkeypatch.setenv("DEEPAGENTS_CODE_LANGSMITH_TRACING", "true")
|
||||
monkeypatch.delenv("LANGSMITH_API_KEY", raising=False)
|
||||
monkeypatch.delenv("LANGSMITH_TRACING", raising=False)
|
||||
monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False)
|
||||
monkeypatch.delenv("LANGCHAIN_ENDPOINT", raising=False)
|
||||
monkeypatch.delenv("DEEPAGENTS_CODE_LANGSMITH_PROJECT", raising=False)
|
||||
auth_store.set_stored_key("langsmith", "lsv2_stored")
|
||||
auth_store.set_stored_key(
|
||||
"langsmith", "lsv2_stored", base_url=LANGSMITH_EU_ENDPOINT
|
||||
)
|
||||
|
||||
with (
|
||||
patch("deepagents_code.config._load_dotenv"),
|
||||
@@ -5347,6 +5564,8 @@ class TestLazyModuleAttributes:
|
||||
|
||||
assert os.environ["LANGSMITH_API_KEY"] == "lsv2_prefixed"
|
||||
assert os.environ["LANGSMITH_TRACING"] == "true"
|
||||
assert "LANGSMITH_ENDPOINT" not in os.environ
|
||||
assert "LANGCHAIN_ENDPOINT" not in os.environ
|
||||
finally:
|
||||
config_mod._bootstrap_state.done = original_done
|
||||
config_mod._bootstrap_state.original_langsmith_project = original_ls
|
||||
|
||||
Reference in New Issue
Block a user