feat(code): expand environment variables in MCP config (#4681)

Deep Agents Code now expands `${VAR}` and `${VAR:-default}` references
in `.mcp.json` server commands, arguments, environment values, URLs, and
headers.

---

Users who share `.mcp.json` files currently have to hard-code
machine-specific paths and credentials outside remote headers. This
makes otherwise portable MCP configuration fail for stdio servers and
remote URLs.

This resolves `${VAR}` and `${VAR:-default}` in `command`, `args`,
`env`, `url`, and `headers` when each server activates. Missing or
malformed references report the exact config field and disable only the
affected server. OAuth login follows the same behavior.

Expansion remains behind the existing project MCP trust checks, so
unapproved project configuration cannot read environment values or
initiate connections.

libs/code/deepagents_code/mcp_auth.py

@@ -48,6 +48,8 @@ from mcp.shared.auth import (
 from pydantic import BaseModel, ConfigDict, ValidationError
 from typing_extensions import override
 
+from deepagents_code.mcp_config import resolve_mcp_server_env
+
 if TYPE_CHECKING:
     from pathlib import Path
 
@@ -158,9 +160,6 @@ class _ExpectedReauthLogFilter(logging.Filter):
 
 logging.getLogger("mcp.client.auth.oauth2").addFilter(_ExpectedReauthLogFilter())
 
-_REF_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
-"""Matches `${VAR}` placeholders inside config strings for env-var substitution."""
-
 _SAFE_SERVER_NAME_RE = re.compile(r"^[A-Za-z0-9_-]+$")
 """Matches server names that are safe to embed in token-file basenames.
 
@@ -180,6 +179,35 @@ rejected as expired by the server — without the margin, a 401 sends the SDK
 into the full re-auth (browser) flow instead of the cheaper refresh grant.
 """
 
+
+def resolve_headers(
+    headers: dict[str, str],
+    *,
+    server_name: str | None = None,
+) -> dict[str, str]:
+    """Resolve environment-variable references in MCP header values.
+
+    This compatibility wrapper preserves the original public helper while
+    delegating interpolation and validation to the shared MCP config resolver.
+
+    Args:
+        headers: Raw header mapping from MCP config.
+        server_name: Optional server name for field-specific error messages.
+
+    Returns:
+        A new dictionary with environment-variable references resolved.
+
+    Raises:
+        TypeError: If a header value is not a string.
+        RuntimeError: If interpolation fails.
+    """  # noqa: DOC502 - `RuntimeError` is raised by the shared config resolver
+    resolved = resolve_mcp_server_env(
+        server_name or "<unknown>",
+        {"headers": headers},
+    )
+    return resolved["headers"]
+
+
 _REFRESH_LOCK_TIMEOUT_SECONDS = 60.0
 """Longest a provider waits for the cross-process token-refresh lock.
 
@@ -191,66 +219,6 @@ refresh token (see `_acquire_refresh_lock`).
 """
 
 
-def resolve_headers(
-    headers: dict[str, str],
-    *,
-    server_name: str | None = None,
-) -> dict[str, str]:
-    """Resolve `${VAR}` env-var references in header values.
-
-    Args:
-        headers: Raw header mapping from MCP config.
-        server_name: Optional server name for error messages.
-
-    Returns:
-        A new dict with env-var references resolved to current values.
-
-    Raises:
-        TypeError: If a header value is not a string.
-        RuntimeError: If a `${VAR}` reference points to an unset env var.
-    """  # noqa: DOC502 - RuntimeError is raised via `_interpolate`
-    resolved: dict[str, str] = {}
-    for name, value in headers.items():
-        if not isinstance(value, str):
-            where = f"mcpServers.{server_name}.headers.{name}" if server_name else name
-            msg = f"{where} must be a string, got {type(value).__name__}"
-            raise TypeError(msg)
-        resolved[name] = _interpolate(value, header=name, server_name=server_name)
-    return resolved
-
-
-def _interpolate(s: str, *, header: str, server_name: str | None) -> str:
-    """Expand `${VAR}` references in `s` against the current environment.
-
-    Args:
-        s: Raw header value.
-        header: Header name, used in error messages.
-        server_name: Owning server name for error messages.
-
-    Returns:
-        Interpolated string.
-
-    Raises:
-        RuntimeError: If a referenced env var is unset.
-    """  # noqa: DOC502 - raised inside the inner `replace` substitution callback
-
-    def replace(match: re.Match[str]) -> str:
-        var_name = match.group(1)
-        val = os.environ.get(var_name)
-        if val is None:
-            where = (
-                f"mcpServers.{server_name}.headers.{header}" if server_name else header
-            )
-            msg = (
-                f"{where} references unset env var {var_name}. "
-                f"Set {var_name} in the environment or remove the reference."
-            )
-            raise RuntimeError(msg)
-        return val
-
-    return _REF_RE.sub(replace, s)
-
-
 def _tokens_dir() -> Path:
     """Return `~/.deepagents/.state/mcp-tokens/`.
 
@@ -1739,6 +1707,14 @@ def format_login_failure(exc: BaseException) -> str:
     if reauth is not None:
         return str(reauth)
 
+    from deepagents_code.mcp_tools import MCPConfigError
+
+    if isinstance(exc, MCPConfigError):
+        # Config-interpolation errors are our own and are raised before the
+        # OAuth handshake, so they carry no token material and their
+        # field-scoped messages are safe (and useful) to render verbatim.
+        return str(exc)
+
     safe_types = (
         _LoopbackCallbackTimeoutError,
         _LoopbackCallbackUnavailableError,
@@ -1894,15 +1870,18 @@ async def login(
 
     Raises:
         ValueError: If `server_config` isn't an http/sse server.
-        RuntimeError: If header env-var interpolation fails, the device
-            flow fails or times out, or the OAuth handshake aborts.
-    """  # noqa: DOC502 - `RuntimeError` surfaces via `resolve_headers` / `_run_device_flow`
+        MCPConfigError: If config env-var interpolation fails or a
+            supported field has the wrong type (a non-string value, or
+            args/env/headers with the wrong container type).
+        RuntimeError: If the device flow fails or times out, or the
+            OAuth handshake aborts.
+    """  # noqa: DOC502 - `RuntimeError` surfaces via the device flow / handshake
     from langchain_mcp_adapters.sessions import (
         SSEConnection,
         StreamableHttpConnection,
     )
 
-    from deepagents_code.mcp_tools import _resolve_server_type
+    from deepagents_code.mcp_tools import MCPConfigError, _resolve_server_type
 
     # OAuth login is discovery-based (RFC 9728), so it works for any remote
     # http/sse server — whether the config opted in with `auth: oauth` or the
@@ -1915,14 +1894,22 @@ async def login(
             "OAuth login is only valid for http/sse."
         )
         raise ValueError(msg)
+    try:
+        resolved_config = resolve_mcp_server_env(server_name, server_config)
+    except (RuntimeError, TypeError) as exc:
+        # Re-raise as MCPConfigError (a ValueError) so callers' existing
+        # config-error handling catches it, and `format_login_failure`
+        # preserves the actionable, field-scoped message instead of
+        # collapsing it to a bare "RuntimeError"/"TypeError".
+        raise MCPConfigError(str(exc)) from exc
 
     from deepagents_code.mcp_providers import resolve_provider
 
-    storage = FileTokenStorage(server_name, server_url=server_config["url"])
-    policy = resolve_provider(server_config["url"])
+    storage = FileTokenStorage(server_name, server_url=resolved_config["url"])
+    policy = resolve_provider(resolved_config["url"])
     result = await policy.run_login(
         server_name=server_name,
-        server_url=server_config["url"],
+        server_url=resolved_config["url"],
         storage=storage,
         ui=ui,
     )
@@ -1937,7 +1924,7 @@ async def login(
 
     provider = build_oauth_provider(
         server_name=server_name,
-        server_url=server_config["url"],
+        server_url=resolved_config["url"],
         storage=storage,
         extra_auth_params=result.extra_auth_params or None,
         ui=ui,
@@ -1946,21 +1933,18 @@ async def login(
     if transport == "http":
         conn = StreamableHttpConnection(
             transport="streamable_http",
-            url=server_config["url"],
+            url=resolved_config["url"],
             auth=provider,
         )
     else:
         conn = SSEConnection(
             transport="sse",
-            url=server_config["url"],
+            url=resolved_config["url"],
             auth=provider,
         )
 
-    if "headers" in server_config:
-        conn["headers"] = resolve_headers(
-            server_config["headers"],
-            server_name=server_name,
-        )
+    if "headers" in resolved_config:
+        conn["headers"] = resolved_config["headers"]
 
     await _drive_handshake({server_name: conn})
     await ui.show_success(success_message)

libs/code/deepagents_code/mcp_config.py

@@ -0,0 +1,176 @@
+"""Validation and environment-variable expansion for MCP server config.
+
+Resolves `${VAR}` and `${VAR:-default}` references in the supported
+configuration fields (`command`, `url`, `args`, `env`, `headers`) and
+validates their types. A `${VAR:-default}` reference falls back to
+`default` when `VAR` is unset *or* empty (POSIX `:-` semantics).
+"""
+
+from __future__ import annotations
+
+import copy
+import os
+import re
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+_ENV_REF_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)(?::-([^{}]*))?\}")
+"""Matches a supported reference: `${VAR}` or `${VAR:-default}`.
+
+Group 1 is the variable name; group 2 (present only for the `:-` form) is
+the default. A bare `$VAR` and a literal `$` are intentionally not matched.
+"""
+
+_ENV_BRACE_RE = re.compile(r"\$\{")
+"""Matches a `${` brace-open, used to catch malformed `${...}` references."""
+
+
+def _interpolate_env(value: str, *, field: str) -> str:
+    """Expand `${VAR}` / `${VAR:-default}` references in one config string.
+
+    A bare `$VAR` (no braces) and a literal `$` pass through untouched;
+    only the braced forms expand. `${VAR:-default}` uses `default` when
+    `VAR` is unset or empty. A `${...}` that does not parse as one of the
+    supported forms (e.g. `${VAR-default}` or an unterminated `${VAR`) is
+    rejected rather than silently emitted, so a typo cannot inject a
+    garbage value into a URL, command, or header.
+
+    Args:
+        value: Raw configuration string.
+        field: Fully qualified field path for error messages.
+
+    Returns:
+        The interpolated string.
+
+    Raises:
+        RuntimeError: If a required environment variable is unset, or the
+            string contains a malformed `${...}` reference.
+    """
+
+    def replace(match: re.Match[str]) -> str:
+        name = match.group(1)
+        default = match.group(2)
+        resolved = os.environ.get(name)
+        # A non-empty value always wins, for both `${VAR}` and `${VAR:-default}`.
+        if resolved:
+            return resolved
+        # `resolved` is now "" (set but empty) or None (unset).
+        if default is not None:
+            # `${VAR:-default}`: `:-` falls back for empty *and* unset (POSIX).
+            return default
+        if resolved is not None:
+            # `${VAR}` set to "": no default, so emit the empty value.
+            return resolved
+        # `${VAR}` unset with no default: the only hard error.
+        msg = (
+            f"{field} references unset env var {name}. "
+            f"Set {name} in the environment or provide a default."
+        )
+        raise RuntimeError(msg)
+
+    # Reject any `${` that isn't the start of a well-formed reference. The
+    # check is against the raw `value` (not the substituted result) so a
+    # resolved value that happens to contain `${` never trips it.
+    ref_spans = [match.span() for match in _ENV_REF_RE.finditer(value)]
+    for brace in _ENV_BRACE_RE.finditer(value):
+        if not any(start <= brace.start() < end for start, end in ref_spans):
+            msg = (
+                f"{field} contains a malformed '${{...}}' reference. "
+                "Use '${VAR}' or '${VAR:-default}'."
+            )
+            raise RuntimeError(msg)
+
+    return _ENV_REF_RE.sub(replace, value)
+
+
+def _resolve_string(value: object, *, field: str) -> str:
+    """Validate and interpolate one string field.
+
+    Args:
+        value: Raw field value.
+        field: Fully qualified field path for error messages.
+
+    Returns:
+        The validated and interpolated string.
+
+    Raises:
+        TypeError: If the field value is not a string.
+    """
+    if not isinstance(value, str):
+        msg = f"{field} must be a string, got {type(value).__name__}"
+        raise TypeError(msg)
+    return _interpolate_env(value, field=field)
+
+
+def _resolve_mapping_values(
+    values: Mapping[str, object],
+    *,
+    field: str,
+) -> dict[str, str]:
+    """Validate and interpolate string values in a mapping field.
+
+    Args:
+        values: Raw mapping values.
+        field: Fully qualified field path for error messages.
+
+    Returns:
+        A new mapping with validated and interpolated values.
+    """
+    return {
+        name: _resolve_string(value, field=f"{field}.{name}")
+        for name, value in values.items()
+    }
+
+
+def resolve_mcp_server_env(
+    server_name: str,
+    server_config: Mapping[str, object],
+) -> dict[str, Any]:
+    """Resolve `${VAR}` references in one MCP server's supported fields.
+
+    Interpolates the `command`, `url`, `args`, `env`, and `headers`
+    fields (see `_interpolate_env` for the reference syntax); every other
+    field is copied through verbatim. The input is not mutated.
+
+    Args:
+        server_name: Server name used in field-specific error messages.
+        server_config: Raw server configuration.
+
+    Returns:
+        A resolved copy of the server configuration.
+
+    Raises:
+        TypeError: If a supported field has the wrong type — a non-string
+            scalar value, or `args`/`env`/`headers` with the wrong container
+            type.
+        RuntimeError: If a required environment variable is unset.
+    """  # noqa: DOC502 - `RuntimeError` is raised by `_interpolate_env`
+    resolved: dict[str, Any] = copy.deepcopy(dict(server_config))
+    prefix = f"mcpServers.{server_name}"
+
+    for name in ("command", "url"):
+        if name in resolved:
+            resolved[name] = _resolve_string(resolved[name], field=f"{prefix}.{name}")
+
+    if "args" in resolved:
+        args = resolved["args"]
+        if not isinstance(args, list):
+            msg = f"{prefix}.args must be a list, got {type(args).__name__}"
+            raise TypeError(msg)
+        resolved["args"] = [
+            _resolve_string(value, field=f"{prefix}.args[{index}]")
+            for index, value in enumerate(args)
+        ]
+
+    for name in ("env", "headers"):
+        if name not in resolved:
+            continue
+        values = resolved[name]
+        if not isinstance(values, dict):
+            msg = f"{prefix}.{name} must be a dictionary, got {type(values).__name__}"
+            raise TypeError(msg)
+        resolved[name] = _resolve_mapping_values(values, field=f"{prefix}.{name}")
+
+    return resolved

libs/code/deepagents_code/mcp_tools.py

@@ -21,6 +21,8 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast, overload
 
+from deepagents_code.mcp_config import resolve_mcp_server_env
+
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable, Mapping, Sequence
 
@@ -470,7 +472,7 @@ def _resolve_server_type(server_config: Mapping[str, Any]) -> str:
 def _validate_server_config(server_name: str, server_config: dict[str, Any]) -> None:
     """Validate a single server configuration.
 
-    Performs only shape checks — `${VAR}` header interpolation is deferred
+    Performs only shape checks — `${VAR}` config interpolation is deferred
     to activation time so one unset env var only fails its own server
     rather than hiding every other MCP entry in the same file.
 
@@ -849,7 +851,7 @@ def load_mcp_config(config_path: str) -> dict[str, Any]:
         json.JSONDecodeError: If config file contains invalid JSON.
         TypeError: If config fields have wrong types.
         ValueError: If config is missing required fields.
-    """  # noqa: DOC502 - raised indirectly by `_load_mcp_config_json` / `_validate_server_config` (which does shape-only checks; `${VAR}` header interpolation is deferred to activation time, so no RuntimeError here)
+    """  # noqa: DOC502 - raised indirectly by `_load_mcp_config_json` / `_validate_server_config` (which does shape-only checks; `${VAR}` config interpolation is deferred to activation time, so no RuntimeError here)
     config = _load_mcp_config_top_level(Path(config_path))
     _validate_mcp_config_servers(config)
 
@@ -1109,7 +1111,7 @@ def _check_stdio_server(server_name: str, server_config: dict[str, Any]) -> None
         raise RuntimeError(msg)
     if shutil.which(command) is None:
         msg = (
-            f"MCP server '{server_name}': command '{command}' not found on PATH. "
+            f"MCP server '{server_name}': configured command not found on PATH. "
             "Install it or check your MCP config."
         )
         raise RuntimeError(msg)
@@ -1135,19 +1137,49 @@ async def _check_remote_server(server_name: str, server_config: dict[str, Any])
         async with httpx.AsyncClient(timeout=2.0) as client:
             response = await client.head(url)
     except (httpx.HTTPError, httpx.InvalidURL, OSError) as exc:
+        # Name the failure *class* (e.g. `ConnectTimeout`, `InvalidURL`) so the
+        # failure mode stays diagnosable, but keep the URL redacted: `str(exc)`
+        # echoes the URL (which may carry `${VAR}`-injected credentials), while
+        # the class name never does.
         msg = (
-            f"MCP server '{server_name}': URL '{url}' is unreachable: {exc}. "
+            f"MCP server '{server_name}': configured URL is unreachable "
+            f"({type(exc).__name__}). "
             "Check that the URL is correct and the server is running."
         )
         raise RuntimeError(msg) from exc
     if response.status_code >= 500:  # noqa: PLR2004  # HTTP server-error band
         msg = (
-            f"MCP server '{server_name}': {url} returned HTTP "
+            f"MCP server '{server_name}': configured URL returned HTTP "
             f"{response.status_code}. Server may be down; retry later."
         )
         raise RuntimeError(msg)
 
 
+def _config_uses_env_interpolation(server_config: dict[str, Any]) -> bool:
+    """Return whether a supported config value contains an env reference.
+
+    Exceptions raised after interpolation may include resolved connection
+    values in their messages or traceback. Treat every environment-derived
+    value as potentially sensitive so those failures can be reported without
+    exposing the resolved value.
+
+    Args:
+        server_config: Raw, unresolved MCP server configuration.
+
+    Returns:
+        Whether a supported value contains a `${...}` reference.
+    """
+    scalar_values = [server_config.get("command"), server_config.get("url")]
+    sequence_values = server_config.get("args")
+    if isinstance(sequence_values, list):
+        scalar_values.extend(sequence_values)
+    for field in ("env", "headers"):
+        mapping = server_config.get(field)
+        if isinstance(mapping, dict):
+            scalar_values.extend(mapping.values())
+    return any(isinstance(value, str) and "${" in value for value in scalar_values)
+
+
 async def _discover_tools(session: ClientSession) -> list[Any]:
     """Enumerate MCP tools from `session`, paginating until exhausted.
 
@@ -1635,6 +1667,25 @@ async def _load_tools_from_config(
             ready `Connection` otherwise.
         """
         server_type = transports[server_name]
+        # Capture this from the *raw* config, before resolution below rebinds
+        # `server_config` to the expanded copy. Once `${...}` refs are expanded,
+        # a downstream setup error may echo the resolved (secret-bearing) value,
+        # so those messages are redacted; plain configs keep full detail.
+        redact_failure_details = _config_uses_env_interpolation(server_config)
+        # Config env-var resolution is the only step that raises `TypeError`
+        # (non-string field). Keep it in its own `try` so an unexpected
+        # `TypeError` from the connectivity checks below — whose contract is
+        # `RuntimeError` only — surfaces as a real bug instead of being
+        # relabeled as a per-server config skip.
+        try:
+            server_config = resolve_mcp_server_env(server_name, server_config)
+        except (RuntimeError, TypeError) as exc:
+            logger.warning(
+                "MCP server '%s' skipped: config error: %s",
+                server_name,
+                exc,
+            )
+            return ("error", str(exc))
         try:
             if server_type in _SUPPORTED_REMOTE_TYPES:
                 await _check_remote_server(server_name, server_config)
@@ -1664,12 +1715,7 @@ async def _load_tools_from_config(
                     )
 
                 if "headers" in server_config:
-                    from deepagents_code.mcp_auth import resolve_headers
-
-                    conn["headers"] = resolve_headers(
-                        server_config["headers"],
-                        server_name=server_name,
-                    )
+                    conn["headers"] = server_config["headers"]
 
                 from deepagents_code.mcp_auth import (
                     FileTokenStorage,
@@ -1719,12 +1765,25 @@ async def _load_tools_from_config(
                 transport="stdio",
             )
         except (OSError, RuntimeError, TypeError, ValueError) as exc:
-            logger.warning(
-                "MCP server '%s' skipped: config/setup failed: %s",
-                server_name,
-                exc,
-            )
-            return ("error", str(exc))
+            if redact_failure_details:
+                error = (
+                    f"MCP server {server_name!r}: setup failed after "
+                    "resolving environment variables."
+                )
+                logger.warning(
+                    "MCP server '%s' skipped: config/setup failed (%s; details "
+                    "redacted because config uses environment interpolation)",
+                    server_name,
+                    exc.__class__.__name__,
+                )
+            else:
+                error = str(exc)
+                logger.warning(
+                    "MCP server '%s' skipped: config/setup failed",
+                    server_name,
+                    exc_info=exc,
+                )
+            return ("error", error)
 
     # Preflight + connection build runs concurrently across servers (bounded).
     # Results come back in submission order, so `skipped`/`connections` are
@@ -1770,13 +1829,33 @@ async def _load_tools_from_config(
         Returns:
             The server's LangChain tools plus its `MCPServerInfo` entry.
         """  # noqa: DOC501 - CancelledError/KeyboardInterrupt/SystemExit are re-raised pass-throughs
+        redact_failure_details = _config_uses_env_interpolation(server_config)
+
+        def _log_caught_exception(
+            level: int,
+            message: str,
+            caught: BaseException,
+        ) -> None:
+            """Log a caught exception without exposing resolved config values."""
+            if redact_failure_details:
+                rendered_message = message % server_name
+                logger.log(
+                    level,
+                    "%s (%s; details redacted because config uses environment "
+                    "interpolation)",
+                    rendered_message,
+                    caught.__class__.__name__,
+                )
+            else:
+                logger.log(level, message, server_name, exc_info=caught)
+
         try:
             async with create_session(connections[server_name]) as discover_session:
                 await discover_session.initialize()
                 mcp_tools = await _discover_tools(discover_session)
         except (asyncio.CancelledError, KeyboardInterrupt, SystemExit):
             raise
-        except Exception as exc:
+        except Exception as exc:  # noqa: BLE001 - isolate third-party discovery failures per server
             from deepagents_code.mcp_auth import (
                 find_oauth_challenge,
                 find_reauth_required,
@@ -1790,17 +1869,17 @@ async def _load_tools_from_config(
                     if transport in _SUPPORTED_REMOTE_TYPES
                     else None
                 )
-            except Exception:
+            except Exception as classify_exc:  # noqa: BLE001 - classification must not abort other servers
                 # Classifying the failure is best-effort. If a classifier
                 # itself raises, degrade this one server to a plain error
                 # rather than letting the exception abort tool loading for
                 # every remaining server.
                 reauth = None
                 challenge_url = None
-                logger.debug(
+                _log_caught_exception(
+                    logging.DEBUG,
                     "MCP server '%s': failed to classify discovery error",
-                    server_name,
-                    exc_info=True,
+                    classify_exc,
                 )
 
             if reauth is not None:
@@ -1810,19 +1889,21 @@ async def _load_tools_from_config(
                 # re-login, and keep the original exception only in debug logs
                 # so expected re-auth skips don't flood non-interactive output.
                 status = "unauthenticated"
-                error = (
-                    f"{reauth} "
-                    "(token refresh failed; the original error is in debug logs)"
+                detail = (
+                    "details redacted because config uses environment interpolation"
+                    if redact_failure_details
+                    else "the original error is in debug logs"
                 )
+                error = f"{reauth} (token refresh failed; {detail})"
                 logger.warning(
                     "MCP server '%s' skipped: %s",
                     server_name,
                     error,
                 )
-                logger.debug(
+                _log_caught_exception(
+                    logging.DEBUG,
                     "MCP server '%s' skipped: tool discovery failed",
-                    server_name,
-                    exc_info=True,
+                    exc,
                 )
             elif challenge_url is not None:
                 # A remote server answered with a 401 OAuth challenge
@@ -1840,18 +1921,25 @@ async def _load_tools_from_config(
                     server_name,
                     error,
                 )
-                logger.debug(
+                _log_caught_exception(
+                    logging.DEBUG,
                     "MCP server '%s' skipped: 401 OAuth challenge detected",
-                    server_name,
-                    exc_info=True,
+                    exc,
                 )
             else:
                 status = "error"
-                error = str(exc)
-                logger.warning(
+                error = (
+                    (
+                        f"MCP server {server_name!r}: tool discovery failed "
+                        "after resolving environment variables."
+                    )
+                    if redact_failure_details
+                    else str(exc)
+                )
+                _log_caught_exception(
+                    logging.WARNING,
                     "MCP server '%s' skipped: tool discovery failed",
-                    server_name,
-                    exc_info=True,
+                    exc,
                 )
             return [], MCPServerInfo(
                 name=server_name,
@@ -1938,17 +2026,25 @@ async def _load_tools_from_config(
                 )
         except (asyncio.CancelledError, KeyboardInterrupt, SystemExit):
             raise
-        except Exception as exc:
-            logger.warning(
+        except Exception as exc:  # noqa: BLE001 - isolate third-party tool conversion failures per server
+            error = (
+                (
+                    f"MCP server {server_name!r}: tool construction failed "
+                    "after resolving environment variables."
+                )
+                if redact_failure_details
+                else str(exc)
+            )
+            _log_caught_exception(
+                logging.WARNING,
                 "MCP server '%s' skipped: tool construction failed",
-                server_name,
-                exc_info=True,
+                exc,
             )
             return [], MCPServerInfo(
                 name=server_name,
                 transport=transport,
                 status="error",
-                error=str(exc),
+                error=error,
             )
 
         return server_tools, MCPServerInfo(
@@ -2119,8 +2215,8 @@ async def resolve_and_load_mcp_tools(
             types.
         ValueError: If `explicit_config_path` is missing required fields
             or declares an unsupported transport.
-        RuntimeError: If the merged MCP config is malformed. (Header `${VAR}`
-            interpolation is deferred to activation inside
+        RuntimeError: If the merged MCP config is malformed. (`${VAR}`
+            config interpolation is deferred to activation inside
             `_load_tools_from_config`, which captures such failures into the
             returned `server_infos` rather than raising here.)
     """  # noqa: DOC502 - FileNotFoundError / JSONDecodeError / TypeError / ValueError surface via `load_mcp_config`
@@ -2217,7 +2313,7 @@ async def resolve_and_load_mcp_tools(
         # here (rather than loading all or none) preserves the SSRF/header-
         # exfiltration gate: a non-allowlisted remote entry from an attacker-
         # controlled .mcp.json never reaches the preflight HEAD probe or the
-        # `${VAR}` header interpolation during the discovery handshake.
+        # `${VAR}` config interpolation during the discovery handshake.
         servers = config["mcpServers"]
         kept: dict[str, Any] = {}
         for name, server in servers.items():

libs/code/tests/unit_tests/test_mcp_auth.py

@@ -30,6 +30,34 @@ _BEARER_CHALLENGE = f'Bearer resource_metadata="{_RESOURCE_METADATA_URL}"'
 """A minimal RFC 9728 Bearer challenge pointing at the resource metadata."""
 
 
+class TestResolveHeaders:
+    """Compatibility coverage for the public header resolver."""
+
+    def test_delegates_to_shared_interpolation(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """The legacy helper remains importable and supports shared syntax."""
+        monkeypatch.delenv("MCP_HEADER_TOKEN", raising=False)
+
+        resolved = resolve_headers(
+            {"Authorization": "Bearer ${MCP_HEADER_TOKEN:-fallback}"},
+            server_name="remote",
+        )
+
+        assert resolved == {"Authorization": "Bearer fallback"}
+
+    def test_preserves_original_mapping(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Compatibility resolution returns a copy without mutating its input."""
+        monkeypatch.setenv("MCP_HEADER_TOKEN", "resolved")
+        headers = {"Authorization": "Bearer ${MCP_HEADER_TOKEN}"}
+
+        resolved = resolve_headers(headers)
+
+        assert resolved == {"Authorization": "Bearer resolved"}
+        assert headers == {"Authorization": "Bearer ${MCP_HEADER_TOKEN}"}
+
+
 def _http_status_error(
     status_code: int,
     *,
@@ -59,37 +87,6 @@ def fake_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
     return fake
 
 
-class TestResolveHeaders:
-    """Tests for static MCP header interpolation."""
-
-    def test_resolves_single_env_var(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        """A single `${VAR}` placeholder resolves to its env value."""
-        monkeypatch.setenv("FOO", "bar")
-        assert resolve_headers({"Authorization": "Bearer ${FOO}"}) == {
-            "Authorization": "Bearer bar"
-        }
-
-    def test_resolves_multiple_env_vars(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        """Multiple placeholders resolve left-to-right."""
-        monkeypatch.setenv("A", "alpha")
-        monkeypatch.setenv("B", "beta")
-        assert resolve_headers({"X-Combo": "${A}-${B}"}) == {"X-Combo": "alpha-beta"}
-
-    def test_non_string_value_raises(self) -> None:
-        """Header values must be strings."""
-        with pytest.raises(TypeError, match="must be a string"):
-            resolve_headers({"X-Bad": 123}, server_name="srv")  # ty: ignore
-
-    def test_unset_env_var_raises(self) -> None:
-        """Unset placeholders fail with a helpful message."""
-        with pytest.raises(RuntimeError, match="unset env var"):
-            resolve_headers({"Authorization": "Bearer ${MISSING}"})
-
-    def test_plain_text_value_is_unchanged(self) -> None:
-        """Strings without placeholders pass through unchanged."""
-        assert resolve_headers({"X-Plain": "hello"}) == {"X-Plain": "hello"}
-
-
 def _make_tokens(access_token: str = "at"):
     return OAuthToken(
         access_token=access_token,
@@ -1397,6 +1394,19 @@ class TestFormatLoginFailure:
         assert sentinel not in summary
         assert "FakeMcpError" in summary
 
+    def test_preserves_message_for_config_errors(self) -> None:
+        """Config errors are pre-handshake and token-free, so keep the message.
+
+        These carry the actionable field path (e.g. which var is unset);
+        collapsing them to a bare class name would strip the only guidance
+        the user has for fixing their `.mcp.json`.
+        """
+        from deepagents_code.mcp_tools import MCPConfigError
+
+        message = "mcpServers.notion.url references unset env var MCP_GATEWAY_HOST."
+        summary = format_login_failure(MCPConfigError(message))
+        assert summary == message
+
     def test_includes_message_for_known_loopback_errors(self) -> None:
         """Loopback-internal exceptions are token-free and may include their message."""
         from deepagents_code.mcp_auth import _LoopbackCallbackTimeoutError
@@ -2637,6 +2647,7 @@ class TestLogin:
         """Configured static headers flow into the OAuth handshake connection."""
         from deepagents_code.mcp_auth import login
 
+        monkeypatch.setenv("MCP_GATEWAY_HOST", "mcp.notion.com")
         monkeypatch.setenv("MCP_GATEWAY_TOKEN", "gw-token")
         captured: dict[str, Any] = {}
 
@@ -2651,7 +2662,7 @@ class TestLogin:
                 server_name="notion",
                 server_config={
                     "transport": "http",
-                    "url": "https://mcp.notion.com/mcp",
+                    "url": "https://${MCP_GATEWAY_HOST}/mcp",
                     "auth": "oauth",
                     "headers": {
                         "X-Tenant": "acme",
@@ -2661,6 +2672,7 @@ class TestLogin:
                 ui=CliOAuthInteraction(),
             )
 
+        assert captured["url"] == "https://mcp.notion.com/mcp"
         assert captured["headers"] == {
             "X-Tenant": "acme",
             "Authorization": "Bearer gw-token",
@@ -2670,8 +2682,9 @@ class TestLogin:
         """Unset env vars in static headers fail before the handshake."""
         from deepagents_code.mcp_auth import login
         from deepagents_code.mcp_oauth_ui import CliOAuthInteraction
+        from deepagents_code.mcp_tools import MCPConfigError
 
-        with pytest.raises(RuntimeError, match="unset env var"):
+        with pytest.raises(MCPConfigError, match="unset env var"):
             await login(
                 server_name="notion",
                 server_config={
@@ -2683,6 +2696,49 @@ class TestLogin:
                 ui=CliOAuthInteraction(),
             )
 
+    async def test_login_unset_env_var_in_url_raises_config_error(self) -> None:
+        """An unset var in a non-header field fails with its field path."""
+        from deepagents_code.mcp_auth import login
+        from deepagents_code.mcp_oauth_ui import CliOAuthInteraction
+        from deepagents_code.mcp_tools import MCPConfigError
+
+        with pytest.raises(MCPConfigError, match=r"mcpServers\.notion\.url"):
+            await login(
+                server_name="notion",
+                server_config={
+                    "transport": "http",
+                    "url": "https://${MISSING_HOST}/mcp",
+                    "auth": "oauth",
+                },
+                ui=CliOAuthInteraction(),
+            )
+
+    async def test_login_non_string_field_raises_config_error(self) -> None:
+        """A non-string supported field is wrapped as `MCPConfigError` too.
+
+        Exercises the `TypeError` arm of `login()`'s resolution wrapper (the
+        unset-var tests only cover the `RuntimeError` arm).
+        """
+        from deepagents_code.mcp_auth import login
+        from deepagents_code.mcp_oauth_ui import CliOAuthInteraction
+        from deepagents_code.mcp_tools import MCPConfigError
+
+        # Deliberately malformed (non-string header value) to hit the
+        # `TypeError` arm; typed separately so the intent is explicit.
+        bad_config: dict[str, Any] = {
+            "transport": "http",
+            "url": "https://mcp.example.com/mcp",
+            "auth": "oauth",
+            "headers": {"X-Bad": 1},
+        }
+
+        with pytest.raises(MCPConfigError, match=r"mcpServers\.notion\.headers\.X-Bad"):
+            await login(
+                server_name="notion",
+                server_config=bad_config,  # ty: ignore
+                ui=CliOAuthInteraction(),
+            )
+
     async def test_github_login_runs_device_flow_and_seeds_client(self) -> None:
         """GitHub URLs short-circuit to device flow and persist client info."""
         from mcp.shared.auth import OAuthToken

libs/code/tests/unit_tests/test_mcp_config.py

@@ -0,0 +1,214 @@
+"""Tests for MCP configuration environment-variable expansion."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from deepagents_code.mcp_config import resolve_mcp_server_env
+
+
+class TestResolveMcpServerEnv:
+    """Tests for supported `.mcp.json` interpolation fields."""
+
+    def test_resolves_stdio_fields_without_mutating_source(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """`command`, `args`, and `env` resolve in a copied config."""
+        monkeypatch.setenv("MCP_HOME", "/opt/mcp")
+        monkeypatch.setenv("MCP_TOKEN", "secret")
+        monkeypatch.delenv("MCP_CACHE", raising=False)
+        config: dict[str, Any] = {
+            "command": "${MCP_HOME}/bin/server",
+            "args": ["--root", "${MCP_HOME}", "${MCP_CACHE:-/tmp/cache}"],
+            "env": {
+                "TOKEN": "prefix-${MCP_TOKEN}",
+                "CACHE": "${MCP_CACHE:-/tmp/cache}",
+            },
+        }
+
+        resolved = resolve_mcp_server_env("discourse", config)
+
+        assert resolved == {
+            "command": "/opt/mcp/bin/server",
+            "args": ["--root", "/opt/mcp", "/tmp/cache"],
+            "env": {"TOKEN": "prefix-secret", "CACHE": "/tmp/cache"},
+        }
+        assert config["command"] == "${MCP_HOME}/bin/server"
+        assert config["env"]["TOKEN"] == "prefix-${MCP_TOKEN}"
+
+    def test_resolves_remote_url_and_headers(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """`url` and header values resolve multiple references and defaults."""
+        monkeypatch.setenv("MCP_HOST", "mcp.example.com")
+        monkeypatch.setenv("MCP_TOKEN", "token")
+        monkeypatch.delenv("MCP_SCHEME", raising=False)
+
+        resolved = resolve_mcp_server_env(
+            "remote",
+            {
+                "url": "${MCP_SCHEME:-https}://${MCP_HOST}/mcp",
+                "headers": {
+                    "Authorization": "Bearer ${MCP_TOKEN}",
+                    "X-Origin": "${MCP_SCHEME:-https}-${MCP_HOST}",
+                },
+            },
+        )
+
+        assert resolved["url"] == "https://mcp.example.com/mcp"
+        assert resolved["headers"] == {
+            "Authorization": "Bearer token",
+            "X-Origin": "https-mcp.example.com",
+        }
+
+    def test_empty_variable_uses_default(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """The `:-` form uses its default for an empty variable."""
+        monkeypatch.setenv("MCP_SCHEME", "")
+
+        resolved = resolve_mcp_server_env(
+            "remote",
+            {"url": "${MCP_SCHEME:-https}://example.com"},
+        )
+
+        assert resolved["url"] == "https://example.com"
+
+    def test_empty_default_yields_empty_string(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """`${VAR:-}` (empty default) resolves to `""` for an unset var."""
+        monkeypatch.delenv("MCP_OPT", raising=False)
+
+        resolved = resolve_mcp_server_env(
+            "srv",
+            {"command": "node", "env": {"OPT": "${MCP_OPT:-}"}},
+        )
+
+        assert resolved["env"] == {"OPT": ""}
+
+    def test_bare_reference_set_empty_emits_empty_without_error(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """A bare `${VAR}` set to `""` emits the empty value, not an error.
+
+        Distinct from the unset case (which raises): `:-`-less refs only hard
+        error when the variable is *absent*, not when it is set-but-empty.
+        """
+        monkeypatch.setenv("MCP_EMPTY", "")
+
+        resolved = resolve_mcp_server_env("srv", {"command": "${MCP_EMPTY}/x"})
+
+        assert resolved["command"] == "/x"
+
+    def test_resolved_value_containing_brace_is_not_rescanned(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """A resolved value that itself contains `${` is emitted verbatim.
+
+        The malformed-reference guard runs against the raw config string, so a
+        substituted value that happens to contain `${...}` neither re-expands
+        nor trips the malformed check.
+        """
+        monkeypatch.setenv("MCP_LITERAL", "keep-${NOT_A_REF}-literal")
+
+        resolved = resolve_mcp_server_env("srv", {"command": "${MCP_LITERAL}"})
+
+        assert resolved["command"] == "keep-${NOT_A_REF}-literal"
+
+    def test_remote_fields_do_not_mutate_source(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """Resolving `url`/`headers` leaves the source config (and its dicts) intact."""
+        monkeypatch.setenv("MCP_TOKEN", "token")
+        config: dict[str, Any] = {
+            "url": "https://example.com",
+            "headers": {"Authorization": "Bearer ${MCP_TOKEN}"},
+        }
+
+        resolved = resolve_mcp_server_env("remote", config)
+
+        assert resolved["headers"] == {"Authorization": "Bearer token"}
+        assert config["headers"] == {"Authorization": "Bearer ${MCP_TOKEN}"}
+
+    def test_plain_dollar_and_unsupported_fields_are_unchanged(self) -> None:
+        """Only braced references in the supported field allowlist expand."""
+        config = {
+            "command": "$HOME/bin/server",
+            "allowedTools": ["${SHOULD_NOT_EXPAND}"],
+        }
+
+        resolved = resolve_mcp_server_env("srv", config)
+
+        assert resolved == config
+
+    def test_unset_variable_reports_exact_field_path(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """Missing required variables identify the server and field."""
+        monkeypatch.delenv("MISSING_MCP_PATH", raising=False)
+
+        with pytest.raises(
+            RuntimeError,
+            match=r"mcpServers\.srv\.args\[1\].*MISSING_MCP_PATH",
+        ):
+            resolve_mcp_server_env(
+                "srv",
+                {"command": "node", "args": ["--root", "${MISSING_MCP_PATH}"]},
+            )
+
+    def test_non_string_supported_value_reports_exact_field_path(self) -> None:
+        """Malformed supported values fail with a field-specific error."""
+        with pytest.raises(TypeError, match=r"mcpServers\.srv\.env\.PORT"):
+            resolve_mcp_server_env("srv", {"command": "node", "env": {"PORT": 1}})
+
+    def test_non_string_args_element_reports_indexed_field_path(self) -> None:
+        """A non-string element inside `args` names its index."""
+        with pytest.raises(TypeError, match=r"mcpServers\.srv\.args\[0\]"):
+            resolve_mcp_server_env("srv", {"command": "node", "args": [1]})
+
+    def test_non_string_header_value_reports_field_path(self) -> None:
+        """A non-string header value names the offending header."""
+        with pytest.raises(TypeError, match=r"mcpServers\.srv\.headers\.X-Bad"):
+            resolve_mcp_server_env(
+                "srv",
+                {"url": "https://x", "headers": {"X-Bad": 1}},
+            )
+
+    def test_args_not_a_list_reports_field_path(self) -> None:
+        """`args` must be a list, not a bare string."""
+        with pytest.raises(TypeError, match=r"mcpServers\.srv\.args must be a list"):
+            resolve_mcp_server_env("srv", {"command": "node", "args": "solo"})
+
+    def test_mapping_field_not_a_dict_reports_field_path(self) -> None:
+        """`env`/`headers` must be dictionaries."""
+        with pytest.raises(
+            TypeError,
+            match=r"mcpServers\.srv\.headers must be a dictionary",
+        ):
+            resolve_mcp_server_env("srv", {"url": "https://x", "headers": ["nope"]})
+
+    @pytest.mark.parametrize(
+        "value",
+        [
+            "${VAR-default}",
+            "${VAR:default}",
+            "prefix-${VAR",
+            "${A:-foo${BAD}",
+            "${A:-${B}}",
+        ],
+    )
+    def test_malformed_reference_is_rejected(self, value: str) -> None:
+        """An unparseable `${...}` fails instead of being emitted verbatim."""
+        with pytest.raises(RuntimeError, match=r"malformed"):
+            resolve_mcp_server_env("srv", {"command": value})

libs/code/tests/unit_tests/test_mcp_tools.py

@@ -1065,27 +1065,117 @@ class TestGetMCPTools:
         assert manager is not None
         await manager.cleanup()
 
-    async def test_remote_headers_are_resolved_and_passed(
+    async def test_remote_url_and_headers_are_resolved_and_passed(
         self,
         monkeypatch: pytest.MonkeyPatch,
         fake_create_session: tuple[AsyncMock, list[dict[str, Any]]],
     ) -> None:
-        """Resolved static headers are attached to remote connections."""
+        """Resolved URLs and static headers reach remote connections."""
+        monkeypatch.setenv("DA_MCP_HOST", "mcp.linear.app")
         monkeypatch.setenv("DA_TOKEN", "tok-123")
         _session, recorded = fake_create_session
         config = {
             "mcpServers": {
                 "linear": {
                     "transport": "http",
-                    "url": "https://mcp.linear.app/mcp",
+                    "url": "https://${DA_MCP_HOST}/mcp",
                     "headers": {"Authorization": "Bearer ${DA_TOKEN}"},
                 }
             }
         }
 
         await _load_tools_from_config(config)
+        assert recorded[0]["url"] == "https://mcp.linear.app/mcp"
         assert recorded[0]["headers"] == {"Authorization": "Bearer tok-123"}
 
+    async def test_stdio_fields_resolve_before_preflight_and_connection(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        fake_create_session: tuple[AsyncMock, list[dict[str, Any]]],
+    ) -> None:
+        """Stdio preflight and connection creation use resolved values."""
+        monkeypatch.setenv("DA_MCP_HOME", "/opt/mcp")
+        monkeypatch.setenv("DA_MCP_TOKEN", "token")
+        _session, recorded = fake_create_session
+        checked: list[dict[str, Any]] = []
+        config = {
+            "mcpServers": {
+                "srv": {
+                    "command": "${DA_MCP_HOME}/server",
+                    "args": ["--root", "${DA_MCP_HOME}"],
+                    "env": {"TOKEN": "${DA_MCP_TOKEN}"},
+                }
+            }
+        }
+
+        with patch(
+            "deepagents_code.mcp_tools._check_stdio_server",
+            side_effect=lambda _name, server: checked.append(server),
+        ):
+            await _load_tools_from_config(config)
+
+        assert checked[0]["command"] == "/opt/mcp/server"
+        assert checked[0]["args"] == ["--root", "/opt/mcp"]
+        assert recorded[0] == {
+            "command": "/opt/mcp/server",
+            "args": ["--root", "/opt/mcp"],
+            "env": {"TOKEN": "token"},
+            "transport": "stdio",
+        }
+
+    async def test_unset_variable_skips_only_affected_server(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        fake_create_session: tuple[AsyncMock, list[dict[str, Any]]],
+    ) -> None:
+        """An unresolved field does not prevent sibling servers from loading."""
+        monkeypatch.delenv("MISSING_DA_MCP_PATH", raising=False)
+        _session, recorded = fake_create_session
+        config = {
+            "mcpServers": {
+                "broken": {
+                    "command": "node",
+                    "args": ["${MISSING_DA_MCP_PATH}"],
+                },
+                "working": {"command": "node", "args": ["server.js"]},
+            }
+        }
+
+        _tools, manager, infos = await _load_tools_from_config(config)
+
+        assert [info.name for info in infos] == ["broken", "working"]
+        assert infos[0].status == "error"
+        assert "mcpServers.broken.args[0]" in (infos[0].error or "")
+        assert infos[1].status == "ok"
+        assert [connection["args"] for connection in recorded] == [["server.js"]]
+        assert manager is not None
+        await manager.cleanup()
+
+    async def test_non_string_field_skips_only_affected_server(
+        self,
+        fake_create_session: tuple[AsyncMock, list[dict[str, Any]]],
+    ) -> None:
+        """A `TypeError` from resolution skips its server, not its siblings."""
+        _session, recorded = fake_create_session
+        config = {
+            "mcpServers": {
+                # `env` is a dict (passes shape validation) but its value is
+                # not a string, so resolution raises `TypeError`.
+                "broken": {"command": "node", "env": {"PORT": 1}},
+                "working": {"command": "node", "args": ["server.js"]},
+            }
+        }
+
+        _tools, manager, infos = await _load_tools_from_config(config)
+
+        assert [info.name for info in infos] == ["broken", "working"]
+        assert infos[0].status == "error"
+        assert "mcpServers.broken.env.PORT" in (infos[0].error or "")
+        assert infos[1].status == "ok"
+        assert [connection["args"] for connection in recorded] == [["server.js"]]
+        assert manager is not None
+        await manager.cleanup()
+
     async def test_empty_env_is_coerced_to_none(
         self,
         fake_create_session: tuple[AsyncMock, list[dict[str, Any]]],
@@ -2126,7 +2216,7 @@ class TestHealthChecks:
             assert server_infos[0].name == "srv"
             assert server_infos[0].status == "error"
             error = server_infos[0].error or ""
-            assert "command 'missing' not found on PATH" in error
+            assert "configured command not found on PATH" in error
             assert manager is not None
         finally:
             if manager is not None:
@@ -2147,6 +2237,171 @@ class TestHealthChecks:
         ):
             await _check_remote_server("srv", {"url": "http://down:9999"})
 
+    async def test_expanded_url_is_redacted_from_preflight_error(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """Expanded URL credentials never reach status or warning text."""
+        secret = "url-token-must-not-leak"
+        monkeypatch.setenv("MCP_TOKEN", secret)
+        client = AsyncMock()
+        client.head.side_effect = httpx.InvalidURL(f"invalid URL containing {secret}")
+        client.__aenter__ = AsyncMock(return_value=client)
+        client.__aexit__ = AsyncMock(return_value=False)
+        caplog.set_level(logging.WARNING, logger="deepagents_code.mcp_tools")
+
+        with patch("httpx.AsyncClient", return_value=client):
+            tools, manager, infos = await _load_tools_from_config(
+                {
+                    "mcpServers": {
+                        "remote": {
+                            "transport": "http",
+                            "url": "not a url?token=${MCP_TOKEN}",
+                        }
+                    }
+                }
+            )
+
+        assert tools == []
+        assert infos[0].status == "error"
+        assert "configured URL is unreachable" in (infos[0].error or "")
+        # The failure *class* is surfaced for diagnosability; it never
+        # embeds the URL, so it is safe to include even when redacting.
+        assert "InvalidURL" in (infos[0].error or "")
+        assert secret not in (infos[0].error or "")
+        assert secret not in caplog.text
+        assert manager is not None
+        await manager.cleanup()
+
+    async def test_expanded_url_is_redacted_from_discovery_error(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """Resolved URL credentials never reach discovery status or debug logs."""
+        secret = "discovery-url-token-must-not-leak"
+        monkeypatch.setenv("MCP_TOKEN", secret)
+        request = httpx.Request(
+            "POST",
+            f"https://mcp.example.com/mcp?token={secret}",
+        )
+        response = httpx.Response(500, request=request)
+        discovery_error = httpx.HTTPStatusError(
+            f"server error for URL {request.url}",
+            request=request,
+            response=response,
+        )
+
+        @asynccontextmanager
+        async def _fail_discovery(
+            _connection: dict[str, Any],
+            *,
+            _mcp_callbacks: object | None = None,
+        ) -> AsyncIterator[None]:
+            raise discovery_error
+            yield
+
+        caplog.set_level(logging.DEBUG, logger="deepagents_code.mcp_tools")
+        with (
+            patch(
+                "deepagents_code.mcp_tools._check_remote_server",
+                new_callable=AsyncMock,
+            ),
+            patch(
+                "langchain_mcp_adapters.sessions.create_session",
+                _fail_discovery,
+            ),
+        ):
+            tools, manager, infos = await _load_tools_from_config(
+                {
+                    "mcpServers": {
+                        "remote": {
+                            "transport": "http",
+                            "url": ("https://mcp.example.com/mcp?token=${MCP_TOKEN}"),
+                        }
+                    }
+                }
+            )
+
+        assert tools == []
+        assert infos[0].status == "error"
+        assert "tool discovery failed" in (infos[0].error or "")
+        assert secret not in (infos[0].error or "")
+        assert secret not in caplog.text
+        assert manager is not None
+        await manager.cleanup()
+
+    async def test_expanded_value_is_redacted_from_connection_build_error(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """Resolved values never reach the connection-build error path either.
+
+        Covers the `_preflight_and_connect` setup catch (token-store / provider
+        construction), which runs after preflight succeeds and after `${...}`
+        refs are expanded.
+        """
+        secret = "build-token-must-not-leak"
+        monkeypatch.setenv("MCP_TOKEN", secret)
+        storage = MagicMock()
+        storage.get_tokens = AsyncMock(
+            side_effect=RuntimeError(f"token store failure for {secret}")
+        )
+        caplog.set_level(logging.DEBUG, logger="deepagents_code.mcp_tools")
+
+        with (
+            patch(
+                "deepagents_code.mcp_tools._check_remote_server",
+                new_callable=AsyncMock,
+            ),
+            patch("deepagents_code.mcp_auth.FileTokenStorage", return_value=storage),
+        ):
+            tools, manager, infos = await _load_tools_from_config(
+                {
+                    "mcpServers": {
+                        "remote": {
+                            "transport": "http",
+                            "url": "https://mcp.example.com/mcp?token=${MCP_TOKEN}",
+                        }
+                    }
+                }
+            )
+
+        assert tools == []
+        assert infos[0].status == "error"
+        assert "setup failed after resolving environment variables" in (
+            infos[0].error or ""
+        )
+        assert secret not in (infos[0].error or "")
+        assert secret not in caplog.text
+        assert manager is not None
+        await manager.cleanup()
+
+    async def test_expanded_command_is_redacted_from_preflight_error(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """Expanded commands never reach status or warning text."""
+        secret = "command-token-must-not-leak"
+        monkeypatch.setenv("MCP_COMMAND", secret)
+        caplog.set_level(logging.WARNING, logger="deepagents_code.mcp_tools")
+
+        with patch("deepagents_code.mcp_tools.shutil.which", return_value=None):
+            tools, manager, infos = await _load_tools_from_config(
+                {"mcpServers": {"stdio": {"command": "${MCP_COMMAND}"}}}
+            )
+
+        assert tools == []
+        assert infos[0].status == "error"
+        assert "configured command not found on PATH" in (infos[0].error or "")
+        assert secret not in (infos[0].error or "")
+        assert secret not in caplog.text
+        assert manager is not None
+        await manager.cleanup()
+
 
 class TestToolOrdering:
     """Tools are sorted deterministically by final name."""