"""Tool inventory assembly and visibility helpers for GroundingAgent.""" from __future__ import annotations import copy import json import os from typing import Any, Iterable from openspace.grounding.core.types import BackendType from openspace.services.tooling.context import ToolUseContext from openspace.utils.logging import Logger logger = Logger.get_logger(__name__) def tool_set_signature(tools: list[Any]) -> str: normalized_tools: list[dict[str, Any]] = [] for tool in tools: schema = getattr(tool, "schema", None) runtime_info = getattr(tool, "runtime_info", None) backend = getattr(getattr(tool, "backend_type", None), "value", None) if runtime_info is not None: backend = getattr(getattr(runtime_info, "backend", None), "value", backend) normalized_tools.append( { "name": getattr(tool, "name", None), "aliases": list(getattr(tool, "aliases", []) or []), "description": getattr(tool, "description", None), "parameters": getattr(schema, "parameters", None), "backend": backend, "server_name": getattr(runtime_info, "server_name", None), "is_bound": bool(getattr(tool, "is_bound", False)), "is_deferred": bool(getattr(tool, "is_deferred", False)), } ) normalized_tools.sort( key=lambda item: ( str(item.get("name") or ""), str(item.get("backend") or ""), str(item.get("server_name") or ""), ) ) return json.dumps(normalized_tools, sort_keys=True, default=str) def resolve_tui_available(agent: Any, context: dict[str, Any]) -> bool: explicit = context.get("tui_available") if explicit is not None: return bool(explicit) bridge = context.get("tui_bridge") if bridge is None: bridge = getattr(agent, "_tui_bridge", None) if bridge is None: return False proxy_owner = getattr(bridge, "_owner", None) if proxy_owner is not None and hasattr(proxy_owner, "_tui_bridge"): return bool(getattr(proxy_owner, "_tui_bridge")) bridge_owner = getattr(bridge, "__self__", None) if bridge_owner is not None and hasattr(bridge_owner, "_tui_bridge"): return bool(getattr(bridge_owner, "_tui_bridge")) return True def resolve_async_agent(context: dict[str, Any]) -> bool: explicit = context.get("is_async_agent") if explicit is not None: return bool(explicit) return str(context.get("agent_id") or "primary") != "primary" def resolve_permission_context( *, cwd: str, permission_mode: str | None, context: dict[str, Any], ) -> Any: explicit_permission_context = context.get("permission_context") if explicit_permission_context is not None: return explicit_permission_context try: from openspace.grounding.core.permissions.loader import ( load_tool_permission_context, ) return load_tool_permission_context(cwd, permission_mode) except Exception as exc: logger.error( "Failed to load tool permission context for %s (%s)", cwd, permission_mode, exc_info=True, ) raise RuntimeError( "Failed to load tool permission settings for this workspace" ) from exc def filter_tools_for_permission_mode( tools: list[Any], tool_use_context: ToolUseContext, ) -> list[Any]: if str(getattr(tool_use_context, "permission_mode", "") or "") != "plan": return list(tools) from openspace.agents.agent_tool_utils import EXIT_PLAN_MODE_TOOL_NAME from openspace.tool_runtime.pipeline.execution import ( FILE_EDIT_TOOL_NAME, FILE_WRITE_TOOL_NAME, NOTEBOOK_EDIT_TOOL_NAME, ) allowed_write_tools = { FILE_EDIT_TOOL_NAME, FILE_WRITE_TOOL_NAME, NOTEBOOK_EDIT_TOOL_NAME, EXIT_PLAN_MODE_TOOL_NAME, } filtered: list[Any] = [] for tool in tools: name = str(getattr(tool, "name", "") or "") if name in allowed_write_tools: filtered.append(tool) continue try: if bool(tool.is_read_only({})): filtered.append(tool) except Exception: continue return filtered def append_todo_write_tool(tools: list[Any]) -> None: from openspace.agents.agent_tool_utils import TODO_WRITE_TOOL_NAME, tool_matches_name from openspace.tools.todo_tool import TodoWriteTool if any(tool_matches_name(tool, TODO_WRITE_TOOL_NAME) for tool in tools): return tool = TodoWriteTool() if not tool.is_enabled(): return tool.bind_runtime_info(backend=BackendType.META, session_name="agent") tools.append(tool) def append_sleep_and_brief_tools(tools: list[Any]) -> None: from openspace.agents.agent_tool_utils import tool_matches_name from openspace.tools.brief_tool import BRIEF_TOOL_NAME, BriefTool from openspace.tools.sleep_tool import SLEEP_TOOL_NAME, SleepTool for name, tool_cls in ( (SLEEP_TOOL_NAME, SleepTool), (BRIEF_TOOL_NAME, BriefTool), ): if any(tool_matches_name(tool, name) for tool in tools): continue tool = tool_cls() tool.bind_runtime_info(backend=BackendType.META, session_name="agent") tools.append(tool) def append_schedule_cron_tools(tools: list[Any]) -> None: from openspace.agents.agent_tool_utils import tool_matches_name from openspace.tools.schedule_cron_tool import ( SCHEDULE_CRON_CREATE_TOOL_NAME, SCHEDULE_CRON_DELETE_TOOL_NAME, SCHEDULE_CRON_LIST_TOOL_NAME, ScheduleCronCreateTool, ScheduleCronDeleteTool, ScheduleCronListTool, ) for name, tool_cls in ( (SCHEDULE_CRON_CREATE_TOOL_NAME, ScheduleCronCreateTool), (SCHEDULE_CRON_DELETE_TOOL_NAME, ScheduleCronDeleteTool), (SCHEDULE_CRON_LIST_TOOL_NAME, ScheduleCronListTool), ): if any(tool_matches_name(tool, name) for tool in tools): continue tool = tool_cls() tool.bind_runtime_info(backend=BackendType.META, session_name="agent") tools.append(tool) def get_agent_definition_registry(agent: Any) -> Any: if getattr(agent, "_agent_definition_registry", None) is None: from openspace.agents.agent_definitions import AgentDefinitionRegistry agent._agent_definition_registry = AgentDefinitionRegistry() return agent._agent_definition_registry def resolve_agent_definitions( agent: Any, context: dict[str, Any], tools: list[Any], ) -> Any: existing = context.get("agent_definitions") if existing is not None: return existing registry = get_agent_definition_registry(agent) workspace_dir = context.get("workspace_dir") if workspace_dir: try: from openspace.agents.agent_definitions import AgentSource registry.load_custom_agents( os.path.join(str(workspace_dir), ".openspace", "agents"), source=AgentSource.PROJECT_SETTINGS, ) except Exception: logger.debug("Failed to load project agent definitions", exc_info=True) return registry.result( allowed_agent_types=context.get("allowed_agent_types"), available_mcp_servers=mcp_servers_from_tools(tools), ) def with_agent_tool( agent: Any, tools: list[Any], *, context: dict[str, Any], agent_definitions: Any, ) -> list[Any]: scope_names = { str(getattr(item, "value", item)) for item in getattr(agent, "_backend_scope", []) } if "meta" not in scope_names: return list(tools) from openspace.agents.agent_tool_utils import AGENT_TOOL_NAME, tool_matches_name result = list(tools) if not any(tool_matches_name(tool, AGENT_TOOL_NAME) for tool in result): from openspace.agents.agent_tool import build_agent_tool tool = build_agent_tool( registry=get_agent_definition_registry(agent), parent_agent=agent, grounding_client=agent.grounding_client, llm_client=getattr(agent, "_llm_client", None), orchestrator=getattr(agent, "_multi_agent_orchestrator", None), ) result.append(tool) append_multi_agent_control_tools(result) append_ask_user_question_tool(result) append_todo_write_tool(result) append_plan_mode_tools(result) append_config_tool(result) append_lsp_tool(result) append_sleep_and_brief_tools(result) append_schedule_cron_tools(result) context["agent_definitions"] = agent_definitions return result def should_append_agent_tools(agent: Any, context: dict[str, Any]) -> bool: explicit = context.get("agent_tools_enabled") if explicit is not None: return bool(explicit) grounding_client = agent.grounding_client return grounding_client is not None and hasattr(grounding_client, "list_tools") def append_config_tool(tools: list[Any]) -> None: from openspace.agents.agent_tool_utils import CONFIG_TOOL_NAME, tool_matches_name from openspace.tools.config_tool import ConfigTool if any(tool_matches_name(tool, CONFIG_TOOL_NAME) for tool in tools): return tool = ConfigTool() if not tool.is_enabled(): return tool.bind_runtime_info(backend=BackendType.META, session_name="agent") tools.append(tool) def append_lsp_tool(tools: list[Any]) -> None: from openspace.agents.agent_tool_utils import tool_matches_name from openspace.tools.lsp_tool import LSP_TOOL_NAME, LSPTool if any(tool_matches_name(tool, LSP_TOOL_NAME) for tool in tools): return tool = LSPTool() tool.bind_runtime_info(backend=BackendType.META, session_name="agent") tools.append(tool) def append_plan_mode_tools(tools: list[Any]) -> None: from openspace.agents.agent_tool_utils import ( ENTER_PLAN_MODE_TOOL_NAME, EXIT_PLAN_MODE_TOOL_NAME, tool_matches_name, ) from openspace.tools.plan_mode_tools import EnterPlanModeTool, ExitPlanModeTool existing = list(tools) for name, tool_cls in ( (EXIT_PLAN_MODE_TOOL_NAME, ExitPlanModeTool), (ENTER_PLAN_MODE_TOOL_NAME, EnterPlanModeTool), ): if any(tool_matches_name(tool, name) for tool in existing): continue tool = tool_cls() tool.bind_runtime_info(backend=BackendType.META, session_name="agent") tools.append(tool) existing.append(tool) def append_ask_user_question_tool(tools: list[Any]) -> None: from openspace.agents.agent_tool_utils import ASK_USER_QUESTION_TOOL_NAME, tool_matches_name from openspace.tools.ask_user_tool import AskUserQuestionTool if any(tool_matches_name(tool, ASK_USER_QUESTION_TOOL_NAME) for tool in tools): return tool = AskUserQuestionTool() tool.bind_runtime_info(backend=BackendType.META, session_name="agent") tools.append(tool) def append_multi_agent_control_tools(tools: list[Any]) -> None: from openspace.agents.agent_tool_utils import tool_matches_name from openspace.tools.task_tools import ( TASK_GET_TOOL_NAME, TASK_LIST_TOOL_NAME, TASK_STOP_TOOL_NAME, TaskGetTool, TaskListTool, TaskStopTool, ) from openspace.tools.team_tools import ( SEND_MESSAGE_TOOL_NAME, TEAM_CREATE_TOOL_NAME, TEAM_DELETE_TOOL_NAME, SendMessageTool, TeamCreateTool, TeamDeleteTool, ) existing = list(tools) for name, tool_cls in ( (TASK_GET_TOOL_NAME, TaskGetTool), (TASK_LIST_TOOL_NAME, TaskListTool), (TASK_STOP_TOOL_NAME, TaskStopTool), (SEND_MESSAGE_TOOL_NAME, SendMessageTool), (TEAM_CREATE_TOOL_NAME, TeamCreateTool), (TEAM_DELETE_TOOL_NAME, TeamDeleteTool), ): if any(tool_matches_name(tool, name) for tool in existing): continue tool = tool_cls() tool.bind_runtime_info(backend=BackendType.META, session_name="agent") tools.append(tool) existing.append(tool) def bind_agent_tools_to_context( tools: Iterable[Any], tool_use_context: ToolUseContext, ) -> None: from openspace.agents.agent_tool_utils import ( AGENT_TOOL_NAME, CONFIG_TOOL_NAME, ENTER_PLAN_MODE_TOOL_NAME, EXIT_PLAN_MODE_TOOL_NAME, SEND_MESSAGE_TOOL_NAME, TASK_GET_TOOL_NAME, TASK_LIST_TOOL_NAME, TASK_STOP_TOOL_NAME, TODO_WRITE_TOOL_NAME, tool_matches_name, ) from openspace.tools.brief_tool import BRIEF_TOOL_NAME from openspace.tools.lsp_tool import LSP_TOOL_NAME from openspace.tools.schedule_cron_tool import ( SCHEDULE_CRON_CREATE_TOOL_NAME, SCHEDULE_CRON_DELETE_TOOL_NAME, SCHEDULE_CRON_LIST_TOOL_NAME, ) from openspace.tools.sleep_tool import SLEEP_TOOL_NAME from openspace.tools.team_tools import TEAM_CREATE_TOOL_NAME, TEAM_DELETE_TOOL_NAME for tool in tools: if not ( tool_matches_name(tool, AGENT_TOOL_NAME) or tool_matches_name(tool, CONFIG_TOOL_NAME) or tool_matches_name(tool, TASK_GET_TOOL_NAME) or tool_matches_name(tool, TASK_LIST_TOOL_NAME) or tool_matches_name(tool, TASK_STOP_TOOL_NAME) or tool_matches_name(tool, SEND_MESSAGE_TOOL_NAME) or tool_matches_name(tool, TEAM_CREATE_TOOL_NAME) or tool_matches_name(tool, TEAM_DELETE_TOOL_NAME) or tool_matches_name(tool, TODO_WRITE_TOOL_NAME) or tool_matches_name(tool, SLEEP_TOOL_NAME) or tool_matches_name(tool, BRIEF_TOOL_NAME) or tool_matches_name(tool, SCHEDULE_CRON_CREATE_TOOL_NAME) or tool_matches_name(tool, SCHEDULE_CRON_DELETE_TOOL_NAME) or tool_matches_name(tool, SCHEDULE_CRON_LIST_TOOL_NAME) or tool_matches_name(tool, LSP_TOOL_NAME) or tool_matches_name(tool, ENTER_PLAN_MODE_TOOL_NAME) or tool_matches_name(tool, EXIT_PLAN_MODE_TOOL_NAME) ): continue setter = getattr(tool, "set_context", None) if callable(setter): setter(tool_use_context) def mcp_servers_from_tools(tools: Iterable[Any]) -> list[str]: servers: set[str] = set() for tool in tools: name = str(getattr(tool, "name", "") or "") if name.startswith("mcp__"): parts = name.split("__") if len(parts) >= 3 and parts[1]: servers.add(parts[1]) continue runtime = getattr(tool, "runtime_info", None) server_name = getattr(runtime, "server_name", None) if runtime else None if server_name: servers.add(str(server_name)) return sorted(servers) def find_tool_by_name( tool_name: str, *, tool_map: dict[str, Any] | None = None, tools: list[Any] | None = None, ) -> Any | None: if tool_map and tool_name in tool_map: return tool_map[tool_name] for tool in tools or []: aliases = getattr(tool, "aliases", []) or [] if getattr(tool, "name", None) == tool_name or tool_name in aliases: return tool return None def build_iteration_tool_results( *, tool_calls: list[dict[str, Any]], tool_map: dict[str, Any], result_messages: list[dict[str, Any]], tools: list[Any], ) -> list[dict[str, Any]]: tool_result_by_id: dict[str, dict[str, Any]] = {} for message in result_messages: if message.get("role") != "tool": continue tool_call_id = str(message.get("tool_call_id") or "") if not tool_call_id: continue tool_result_by_id.setdefault(tool_call_id, message) iteration_results: list[dict[str, Any]] = [] for tool_call in tool_calls: tool_call_id = str(tool_call.get("id") or "") tool_name = str((tool_call.get("function") or {}).get("name") or "unknown") result_message = tool_result_by_id.get(tool_call_id, {}) result_meta = result_message.get("_meta") if not isinstance(result_meta, dict): result_meta = {} tool_obj = find_tool_by_name(tool_name, tool_map=tool_map, tools=tools) runtime_info = getattr(tool_obj, "runtime_info", None) backend = getattr( getattr(runtime_info, "backend", None), "value", getattr(getattr(tool_obj, "backend_type", None), "value", None), ) status = str(result_meta.get("status") or "unknown") content = result_message.get("content") error = None if status in {"error", "denied", "cancelled"} and isinstance(content, str): error = content.removeprefix("Error: ").strip() or content iteration_results.append( { "tool_call": copy.deepcopy(tool_call), "tool_name": tool_name, "backend": backend, "server_name": getattr(runtime_info, "server_name", None), "status": status, "content": content, "error": error, "execution_time": result_meta.get("execution_time"), "metadata": copy.deepcopy(result_meta.get("tool_result_metadata") or {}), } ) return iteration_results def resolve_memory_mode(value: Any = None) -> str: try: from openspace.services.memory.daily_log import get_memory_mode return get_memory_mode(str(value) if value is not None else None) except Exception: raw = str(value or "").strip().lower() return "daily_log" if raw in {"daily_log", "daily-log", "dailylog", "logs"} else "direct" def with_memory_mode_tools(tools: list[Any], memory_mode: str) -> list[Any]: if memory_mode != "daily_log": return list(tools) selected = list(tools) if any(getattr(tool, "name", None) == "memory_log" for tool in selected): return selected try: from openspace.services.memory.daily_log import MemoryLogTool tool = MemoryLogTool() tool.bind_runtime_info( backend=BackendType.SHELL, session_name="memory", ) selected.append(tool) except Exception: logger.debug("Could not add memory_log tool for daily_log mode", exc_info=True) return selected def skills_disabled_for_context(context: dict[str, Any]) -> bool: return bool(context.get("skills_disabled")) def without_skill_protocol_tools(tools: Iterable[Any]) -> list[Any]: try: from openspace.skill_engine.protocol import ( DISCOVER_SKILLS_TOOL_NAME, SKILL_TOOL_NAME, tool_matches_name as skill_tool_matches_name, ) except Exception: return list(tools) filtered: list[Any] = [] for tool in tools: if ( skill_tool_matches_name(tool, SKILL_TOOL_NAME) or skill_tool_matches_name(tool, DISCOVER_SKILLS_TOOL_NAME) ): continue filtered.append(tool) return filtered def deferred_tool_names( tools: list[Any], *, discovered_tool_names: Iterable[str] = (), ) -> list[str]: discovered = {str(name) for name in discovered_tool_names} return sorted( { tool.name for tool in tools if getattr(tool, "is_deferred", False) and getattr(tool, "name", None) and tool.name not in discovered } ) def build_active_tools( tools: list[Any], *, discovered_tool_names: Iterable[str] = (), active_tool_names: Iterable[str] | None = None, deferred_tool_names: Iterable[str] | None = None, ) -> list[Any]: from openspace.grounding.core.tool_discovery import ( TOOL_DISCOVERY_TOOL_NAME, ToolSearchTool, ) discovered = {str(name) for name in discovered_tool_names} hard_active = ( {str(name) for name in active_tool_names} if active_tool_names is not None else None ) hard_deferred = {str(name) for name in (deferred_tool_names or ())} if hard_active is None: active = [ tool for tool in tools if not getattr(tool, "is_deferred", False) or getattr(tool, "name", None) in discovered ] remaining_deferred = [ tool for tool in tools if getattr(tool, "is_deferred", False) and getattr(tool, "name", None) not in discovered ] else: active = [ tool for tool in tools if ( getattr(tool, "name", None) in hard_active or getattr(tool, "name", None) in discovered ) ] remaining_deferred = [ tool for tool in tools if ( getattr(tool, "name", None) in hard_deferred and getattr(tool, "name", None) not in discovered ) ] active_names = {getattr(tool, "name", "") for tool in active} if remaining_deferred and TOOL_DISCOVERY_TOOL_NAME not in active_names: active.append(ToolSearchTool(all_tools=tools)) return active def scoped_tool_backends(agent: Any) -> list[BackendType]: return [BackendType(name) for name in getattr(agent, "_backend_scope", [])] def with_skill_protocol_tools( agent: Any, tools: list[Any], backends: list[BackendType], ) -> list[Any]: registry = getattr(agent, "_skill_registry", None) if not (registry and registry.list_skills()): return tools from openspace.skill_engine.protocol import ( DISCOVER_SKILLS_TOOL_NAME, SKILL_TOOL_NAME, DiscoverSkillsTool, SkillDiscoveryService, SkillTool, ) existing_names = {getattr(tool, "name", "") for tool in tools} if SKILL_TOOL_NAME not in existing_names: skill_tool = SkillTool( registry, skill_store=getattr(agent, "_skill_store", None), ) skill_tool.bind_runtime_info( backend=BackendType.META, session_name="internal", ) tools.append(skill_tool) existing_names.add(SKILL_TOOL_NAME) if ( getattr(agent, "_skill_discovery_enabled", True) and DISCOVER_SKILLS_TOOL_NAME not in existing_names ): discover_tool = DiscoverSkillsTool( SkillDiscoveryService( registry, store=getattr(agent, "_skill_store", None), llm_client=( getattr(agent, "_skill_selection_llm", None) or getattr(agent, "_tool_retrieval_llm", None) or getattr(agent, "_llm_client", None) ), ) ) discover_tool.bind_runtime_info( backend=BackendType.META, session_name="internal", ) tools.append(discover_tool) existing_names.add(DISCOVER_SKILLS_TOOL_NAME) logger.info("Added Skill Protocol tools") return tools async def get_tools_without_auto_preselection(agent: Any) -> list[Any]: grounding_client = agent.grounding_client if not grounding_client: return [] backends = scoped_tool_backends(agent) tools = await grounding_client.list_tools( backend=backends, use_cache=True, ) return with_skill_protocol_tools(agent, list(tools), backends) async def get_available_tools(agent: Any, task_description: str | None) -> list[Any]: grounding_client = agent.grounding_client if not grounding_client: return [] backends = scoped_tool_backends(agent) try: retrieval_llm = getattr(agent, "_tool_retrieval_llm", None) or getattr(agent, "_llm_client", None) tools = await grounding_client.get_tools_with_auto_preselection( task_description=task_description, backend=backends, use_cache=True, llm_callable=retrieval_llm, ) logger.info( "GroundingAgent selected %d tools (auto-preselection preload) from %d backends", len(tools), len(backends), ) except Exception as e: logger.warning("Auto-search tools failed, falling back to full list: %s", e) tools = await load_all_tools(agent, grounding_client) return with_skill_protocol_tools(agent, tools, backends) async def get_tool_universe(agent: Any, preselected_tools: list[Any]) -> list[Any]: grounding_client = agent.grounding_client if not grounding_client: return list(preselected_tools) backends = [BackendType(name) for name in getattr(agent, "_backend_scope", [])] try: tools = await grounding_client.list_tools( backend=backends, use_cache=True, ) except Exception as exc: logger.debug("Failed to load full tool universe; using preselected tools: %s", exc) tools = list(preselected_tools) seen = {getattr(tool, "name", "") for tool in tools} for tool in preselected_tools: name = getattr(tool, "name", "") if name and name not in seen: tools.append(tool) seen.add(name) return tools async def load_all_tools(agent: Any, grounding_client: Any) -> list[Any]: all_tools: list[Any] = [] for backend_name in getattr(agent, "_backend_scope", []): try: backend_type = BackendType(backend_name) tools = await grounding_client.list_tools(backend=backend_type) all_tools.extend(tools) logger.debug("Retrieved %d tools from backend: %s", len(tools), backend_name) except Exception as e: logger.debug("Could not get tools from %s: %s", backend_name, e) logger.info( "GroundingAgent fallback retrieved %d tools from %d backends", len(all_tools), len(getattr(agent, "_backend_scope", [])), ) return all_tools def build_retrieved_tools_list( tools: list[Any], preselection_debug_info: dict[str, Any] | None, ) -> list[dict[str, Any]]: retrieved: list[dict[str, Any]] = [] for tool in tools: tool_info: dict[str, Any] = { "name": getattr(tool, "name", str(tool)), "description": getattr(tool, "description", ""), } runtime_info = getattr(tool, "_runtime_info", None) if runtime_info and hasattr(runtime_info, "backend"): tool_info["backend"] = ( runtime_info.backend.value if hasattr(runtime_info.backend, "value") else str(runtime_info.backend) ) tool_info["server_name"] = runtime_info.server_name elif hasattr(tool, "backend_type"): tool_info["backend"] = ( tool.backend_type.value if hasattr(tool.backend_type, "value") else str(tool.backend_type) ) if preselection_debug_info and preselection_debug_info.get("tool_scores"): for score_info in preselection_debug_info["tool_scores"]: if score_info["name"] == tool_info["name"]: tool_info["similarity_score"] = score_info["score"] if "quality_adjusted_score" in score_info: tool_info["quality_adjusted_score"] = score_info[ "quality_adjusted_score" ] if "quality_penalty" in score_info: tool_info["quality_penalty"] = score_info["quality_penalty"] break retrieved.append(tool_info) return retrieved __all__ = [ "append_ask_user_question_tool", "append_config_tool", "append_lsp_tool", "append_multi_agent_control_tools", "append_plan_mode_tools", "append_schedule_cron_tools", "append_sleep_and_brief_tools", "append_todo_write_tool", "bind_agent_tools_to_context", "build_active_tools", "build_iteration_tool_results", "build_retrieved_tools_list", "deferred_tool_names", "filter_tools_for_permission_mode", "find_tool_by_name", "get_agent_definition_registry", "get_available_tools", "get_tool_universe", "get_tools_without_auto_preselection", "load_all_tools", "mcp_servers_from_tools", "resolve_agent_definitions", "resolve_async_agent", "resolve_memory_mode", "resolve_permission_context", "resolve_tui_available", "scoped_tool_backends", "should_append_agent_tools", "skills_disabled_for_context", "tool_set_signature", "with_agent_tool", "with_memory_mode_tools", "with_skill_protocol_tools", "without_skill_protocol_tools", ]