Spaces:
Running
Running
File size: 3,176 Bytes
9d0fd45 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """Explicit built-in tool registry for the OSS workflows."""
from __future__ import annotations
import logging
from frontier_agent.core.tool import Tool
from plugins.tools.assign_task import assign_task
from plugins.tools.bash import bash
from plugins.tools.collect_reports import collect_reports
from plugins.tools.create_file import create_file
from plugins.tools.create_subagent import create_subagent
from plugins.tools.download_file import download_file
from plugins.tools.file_editor import (
file_editor_create,
file_editor_str_replace,
file_editor_view,
)
from plugins.tools.glob_search import glob_search
from plugins.tools.grep_search import grep_search
from plugins.tools.read_file import read_file
from plugins.tools.recover_result import recover_result
from plugins.tools.run_python_code import run_python_code
from plugins.tools.stop_subagent import stop_subagent
from plugins.tools.submit_report import submit_report
from plugins.tools.task_board import add_task, finish_planning, update_task
from plugins.tools.view_image import view_image
from plugins.tools.web_fetch import web_fetch
from plugins.tools.web_search import web_search
from plugins.tools.write_file import write_file
logger = logging.getLogger(__name__)
_BUILTIN_TOOLS: list[Tool] = [
web_search,
web_fetch,
download_file,
bash,
create_subagent,
assign_task,
add_task,
update_task,
finish_planning,
collect_reports,
stop_subagent,
read_file,
create_file,
write_file,
file_editor_view,
file_editor_create,
file_editor_str_replace,
submit_report,
view_image,
grep_search,
glob_search,
run_python_code,
recover_result,
]
def get_builtin_tools() -> dict[str, Tool]:
"""Return the allowlisted built-ins as a name-to-tool mapping."""
return {tool.name: tool for tool in _BUILTIN_TOOLS}
class ToolRegistry:
"""Central registration and fail-closed role filtering."""
def __init__(self) -> None:
self._tools: dict[str, Tool] = {}
def register(self, tool: Tool) -> None:
self._tools[tool.name] = tool
def register_all(self, tools: dict[str, Tool]) -> None:
self._tools.update(tools)
def get(self, name: str) -> Tool | None:
return self._tools.get(name)
def get_all(self) -> dict[str, Tool]:
return dict(self._tools)
def get_for_role(self, role_id: str) -> list[Tool]:
try:
from frontier_agent.core.runtime.registries import services
from frontier_agent.core.runtime.registries.agents import AgentRegistry
allowed = set(services.get(AgentRegistry).get_tools_for(role_id))
return [tool for name, tool in self._tools.items() if name in allowed]
except Exception as exc:
logger.warning("Tool lookup for role %s failed: %s", role_id, exc)
return []
def names(self) -> list[str]:
return sorted(self._tools)
def __len__(self) -> int:
return len(self._tools)
def __contains__(self, name: str) -> bool:
return name in self._tools
__all__ = [
"ToolRegistry",
"get_builtin_tools",
]
|