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,33 @@ 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()
+    if script_name and "airbyte-mcp" in script_name:
+        return True
+
+    return "fastmcp" in sys.modules
+
+
 @lru_cache
 def is_langchain() -> bool:
     """Return True if running in a Langchain environment.
@@ -55,15 +81,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

@@ -30,6 +30,14 @@ def _load_dotenv_file(dotenv_path: Path | str) -> None:
 def initialize_secrets() -> None:
     """Initialize dotenv to load environment variables from .env files."""
     # 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)

airbyte/mcp/server.py

@@ -6,12 +6,14 @@
 
 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
 
 
+set_mcp_mode()
 initialize_secrets()
 
 app: FastMCP = FastMCP("airbyte-mcp")