From 9908f3b443f5de6d5838f95dc0dfedfabb1b04e9 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 24 Mar 2026 01:06:06 +0800 Subject: [PATCH] chore: auto commit --- astrbot/_internal/tools/base.py | 9 ++++ astrbot/_internal/tools/registry.py | 83 +++++++++++++++++++++++++++-- 2 files changed, 88 insertions(+), 4 deletions(-) diff --git a/astrbot/_internal/tools/base.py b/astrbot/_internal/tools/base.py index 2212e4dfb..78f389f3e 100644 --- a/astrbot/_internal/tools/base.py +++ b/astrbot/_internal/tools/base.py @@ -108,6 +108,10 @@ class ToolSet: """Get a tool by name.""" return self._tools.get(name) + def get_tool(self, name: str) -> FunctionTool | None: + """Get a tool by name (alias for get).""" + return self.get(name) + def list_tools(self) -> list[FunctionTool]: """List all tools in this set.""" return list(self._tools.values()) @@ -117,3 +121,8 @@ class ToolSet: def __len__(self) -> int: return len(self._tools) + + @property + def tools(self) -> list[FunctionTool]: + """List all tools in this set.""" + return list(self._tools.values()) diff --git a/astrbot/_internal/tools/registry.py b/astrbot/_internal/tools/registry.py index a50b0031e..9cea9e70d 100644 --- a/astrbot/_internal/tools/registry.py +++ b/astrbot/_internal/tools/registry.py @@ -63,11 +63,32 @@ class FunctionToolManager: return False def get_func(self, name: str) -> FunctionTool | None: - """Get a tool by name.""" - for f in self.func_list: + """Get a tool by name. Returns the last active tool if multiple match.""" + last_match: FunctionTool | None = None + for f in reversed(self.func_list): if f.name == name: - return f - return None + if getattr(f, "active", True): + return f + if last_match is None: + last_match = f + return last_match + + def get_full_tool_set(self) -> ToolSet: + """Return a ToolSet with all active tools, deduplicated by name.""" + seen: dict[str, FunctionTool] = {} + for tool in reversed(self.func_list): + if tool.name not in seen and getattr(tool, "active", True): + seen[tool.name] = tool + return ToolSet("default", list(seen.values())) + + def register_internal_tools(self) -> None: + """Register built-in computer tools (shell, python, browser, neo).""" + # Import here to avoid circular imports + from astrbot.core.computer.computer_tool_provider import get_all_tools + + for tool in get_all_tools(): + if self.get_func(tool.name) is None: + self.add(tool) class FuncCall(FunctionToolManager): @@ -75,3 +96,57 @@ class FuncCall(FunctionToolManager): def __init__(self) -> None: super().__init__() + self._mcp_server_runtime_view: dict[str, Any] = {} + + @property + def mcp_server_runtime_view(self) -> dict[str, Any]: + """Return runtime view of MCP servers.""" + return self._mcp_server_runtime_view + + async def enable_mcp_server( + self, name: str, config: dict[str, Any], init_timeout: int = 30 + ) -> None: + """Enable an MCP server (stub implementation).""" + pass + + async def disable_mcp_server( + self, name: str = "", timeout: int = 10, shutdown_timeout: int = 10 + ) -> None: + """Disable an MCP server (stub implementation).""" + pass + + def load_mcp_config(self) -> dict[str, Any]: + """Load MCP configuration (stub implementation).""" + return {"mcpServers": {}} + + def save_mcp_config(self, config: dict[str, Any]) -> bool: + """Save MCP configuration (stub implementation).""" + return True + + def activate_llm_tool(self, name: str) -> bool: + """Activate an LLM tool (stub implementation).""" + return True + + def deactivate_llm_tool(self, name: str) -> bool: + """Deactivate an LLM tool (stub implementation).""" + return True + + async def test_mcp_server_connection(self, config: dict[str, Any]) -> tuple[bool, str]: + """Test MCP server connection (stub implementation).""" + # Import the actual test function if available + try: + from astrbot._internal.protocols.mcp.client import _quick_test_mcp_connection + success, message = await _quick_test_mcp_connection(config) + if not success: + raise Exception(message) + return success, message + except Exception as e: + raise Exception(f"MCP connection test failed: {e!s}") from e + + async def sync_modelscope_mcp_servers(self) -> None: + """Sync ModelScope MCP servers (stub implementation).""" + pass + + def get_full_tool_set(self) -> ToolSet: + """Return a ToolSet with all active tools.""" + return ToolSet("default", [t for t in self.func_list if t.active])