Support tool config in yaml

core/duplex_pipeline.py

@@ -206,7 +206,8 @@ class DuplexPipeline:
         self._runtime_barge_in_min_duration_ms: Optional[int] = None
         self._runtime_knowledge: Dict[str, Any] = {}
         self._runtime_knowledge_base_id: Optional[str] = None
-        self._runtime_tools: List[Any] = []
+        raw_default_tools = settings.tools if isinstance(settings.tools, list) else []
+        self._runtime_tools: List[Any] = list(raw_default_tools)
         self._runtime_tool_executor: Dict[str, str] = {}
         self._pending_tool_waiters: Dict[str, asyncio.Future] = {}
         self._early_tool_results: Dict[str, Dict[str, Any]] = {}
@@ -227,6 +228,8 @@ class DuplexPipeline:
         self._pending_llm_delta: str = ""
         self._last_llm_delta_emit_ms: float = 0.0
 
+        self._runtime_tool_executor = self._resolved_tool_executor_map()
+
         if self._server_tool_executor is None:
             if self._tool_resource_resolver:
                 async def _executor(call: Dict[str, Any]) -> Dict[str, Any]:
@@ -369,7 +372,7 @@ class DuplexPipeline:
                 },
             },
             "tools": {
-                "allowlist": sorted(self._runtime_tool_executor.keys()),
+                "allowlist": self._resolved_tool_allowlist(),
             },
             "tracks": {
                 "audio_in": self.track_audio_in,
@@ -1165,6 +1168,23 @@ class DuplexPipeline:
                 result[name] = executor
         return result
 
+    def _resolved_tool_allowlist(self) -> List[str]:
+        names: set[str] = set()
+        for item in self._runtime_tools:
+            if isinstance(item, str):
+                name = item.strip()
+                if name:
+                    names.add(name)
+                continue
+            if not isinstance(item, dict):
+                continue
+            fn = item.get("function")
+            if isinstance(fn, dict) and fn.get("name"):
+                names.add(str(fn.get("name")).strip())
+            elif item.get("name"):
+                names.add(str(item.get("name")).strip())
+        return sorted([name for name in names if name])
+
     def _tool_name(self, tool_call: Dict[str, Any]) -> str:
         fn = tool_call.get("function")
         if isinstance(fn, dict):