openclaw/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

fix(tts): resolve API keys from ~/.hermes/.env via get_env_value (#17140)

Browse Source 
TTS provider tools (elevenlabs, xai, minimax, mistral, gemini) called
os.getenv("X_API_KEY") directly, which bypassed Hermes's dotenv bridge in
hermes_cli.config. Users who keep their TTS keys only in ~/.hermes/.env saw
"X_API_KEY not set" errors even though the rest of the stack
(agent/credential_pool, hermes_cli/auth) already resolves keys through
get_env_value() — same class of bug as #15914 fixed for those modules.

Switch every TTS env-var lookup (API keys, base URLs, and
check_tts_requirements gates) to get_env_value, which checks os.environ
first and then ~/.hermes/.env. Behaviour for users with keys exported in
the shell is unchanged; users with dotenv-only keys now succeed. The two
diagnostics prints in __main__ are migrated for consistency.

Regression test (tests/tools/test_tts_dotenv_fallback.py):
  - per-provider: each backend reads the dotenv key when only
    ~/.hermes/.env carries it (5 providers).
  - end-to-end: with hermes_cli.config.load_env returning the key and
    os.environ empty, _generate_minimax_tts and check_tts_requirements
    both succeed; reverting tools/tts_tool.py back to os.getenv makes all
    7 tests fail with "MINIMAX_API_KEY not set" / similar.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
briandevans
2026-04-28 15:16:05 -07:00
committed by Teknium
parent ff687c019e
commit 40d25e125b
2 changed files with 246 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

230
tests/tools/test_tts_dotenv_fallback.py Normal file
View File

@ -0,0 +1,230 @@
"""Regression tests for #17140.
TTS provider tools must resolve API keys from ``~/.hermes/.env`` (via
``hermes_cli.config.get_env_value``) and not only from ``os.environ`` —
otherwise users who keep their keys in the dotenv file see "API key not set"
errors even though the key is configured. Same class of bug as #15914 (auth)
already addressed for ``agent/credential_pool`` and ``hermes_cli/auth``.
"""
from unittest.mock import MagicMock, patch
import pytest
@pytest.fixture(autouse=True)
def isolate_env(monkeypatch):
"""Strip every TTS-related env var so the test really exercises the
dotenv code path. If any of these survive into the test, the assertion
that ``get_env_value`` was consulted becomes meaningless because
``os.environ`` already satisfies the lookup.
"""
for key in (
"ELEVENLABS_API_KEY",
"XAI_API_KEY",
"XAI_BASE_URL",
"MINIMAX_API_KEY",
"MISTRAL_API_KEY",
"GEMINI_API_KEY",
"GEMINI_BASE_URL",
"GOOGLE_API_KEY",
):
monkeypatch.delenv(key, raising=False)
class TestDotenvFallbackPerProvider:
"""For each affected provider, when only ``~/.hermes/.env`` carries the
key (mocked via ``hermes_cli.config.load_env``), the provider must find
it. Before the fix, ``os.getenv`` returned ``None`` and the provider
raised ``ValueError("X_API_KEY not set")``.
"""
def test_elevenlabs_reads_dotenv_key(self, tmp_path):
from tools import tts_tool
with patch.object(tts_tool, "get_env_value", return_value="el-dotenv-key"), \
patch.object(tts_tool, "_import_elevenlabs") as mock_import:
mock_client = MagicMock()
mock_client.text_to_speech.convert.return_value = iter([b"audio"])
mock_import.return_value = MagicMock(return_value=mock_client)
output = str(tmp_path / "out.mp3")
tts_tool._generate_elevenlabs("hi", output, {})
mock_import.return_value.assert_called_once_with(api_key="el-dotenv-key")
def test_xai_reads_dotenv_key(self, tmp_path):
from tools import tts_tool
captured: dict = {}
def fake_post(url, **kwargs):
captured["url"] = url
captured["headers"] = kwargs.get("headers", {})
response = MagicMock()
response.content = b"audio"
response.raise_for_status = MagicMock()
return response
with patch.object(tts_tool, "get_env_value", return_value="xai-dotenv-key"), \
patch("requests.post", side_effect=fake_post):
tts_tool._generate_xai_tts("hi", str(tmp_path / "out.mp3"), {})
assert captured["headers"]["Authorization"] == "Bearer xai-dotenv-key"
def test_minimax_reads_dotenv_key(self, tmp_path):
from tools import tts_tool
captured: dict = {}
def fake_post(url, **kwargs):
captured["headers"] = kwargs.get("headers", {})
response = MagicMock()
response.json.return_value = {
"data": {"audio": b"\x00\x01".hex()},
"base_resp": {"status_code": 0},
}
response.raise_for_status = MagicMock()
return response
with patch.object(tts_tool, "get_env_value", return_value="mm-dotenv-key"), \
patch("requests.post", side_effect=fake_post):
tts_tool._generate_minimax_tts("hi", str(tmp_path / "out.mp3"), {})
assert captured["headers"]["Authorization"] == "Bearer mm-dotenv-key"
def test_mistral_reads_dotenv_key(self, tmp_path):
import base64
from tools import tts_tool
seen_keys: list = []
def fake_mistral_factory(*, api_key=None):
seen_keys.append(api_key)
client = MagicMock()
client.__enter__ = MagicMock(return_value=client)
client.__exit__ = MagicMock(return_value=False)
client.audio.speech.complete.return_value = MagicMock(
audio_data=base64.b64encode(b"data").decode()
)
return client
with patch.object(tts_tool, "get_env_value", return_value="mistral-dotenv-key"), \
patch.object(tts_tool, "_import_mistral_client", return_value=fake_mistral_factory):
tts_tool._generate_mistral_tts("hi", str(tmp_path / "out.mp3"), {})
assert seen_keys == ["mistral-dotenv-key"]
def test_gemini_reads_dotenv_key(self, tmp_path):
from tools import tts_tool
captured: dict = {}
def fake_post(url, **kwargs):
captured["params"] = kwargs.get("params", {})
response = MagicMock()
response.status_code = 200
response.json.return_value = {
"candidates": [
{
"content": {
"parts": [
{
"inlineData": {
"data": "AAAA",
"mimeType": "audio/L16;codec=pcm;rate=24000",
}
}
]
}
}
]
}
response.raise_for_status = MagicMock()
return response
# GEMINI_API_KEY hits the first branch; GOOGLE_API_KEY would only be
# consulted if the first returned None. Use a side-effect-style mock
# to verify the lookup order matches the production code.
seen_lookups: list = []
def fake_get_env_value(key):
seen_lookups.append(key)
if key == "GEMINI_API_KEY":
return "gemini-dotenv-key"
return None
with patch.object(tts_tool, "get_env_value", side_effect=fake_get_env_value), \
patch("requests.post", side_effect=fake_post):
tts_tool._generate_gemini_tts("hi", str(tmp_path / "out.wav"), {})
assert "GEMINI_API_KEY" in seen_lookups
assert captured["params"]["key"] == "gemini-dotenv-key"
class TestRegressionGuard:
"""Goal-backward proof that the old behaviour ('only check ``os.environ``')
breaks reading from a dotenv-only key, and the new behaviour fixes it.
Implemented as an end-to-end probe that patches
``hermes_cli.config.load_env`` to simulate ``~/.hermes/.env`` carrying the
key while ``os.environ`` does not.
"""
def test_minimax_missing_when_only_in_dotenv_before_fix(self, tmp_path, monkeypatch):
from tools import tts_tool
monkeypatch.delenv("MINIMAX_API_KEY", raising=False)
# Simulate ~/.hermes/.env carrying the key (load_env returns the dict
# that get_env_value falls back to). The pre-fix ``os.getenv`` call
# ignores this entirely and raises ValueError.
with patch(
"hermes_cli.config.load_env",
return_value={"MINIMAX_API_KEY": "dotenv-secret"},
):
# Sanity-check: get_env_value resolves through load_env when
# os.environ is empty.
from hermes_cli.config import get_env_value as live_get
assert live_get("MINIMAX_API_KEY") == "dotenv-secret"
# And the production code path now consumes the resolved value
# instead of raising "MINIMAX_API_KEY not set".
captured: dict = {}
def fake_post(url, **kwargs):
captured["headers"] = kwargs.get("headers", {})
response = MagicMock()
response.json.return_value = {
"data": {"audio": b"\x00".hex()},
"base_resp": {"status_code": 0},
}
response.raise_for_status = MagicMock()
return response
with patch("requests.post", side_effect=fake_post):
tts_tool._generate_minimax_tts(
"hi", str(tmp_path / "out.mp3"), {}
)
assert captured["headers"]["Authorization"] == "Bearer dotenv-secret"
def test_check_tts_requirements_sees_dotenv_minimax(self, monkeypatch):
"""``check_tts_requirements`` is the gate that decides whether
``/voice on`` is even offered. If it only checked ``os.environ`` it
would say "no provider available" for users who keep MINIMAX_API_KEY
in ``~/.hermes/.env``, even though the dispatcher would later succeed.
"""
from tools import tts_tool
monkeypatch.delenv("MINIMAX_API_KEY", raising=False)
with patch(
"hermes_cli.config.load_env",
return_value={"MINIMAX_API_KEY": "dotenv-secret"},
), patch.object(tts_tool, "_import_edge_tts", side_effect=ImportError), \
patch.object(tts_tool, "_import_elevenlabs", side_effect=ImportError), \
patch.object(tts_tool, "_import_openai_client", side_effect=ImportError), \
patch.object(tts_tool, "_check_neutts_available", return_value=False), \
patch.object(tts_tool, "_check_kittentts_available", return_value=False):
assert tts_tool.check_tts_requirements() is True

31
tools/tts_tool.py
View File

@ -44,6 +44,7 @@ from urllib.parse import urljoin
from hermes_constants import display_hermes_home
logger = logging.getLogger(__name__)
from hermes_cli.config import get_env_value
from tools.managed_tool_gateway import resolve_managed_tool_gateway
from tools.tool_backend_helpers import managed_nous_tools_enabled, prefers_gateway, resolve_openai_audio_api_key
from tools.xai_http import hermes_xai_user_agent
@ -312,7 +313,7 @@ def _generate_elevenlabs(text: str, output_path: str, tts_config: Dict[str, Any]
Returns:
Path to the saved audio file.
"""
api_key = os.getenv("ELEVENLABS_API_KEY", "")
api_key = (get_env_value("ELEVENLABS_API_KEY") or "")
if not api_key:
raise ValueError("ELEVENLABS_API_KEY not set. Get one at https://elevenlabs.io/")
@ -406,7 +407,7 @@ def _generate_xai_tts(text: str, output_path: str, tts_config: Dict[str, Any]) -
"""
import requests
api_key = os.getenv("XAI_API_KEY", "").strip()
api_key = (get_env_value("XAI_API_KEY") or "").strip()
if not api_key:
raise ValueError("XAI_API_KEY not set. Get one at https://console.x.ai/")
@ -417,7 +418,7 @@ def _generate_xai_tts(text: str, output_path: str, tts_config: Dict[str, Any]) -
bit_rate = int(xai_config.get("bit_rate", DEFAULT_XAI_BIT_RATE))
base_url = str(
xai_config.get("base_url")
or os.getenv("XAI_BASE_URL")
or get_env_value("XAI_BASE_URL")
or DEFAULT_XAI_BASE_URL
).strip().rstrip("/")
@ -479,7 +480,7 @@ def _generate_minimax_tts(text: str, output_path: str, tts_config: Dict[str, Any
"""
import requests
api_key = os.getenv("MINIMAX_API_KEY", "")
api_key = (get_env_value("MINIMAX_API_KEY") or "")
if not api_key:
raise ValueError("MINIMAX_API_KEY not set. Get one at https://platform.minimax.io/")
@ -556,7 +557,7 @@ def _generate_mistral_tts(text: str, output_path: str, tts_config: Dict[str, Any
and writes the raw bytes to *output_path*.
Supports native Opus output for Telegram voice bubbles.
"""
api_key = os.getenv("MISTRAL_API_KEY", "")
api_key = (get_env_value("MISTRAL_API_KEY") or "")
if not api_key:
raise ValueError("MISTRAL_API_KEY not set. Get one at https://console.mistral.ai/")
@ -651,7 +652,7 @@ def _generate_gemini_tts(text: str, output_path: str, tts_config: Dict[str, Any]
"""
import requests
api_key = (os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY") or "").strip()
api_key = (get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY") or "").strip()
if not api_key:
raise ValueError(
"GEMINI_API_KEY not set. Get one at https://aistudio.google.com/app/apikey"
@ -662,7 +663,7 @@ def _generate_gemini_tts(text: str, output_path: str, tts_config: Dict[str, Any]
voice = str(gemini_config.get("voice", DEFAULT_GEMINI_TTS_VOICE)).strip() or DEFAULT_GEMINI_TTS_VOICE
base_url = str(
gemini_config.get("base_url")
or os.getenv("GEMINI_BASE_URL")
or get_env_value("GEMINI_BASE_URL")
or DEFAULT_GEMINI_TTS_BASE_URL
).strip().rstrip("/")
@ -1148,7 +1149,7 @@ def check_tts_requirements() -> bool:
pass
try:
_import_elevenlabs()
if os.getenv("ELEVENLABS_API_KEY"):
if get_env_value("ELEVENLABS_API_KEY"):
return True
except ImportError:
pass
@ -1158,15 +1159,15 @@ def check_tts_requirements() -> bool:
return True
except ImportError:
pass
if os.getenv("MINIMAX_API_KEY"):
if get_env_value("MINIMAX_API_KEY"):
return True
if os.getenv("XAI_API_KEY"):
if get_env_value("XAI_API_KEY"):
return True
if os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY"):
if get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY"):
return True
try:
_import_mistral_client()
if os.getenv("MISTRAL_API_KEY"):
if get_env_value("MISTRAL_API_KEY"):
return True
except ImportError:
pass
@ -1278,7 +1279,7 @@ def stream_tts_to_speaker(
{**tts_config, "elevenlabs": {**el_config, "model_id": model_id}},
)
api_key = os.getenv("ELEVENLABS_API_KEY", "")
api_key = (get_env_value("ELEVENLABS_API_KEY") or "")
if not api_key:
logger.warning("ELEVENLABS_API_KEY not set; streaming TTS audio disabled")
else:
@ -1464,13 +1465,13 @@ if __name__ == "__main__":
print("\nProvider availability:")
print(f" Edge TTS: {'installed' if _check(_import_edge_tts, 'edge') else 'not installed (pip install edge-tts)'}")
print(f" ElevenLabs: {'installed' if _check(_import_elevenlabs, 'el') else 'not installed (pip install elevenlabs)'}")
print(f" API Key: {'set' if os.getenv('ELEVENLABS_API_KEY') else 'not set'}")
print(f" API Key: {'set' if get_env_value('ELEVENLABS_API_KEY') else 'not set'}")
print(f" OpenAI: {'installed' if _check(_import_openai_client, 'oai') else 'not installed'}")
print(
" API Key: "
f"{'set' if resolve_openai_audio_api_key() else 'not set (VOICE_TOOLS_OPENAI_KEY or OPENAI_API_KEY)'}"
)
print(f" MiniMax: {'API key set' if os.getenv('MINIMAX_API_KEY') else 'not set (MINIMAX_API_KEY)'}")
print(f" MiniMax: {'API key set' if get_env_value('MINIMAX_API_KEY') else 'not set (MINIMAX_API_KEY)'}")
print(f" ffmpeg: {'✅ found' if _has_ffmpeg() else '❌ not found (needed for Telegram Opus)'}")
print(f"\n Output dir: {DEFAULT_OUTPUT_DIR}")