feat: add MCP mode flags to meta and telemetry modules, auto-register .envrc files as secret managers

airbyte/_util/meta.py

@@ -17,6 +17,9 @@
 import requests
 
 
+_MCP_MODE_ENABLED: bool = False
+"""Whether we are running in MCP (Model Context Protocol) mode."""
+
 COLAB_SESSION_URL = "http://172.28.0.12:9000/api/sessions"
 """URL to get the current Google Colab session information."""
 
@@ -28,10 +31,30 @@ def get_colab_release_version() -> str | None:
     return None
 
 
+@lru_cache
 def is_ci() -> bool:
     return "CI" in os.environ
 
 
+def set_mcp_mode() -> None:
+    """Set flag indicating we are running in MCP (Model Context Protocol) mode.
+
+    This should be called early in MCP server initialization to ensure
+    proper detection and prevent interactive prompts.
+    """
+    global _MCP_MODE_ENABLED
+    _MCP_MODE_ENABLED = True
+
+
+def is_mcp_mode() -> bool:
+    """Return True if running in MCP (Model Context Protocol) mode."""
+    if _MCP_MODE_ENABLED:
+        return True
+
+    script_name = get_python_script_name()
+    return bool(script_name and "airbyte-mcp" in script_name)
+
+
 @lru_cache
 def is_langchain() -> bool:
     """Return True if running in a Langchain environment.
@@ -55,15 +78,16 @@ def is_colab() -> bool:
     return bool(get_colab_release_version())
 
 
-@lru_cache
 def is_interactive() -> bool:
+    """Return True if running in an interactive environment where we can prompt users for input."""
     try:
         if is_colab() or is_jupyter():
             return True
 
-        if is_ci():
+        if is_ci() or is_mcp_mode():
             return False
 
+        # No special modes detected. Return result based on whether stdin and stdout are ttys.
         return bool(
             sys.__stdin__ and sys.__stdin__.isatty() and sys.__stdout__ and sys.__stdout__.isatty()
         )

airbyte/_util/telemetry.py

@@ -184,6 +184,7 @@ def get_env_flags() -> dict[str, Any]:
     flags: dict[str, bool | str] = {
         "CI": meta.is_ci(),
         "LANGCHAIN": meta.is_langchain(),
+        "MCP": meta.is_mcp_mode(),
         "NOTEBOOK_RUNTIME": (
             "GOOGLE_COLAB"
             if meta.is_colab()

airbyte/mcp/_util.py

@@ -9,7 +9,14 @@
 import dotenv
 import yaml
 
-from airbyte.secrets import DotenvSecretManager, GoogleGSMSecretManager, register_secret_manager
+from airbyte._util.meta import is_interactive
+from airbyte.secrets import (
+    DotenvSecretManager,
+    GoogleGSMSecretManager,
+    SecretSourceEnum,
+    register_secret_manager,
+)
+from airbyte.secrets.config import disable_secret_source
 from airbyte.secrets.hydration import deep_update, detect_hardcoded_secrets
 from airbyte.secrets.util import get_secret, is_secret_available
 
@@ -28,8 +35,19 @@ def _load_dotenv_file(dotenv_path: Path | str) -> None:
 
 
 def initialize_secrets() -> None:
-    """Initialize dotenv to load environment variables from .env files."""
+    """Initialize dotenv to load environment variables from .env files.
+
+    Note: Later secret manager registrations have higher priority than earlier ones.
+    """
     # Load the .env file from the current working directory.
+    envrc_path = Path.cwd() / ".envrc"
+    if envrc_path.exists():
+        envrc_secret_mgr = DotenvSecretManager(envrc_path)
+        _load_dotenv_file(envrc_path)
+        register_secret_manager(
+            envrc_secret_mgr,
+        )
+
     if AIRBYTE_MCP_DOTENV_PATH_ENVVAR in os.environ:
         dotenv_path = Path(os.environ[AIRBYTE_MCP_DOTENV_PATH_ENVVAR]).absolute()
         custom_dotenv_secret_mgr = DotenvSecretManager(dotenv_path)
@@ -47,6 +65,10 @@ def initialize_secrets() -> None:
             )
         )
 
+    # Make sure we disable the prompt source in non-interactive environments.
+    if not is_interactive():
+        disable_secret_source(SecretSourceEnum.PROMPT)
+
 
 def resolve_config(  # noqa: PLR0912
     config: dict | str | None = None,

airbyte/mcp/server.py

@@ -6,14 +6,13 @@
 
 from fastmcp import FastMCP
 
+from airbyte._util.meta import set_mcp_mode
 from airbyte.mcp._cloud_ops import register_cloud_ops_tools
 from airbyte.mcp._connector_registry import register_connector_registry_tools
 from airbyte.mcp._local_ops import register_local_ops_tools
 from airbyte.mcp._util import initialize_secrets
 
 
-initialize_secrets()
-
 app: FastMCP = FastMCP("airbyte-mcp")
 register_connector_registry_tools(app)
 register_local_ops_tools(app)
@@ -23,6 +22,8 @@
 def main() -> None:
     """Main entry point for the MCP server."""
     print("Starting Airbyte MCP server.", file=sys.stderr)
+    set_mcp_mode()
+    initialize_secrets()
     try:
         asyncio.run(app.run_stdio_async())
     except KeyboardInterrupt:

tests/unit_tests/test_progress.py

@@ -161,6 +161,9 @@ def test_ci_environment_a_progress_style(monkeypatch):
 def test_ci_environment_b_progress_style(monkeypatch):
     """Test the style in a CI environment."""
     monkeypatch.setenv("CI", "1")
+    from airbyte._util import meta
+
+    meta.is_ci.cache_clear()
     progress = ProgressTracker(source=None, cache=None, destination=None)
     assert progress.style == ProgressStyle.PLAIN