refactor: Centralize and decouple computer-use tool injection logic into a new ComputerToolProvider and associated tool modules.

This commit is contained in:
advent259141
2026-03-10 22:00:23 +08:00
parent bf430e659a
commit ff4412a627
9 changed files with 149 additions and 1 deletions
+1
View File
@@ -374,6 +374,7 @@ class MCPTool(FunctionTool, Generic[TContext]):
self.mcp_tool = mcp_tool
self.mcp_client = mcp_client
self.mcp_server_name = mcp_server_name
self.source = "mcp"
async def call(
self, context: ContextWrapper[TContext], **kwargs
+5
View File
@@ -63,6 +63,11 @@ class FunctionTool(ToolSchema, Generic[TContext]):
Declare this tool as a background task. Background tasks return immediately
with a task identifier while the real work continues asynchronously.
"""
source: str = "plugin"
"""
Origin of this tool: 'plugin' (from star plugins), 'internal' (AstrBot built-in),
or 'mcp' (from MCP servers). Used by WebUI for display grouping.
"""
def __repr__(self) -> str:
return f"FuncTool(name={self.name}, parameters={self.parameters}, description={self.description})"
@@ -161,6 +161,73 @@ def _build_local_mode_prompt() -> str:
class ComputerToolProvider:
"""Provides computer-use tools (local / sandbox) based on session context."""
@staticmethod
def get_all_tools() -> list[FunctionTool]:
"""Return ALL computer-use tools across all runtimes for registration.
Creates **fresh instances** separate from the runtime caches so that
setting ``active=False`` on them does not affect runtime behaviour.
These registration-only instances let the WebUI display and assign
tools without injecting them into actual LLM requests.
At request time, ``get_tools(ctx)`` provides the real, active
instances filtered by runtime.
"""
from astrbot.core.computer.tools import (
AnnotateExecutionTool,
BrowserBatchExecTool,
BrowserExecTool,
CreateSkillCandidateTool,
CreateSkillPayloadTool,
EvaluateSkillCandidateTool,
ExecuteShellTool,
FileDownloadTool,
FileUploadTool,
GetExecutionHistoryTool,
GetSkillPayloadTool,
ListSkillCandidatesTool,
ListSkillReleasesTool,
LocalPythonTool,
PromoteSkillCandidateTool,
PythonTool,
RollbackSkillReleaseTool,
RunBrowserSkillTool,
SyncSkillReleaseTool,
)
all_tools: list[FunctionTool] = [
ExecuteShellTool(),
PythonTool(),
FileUploadTool(),
FileDownloadTool(),
LocalPythonTool(),
BrowserExecTool(),
BrowserBatchExecTool(),
RunBrowserSkillTool(),
GetExecutionHistoryTool(),
AnnotateExecutionTool(),
CreateSkillPayloadTool(),
GetSkillPayloadTool(),
CreateSkillCandidateTool(),
ListSkillCandidatesTool(),
EvaluateSkillCandidateTool(),
PromoteSkillCandidateTool(),
ListSkillReleasesTool(),
RollbackSkillReleaseTool(),
SyncSkillReleaseTool(),
]
# De-duplicate by name and mark inactive so they are visible
# in WebUI but never sent to the LLM via func_list.
seen: set[str] = set()
result: list[FunctionTool] = []
for tool in all_tools:
if tool.name not in seen:
tool.active = False
result.append(tool)
seen.add(tool.name)
return result
def get_tools(self, ctx: ToolProviderContext) -> list[FunctionTool]:
runtime = ctx.computer_use_runtime
if runtime == "none":
@@ -239,3 +306,12 @@ class ComputerToolProvider:
if existing_booter is not None:
return getattr(existing_booter, "capabilities", None)
return None
def get_all_tools() -> list[FunctionTool]:
"""Module-level entry point for ``FunctionToolManager.register_internal_tools()``.
Delegates to ``ComputerToolProvider.get_all_tools()`` which collects
tools from all runtimes (local, sandbox, browser, neo).
"""
return ComputerToolProvider.get_all_tools()
+1 -1
View File
@@ -164,7 +164,7 @@ class CreateSkillPayloadTool(NeoSkillToolBase):
"type": "object",
"properties": {
"payload": {
"anyOf": [{"type": "object"}, {"type": "array"}],
"anyOf": [{"type": "object"}, {"type": "array", "items": {}}],
"description": (
"Skill payload JSON. Typical schema: {skill_markdown, inputs, outputs, meta}. "
"This only stores content and returns payload_ref; it does not create a candidate or release."
@@ -907,6 +907,50 @@ class FunctionToolManager:
except Exception as e:
raise Exception(f"同步 ModelScope MCP 服务器时发生错误: {e!s}")
# Module paths whose ``get_all_tools()`` function returns internal tools.
# To add a new internal-tool provider, simply append its module path here.
_INTERNAL_TOOL_PROVIDERS: list[str] = [
"astrbot.core.tools.cron_tools",
"astrbot.core.tools.kb_query",
"astrbot.core.tools.send_message",
"astrbot.core.computer.computer_tool_provider",
]
def register_internal_tools(self) -> None:
"""Register AstrBot built-in tools from all internal providers.
Each provider module is expected to expose a ``get_all_tools()``
function that returns a list of ``FunctionTool`` instances.
Tools are marked with ``source='internal'`` so the WebUI can
distinguish them from plugin and MCP tools, and subagent
orchestrators can resolve them by name.
Duplicate registration is idempotent (skips if name already present).
"""
import importlib
existing_names = {t.name for t in self.func_list}
for module_path in self._INTERNAL_TOOL_PROVIDERS:
try:
mod = importlib.import_module(module_path)
provider_tools = mod.get_all_tools()
except Exception as e:
logger.warning(
"Failed to load internal tool provider %s: %s",
module_path,
e,
)
continue
for tool in provider_tools:
tool.source = "internal"
if tool.name not in existing_names:
self.func_list.append(tool)
existing_names.add(tool.name)
logger.info("Registered internal tool: %s", tool.name)
def __str__(self) -> str:
return str(self.func_list)
+5
View File
@@ -101,6 +101,11 @@ class Context:
"""Cron job manager, initialized by core lifecycle."""
self.subagent_orchestrator = subagent_orchestrator
# Register built-in tools so they appear in WebUI and can be
# assigned to subagents. Done here (not at module-import time)
# to avoid circular imports.
self.provider_manager.llm_tools.register_internal_tools()
async def llm_generate(
self,
*,
+7
View File
@@ -184,6 +184,12 @@ CREATE_CRON_JOB_TOOL = CreateActiveCronTool()
DELETE_CRON_JOB_TOOL = DeleteCronJobTool()
LIST_CRON_JOBS_TOOL = ListCronJobsTool()
def get_all_tools() -> list[FunctionTool]:
"""Return all cron-related tools for registration."""
return [CREATE_CRON_JOB_TOOL, DELETE_CRON_JOB_TOOL, LIST_CRON_JOBS_TOOL]
__all__ = [
"CREATE_CRON_JOB_TOOL",
"DELETE_CRON_JOB_TOOL",
@@ -191,4 +197,5 @@ __all__ = [
"CreateActiveCronTool",
"DeleteCronJobTool",
"ListCronJobsTool",
"get_all_tools",
]
+5
View File
@@ -132,3 +132,8 @@ async def retrieve_knowledge_base(
KNOWLEDGE_BASE_QUERY_TOOL = KnowledgeBaseQueryTool()
def get_all_tools() -> list[FunctionTool]:
"""Return all knowledge-base tools for registration."""
return [KNOWLEDGE_BASE_QUERY_TOOL]
+5
View File
@@ -211,3 +211,8 @@ class SendMessageToUserTool(FunctionTool[AstrAgentContext]):
SEND_MESSAGE_TO_USER_TOOL = SendMessageToUserTool()
def get_all_tools() -> list[FunctionTool]:
"""Return all send-message tools for registration."""
return [SEND_MESSAGE_TO_USER_TOOL]