auth(veauth): implement speech token retrieval with credential fallback

Author: wangyue.demon

config.yaml.full

@@ -46,6 +46,11 @@ tool:
   web_scraper: 
     endpoint: 
     api_key:    # `token`
+  # [optional] https://console.volcengine.com/speech/new/experience/tts
+  lark:
+    app_id:     # `app_id`
+    api_key:    # `app_secret`
+    speaker:    # `speaker`
   # [optional] https://open.larkoffice.com/app
   lark: 
     endpoint:   # `app_id`

tests/tools/builtin_tools/test_tts.py

@@ -37,9 +37,9 @@ def setUp(self):
         self.patcher_env = patch.dict(
             "os.environ",
             {
-                "TOOL_TTS_APP_ID": "test_app_id",
-                "TOOL_TTS_API_KEY": "test_api_key",
-                "TOOL_TTS_SPEAKER": "test_speaker",
+                "TOOL_VESPEECH_APP_ID": "test_app_id",
+                "TOOL_VESPEECH_API_KEY": "test_api_key",
+                "TOOL_VESPEECH_SPEAKER": "test_speaker",
             },
         )
         self.patcher_env.start()
@@ -64,8 +64,7 @@ def test_tts_success(self, mock_session):
 
         # Assertions
         self.assertIsInstance(result, dict)
-        self.assertIn("text", result)
-        self.assertIn("audio_path", result)
+        self.assertIn("saved_audio_path", result)
         mock_session.return_value.post.assert_called_once()
         mock_response.close.assert_called_once()
 

veadk/auth/veauth/speech_veauth.py

@@ -36,9 +36,9 @@ def get_speech_token(region: str = "cn-beijing") -> str:
         session_token = cred.session_token
 
     res = ve_request(
-        request_body={"ProjectName": "default", "OnlyAvailable": True, "Filter": {}},
+        request_body={"ProjectName": "default", "OnlyAvailable": True},
         header={"X-Security-Token": session_token},
-        action="ListApiKeys",
+        action="ListAPIKeys",
         ak=access_key,
         sk=secret_key,
         service="speech_saas_prod",

veadk/configs/tool_configs.py

@@ -20,6 +20,7 @@
 
 from veadk.auth.veauth.prompt_pilot_veauth import PromptPilotVeAuth
 from veadk.auth.veauth.vesearch_veauth import VesearchVeAuth
+from veadk.auth.veauth.speech_veauth import get_speech_token
 
 
 class PromptPilotConfig(BaseModel):
@@ -38,5 +39,16 @@ def api_key(self) -> str:
         return os.getenv("TOOL_VESEARCH_API_KEY") or VesearchVeAuth().token
 
 
+class VeSpeechConfig(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="TOOL_VESPEECH_")
+
+    endpoint: int | str = ""
+
+    @cached_property
+    def api_key(self) -> str:
+        return os.getenv("TOOL_VESPEECH_API_KEY") or get_speech_token()
+
+
 class BuiltinToolConfigs(BaseModel):
     vesearch: VeSearchConfig = Field(default_factory=VeSearchConfig)
+    vespeech: VeSpeechConfig = Field(default_factory=VeSpeechConfig)

veadk/tools/builtin_tools/tts.py

@@ -23,7 +23,7 @@
 import tempfile
 from typing import Dict, Any
 from google.adk.tools import ToolContext
-from veadk.config import getenv
+from veadk.config import getenv, settings
 from veadk.utils.logger import get_logger
 from veadk.utils.audio_manager import AudioDeviceManager, AudioConfig
 
@@ -60,16 +60,16 @@ def text_to_speech(text: str, tool_context: ToolContext) -> Dict[str, Any]:
     url = "https://openspeech.bytedance.com/api/v3/tts/unidirectional"
     audio_save_path = ""
 
-    app_id = getenv("TOOL_TTS_APP_ID")
-    api_key = getenv("TOOL_TTS_API_KEY")
+    app_id = getenv("TOOL_VESPEECH_APP_ID")
     speaker = getenv(
-        "TOOL_TTS_SPEAKER", "zh_female_vv_uranus_bigtts"
+        "TOOL_VESPEECH_SPEAKER", "zh_female_vv_uranus_bigtts"
     )  # e.g. zh_female_vv_mars_bigtts
+    api_key = settings.tool.vespeech.api_key
     if not all([app_id, api_key, speaker]):
         return {
             "error": (
                 "Tool text_to_speech execution failed. Missing required env vars: "
-                "TOOL_TTS_APP_ID, TOOL_TTS_API_KEY, TOOL_TTS_SPEAKER"
+                "TOOL_VESPEECH_APP_ID, TOOL_VESPEECH_API_KEY, TOOL_VESPEECH_SPEAKER"
             )
         }