Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions airbyte/mcp/_local_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
from airbyte.caches.util import get_default_cache
from airbyte.mcp._util import resolve_config, resolve_list_of_strings
from airbyte.secrets.config import _get_secret_sources
from airbyte.secrets.env_vars import DotenvSecretManager
from airbyte.secrets.google_gsm import GoogleGSMSecretManager
from airbyte.sources.base import Source
from airbyte.sources.registry import get_connector_metadata
Expand Down Expand Up @@ -184,6 +185,20 @@ def list_connector_config_secrets(
return secrets_names


def list_dotenv_secrets() -> dict[str, list[str]]:
"""List all environment variable names declared within declared .env files.

This returns a dictionary mapping the .env file name to a list of environment
variable names. The values of the environment variables are not returned.
"""
result: dict[str, list[str]] = {}
for secrets_mgr in _get_secret_sources():
if isinstance(secrets_mgr, DotenvSecretManager) and secrets_mgr.dotenv_path:
result[secrets_mgr.dotenv_path.absolute().name] = secrets_mgr.list_secrets_names()

return result


# @app.tool() # << deferred
def list_source_streams(
source_connector_name: Annotated[
Expand Down Expand Up @@ -673,6 +688,7 @@ def register_local_ops_tools(app: FastMCP) -> None:
get_source_stream_json_schema,
get_stream_previews,
list_cached_streams,
list_dotenv_secrets,
list_source_streams,
read_source_stream_records,
run_sql_query,
Expand Down
7 changes: 7 additions & 0 deletions airbyte/secrets/env_vars.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,3 +62,10 @@ def get_secret(self, secret_name: str) -> SecretString | None:
return None

return SecretString(dotenv_vars[secret_name])

def list_secrets_names(self) -> list[str]:
"""List all secrets available in the .env file."""
dotenv_vars: dict[str, str | None] = dotenv_values(
dotenv_path=self.dotenv_path,
)
return list(dotenv_vars.keys())