mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-15 17:30:13 +08:00
chore: auto commit
This commit is contained in:
@@ -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())
|
||||
|
||||
@@ -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])
|
||||
|
||||
Reference in New Issue
Block a user