Enforce scoped tool access policies

Add one project/profile-aware resolver for canonical local, plugin, and MCP tool identities.

Apply it to text prompts, Responses schemas, connector stubs, local execution, MCP invocation, and delegated agents while preserving the response and vision invariants owned by the runtime.
This commit is contained in:
Alessandro 2026-08-05 10:53:49 +02:00
parent 641df22868
commit eec18ad839
23 changed files with 1116 additions and 52 deletions

View file

@ -14,6 +14,8 @@
- Preserve ordering where sections depend on earlier context.
- Keep secret-related prompt sections masked and scoped.
- Prompt additions must be bounded and compatible with tool-call contracts.
- Discover local tool prompts through `helpers.subagents.get_paths` and apply
`helpers.tool_policy` before including their text.
## Work Guidance

View file

@ -2,7 +2,7 @@ import os
from typing import Any
from helpers.extension import Extension, extensible
from helpers import files, subagents
from helpers import files, subagents, tool_policy
from helpers.print_style import PrintStyle
from agent import Agent, LoopData
@ -41,7 +41,9 @@ async def build_prompt(agent: Agent) -> str:
basename = os.path.basename(tool_file)
extra = all_tool_kwargs.get(basename, {})
tool = agent.read_prompt(basename, **extra)
tools.append(tool)
tool = tool_policy.filter_tool_prompt(agent, basename, tool)
if tool:
tools.append(tool)
except Exception as e:
PrintStyle().error(f"Error loading tool '{tool_file}': {e}")

View file

@ -28,6 +28,6 @@ async def build_prompt(agent: Agent) -> str:
pre_progress = agent.context.log.progress
agent.context.log.set_progress("Collecting MCP tools")
tools = mcp_config.get_tools_prompt()
tools = mcp_config.get_tools_prompt(agent=agent)
agent.context.log.set_progress(pre_progress)
return tools

View file

@ -434,6 +434,10 @@ class MCPTool(Tool):
return message, additional
async def execute(self, **kwargs: Any):
from helpers.tool_policy import ensure_tool_allowed
if "." in self.name:
ensure_tool_allowed(self.agent, self.name)
error = ""
additional: dict[str, Any] | None = None
try:
@ -1141,7 +1145,7 @@ class MCPConfig(BaseModel):
tools.append({f"{server.name}.{tool['name']}": tool_copy})
return tools
def get_tools_prompt(self, server_name: str = "") -> str:
def get_tools_prompt(self, server_name: str = "", agent: Any | None = None) -> str:
"""Get a prompt for all tools"""
# just to wait for pending initialization
@ -1165,8 +1169,18 @@ class MCPConfig(BaseModel):
tools = server.get_tools()
for tool in tools:
qualified_name = f"{server_name}.{tool['name']}"
if agent is not None:
from helpers.tool_policy import canonical_mcp_id, resolve_tool
if not resolve_tool(
agent,
qualified_name,
canonical_id=canonical_mcp_id(qualified_name),
).allowed:
continue
prompt += (
f"\n### {server_name}.{tool['name']}:\n"
f"\n### {qualified_name}:\n"
f"{tool['description']}\n\n"
# f"#### Categories:\n"
# f"* kind: MCP Server Tool\n"
@ -1188,7 +1202,7 @@ class MCPConfig(BaseModel):
# f' "observations": ["..."],\n' # TODO: this should be a prompt file with placeholders
f' "thoughts": ["..."],\n'
# f' "reflection": ["..."],\n' # TODO: this should be a prompt file with placeholders
f" \"tool_name\": \"{server_name}.{tool['name']}\",\n"
f" \"tool_name\": \"{qualified_name}\",\n"
f' "tool_args": !follow schema above\n'
f"}}\n"
)

View file

@ -82,6 +82,8 @@
- Project-scoped MCP servers overlay global servers by normalized name. The resulting `MCPConfig` cache key is derived from both config strings so project instances refresh when either scope changes.
- Server status and detail responses include `scope`, and MCP tools resolve through `MCPConfig.get_for_agent(agent)` before execution.
- MCP tool names are qualified as `server_name.tool_name`; server names are normalized without dots, and the tool portion may contain dots.
- Agent-facing MCP prompt descriptions filter through the central profile tool
policy, and `MCPTool.execute()` rechecks the same policy before invocation.
- `MCPConfig.get_tool()` tries the supplied qualified name first, then restores an advertised Responses alias from the calling agent's name map; names that still do not identify an MCP tool return `None` unchanged for downstream local-tool resolution.
- Servers may define `disabled_tools` as a list of MCP tool names. Disabled tools are omitted from agent-facing prompts, status counts, `has_tool`, and calls, while detail views can still retrieve them through `get_all_tools()` with a `disabled` flag so users can re-enable them.
- Server-specific `init_timeout` and `tool_timeout` override global MCP client timeout settings for list-tools and call-tool operations.
@ -110,6 +112,7 @@
- Run targeted tests for changed helper behavior; run security regressions for auth, filesystem, WebSocket, tunnel, upload, or secret-handling helpers.
- Related tests observed by source search:
- `tests/test_mcp_handler_multimodal.py`
- `tests/test_tool_policy.py`
## Child DOX Index

View file

@ -220,6 +220,17 @@ def get_plugin_roots(plugin_name: str = "") -> List[str]:
]
def get_plugin_name_from_path(path: str | Path) -> str:
"""Return the plugin directory name for a path under a canonical plugin root."""
candidate = Path(path).absolute()
for root in get_plugin_roots():
try:
return candidate.relative_to(Path(root).absolute()).parts[0]
except (IndexError, ValueError):
continue
return ""
def get_plugins_list():
if cached := cache.get(PLUGINS_LIST_CACHE_AREA, ""):
return cached

View file

@ -21,6 +21,7 @@
- `refresh_plugin_modules(plugin_names: list[str] | None=...)`
- `clear_plugin_cache(plugin_names: list[str] | None=...)`
- `get_plugin_roots(plugin_name: str=...) -> List[str]`: Plugin root directories, ordered by priority (user first).
- `get_plugin_name_from_path(path: str | Path) -> str`: Return the plugin directory name only for paths below a canonical user or bundled plugin root.
- `get_plugins_list()`
- `get_enhanced_plugins_list(custom: bool=..., builtin: bool=..., plugin_names: list[str] | None=...) -> List[PluginListItem]`: Discover plugins by directory convention. First root wins on ID conflict.
- `get_custom_plugins_updates(plugin_names: list[str] | None=...) -> List[PluginUpdateInfo]`
@ -54,7 +55,7 @@
## Key Concepts
- Important called helpers/classes observed in the source: `re.compile`, `Field`, `watchdog.add_watchdog`, `clear_plugin_cache`, `send_frontend_reload_notification`, `DeferredTask.start_task`, `get_plugin_roots`, `result.sort`, `cache.add`, `get_enhanced_plugins_list`, `find_plugin_dir`, `files.get_abs_path`, `files.exists`, `call_plugin_hook`, `delete_plugin`, `files.delete_dir`, `after_plugin_change`, `get_enabled_plugins`, `get_plugins_list`, `get_plugin_meta`.
- Important called helpers/classes observed in the source: `re.compile`, `Field`, `watchdog.add_watchdog`, `clear_plugin_cache`, `send_frontend_reload_notification`, `DeferredTask.start_task`, `get_plugin_roots`, `get_plugin_name_from_path`, `result.sort`, `cache.add`, `get_enhanced_plugins_list`, `find_plugin_dir`, `files.get_abs_path`, `files.exists`, `call_plugin_hook`, `delete_plugin`, `files.delete_dir`, `after_plugin_change`, `get_enabled_plugins`, `get_plugins_list`, `get_plugin_meta`.
- Keep request/response, tool, or helper semantics documented here at the same time as source changes.
## Work Guidance
@ -75,6 +76,7 @@
- `tests/test_document_query_plugin.py`
- `tests/test_error_retry_plugin.py`
- `tests/test_host_browser_connector.py`
- `tests/test_tool_policy.py`
## Child DOX Index

View file

@ -6,7 +6,7 @@ import os
import re
from typing import Any
from helpers import files, subagents
from helpers import files, subagents, tool_policy
FUNCTION_NAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
@ -35,18 +35,32 @@ def build_responses_function_tools(agent: Any) -> tuple[list[dict[str, Any]], di
name_map: dict[str, str] = {}
for tool_name, prompt in _local_tool_prompts(agent):
if not tool_policy.resolve_tool(agent, tool_name).allowed:
continue
native_name = _native_tool_name(tool_name)
name_map[native_name] = tool_name
tools.append(
{
"type": "function",
"name": native_name,
"description": _description_from_prompt(prompt, fallback=tool_name),
"description": _truncate(
tool_policy.tool_prompt_description(
prompt,
tool_name,
fallback=tool_name,
)
),
"parameters": _schema_from_prompt(prompt),
}
)
for tool_name, tool in _mcp_tools(agent):
if not tool_policy.resolve_tool(
agent,
tool_name,
canonical_id=tool_policy.canonical_mcp_id(tool_name),
).allowed:
continue
native_name = _native_tool_name(tool_name)
name_map[native_name] = tool_name
tools.append(
@ -124,7 +138,7 @@ def _mcp_tools(agent: Any) -> list[tuple[str, dict[str, Any]]]:
try:
import helpers.mcp_handler as mcp_helper
raw_tools = mcp_helper.MCPConfig.get_instance().get_tools()
raw_tools = mcp_helper.MCPConfig.get_for_agent(agent).get_tools()
except Exception:
return []
@ -139,7 +153,9 @@ def _mcp_tools(agent: Any) -> list[tuple[str, dict[str, Any]]]:
def _tool_name_from_prompt_basename(basename: str) -> str:
if not basename.startswith(TOOL_PROMPT_PREFIX) or not basename.endswith(TOOL_PROMPT_SUFFIX):
if not basename.startswith(TOOL_PROMPT_PREFIX) or not basename.endswith(
TOOL_PROMPT_SUFFIX
):
return ""
name = basename[len(TOOL_PROMPT_PREFIX) : -len(TOOL_PROMPT_SUFFIX)]
if not name or name in {"tools", "tools_vision"}:
@ -187,25 +203,6 @@ def _native_tool_name(tool_name: str) -> str:
return native[:64]
def _description_from_prompt(prompt: str, *, fallback: str) -> str:
for match in TOOL_DECLARATION_PATTERN.finditer(prompt or ""):
if match.group(1) == fallback:
return _truncate(match.group(2))
in_fence = False
for raw_line in (prompt or "").splitlines():
line = raw_line.strip()
if line.startswith(("```", "~~~")):
in_fence = not in_fence
continue
if in_fence or not line:
continue
if line.startswith("#"):
continue
return _truncate(line)
return fallback
def _schema_from_prompt(prompt: str) -> dict[str, Any]:
schema = _schema_from_embedded_json(prompt)
if schema:
@ -226,8 +223,7 @@ def _schema_from_embedded_json(prompt: str) -> dict[str, Any]:
if index == -1:
return {}
tail = prompt[index + len(marker) :].strip()
match = re.search(r"\{(?:[^{}]|(?R))*\}", tail, flags=re.DOTALL) if hasattr(re, "VERSION1") else None
candidate = match.group(0) if match else _balanced_json_object(tail)
candidate = _balanced_json_object(tail)
if not candidate:
return {}
try:

View file

@ -13,12 +13,19 @@
## Local Contracts
- Build local function tools from enabled `agent.system.tool.*.md` prompt files and include `vision_load` only when the active chat model enables the matching vision prompt.
- Discover local prompt files through `helpers.subagents.get_paths`; this module
owns the Responses-specific prompt-name compatibility rules.
- Local prompt-derived function names use existing bullet declarations that pair a backticked name with `arg` or `args` for multi-tool prompt files, otherwise prefer explicit `"tool_name"` examples, then the first prompt heading, and finally the prompt filename.
- Apply registered tool-prompt render kwargs before deriving native metadata so descriptions never expose unresolved prompt templates.
- Use an explicitly embedded JSON input schema when present. Infer only an unambiguous single backticked argument on an otherwise empty `args:` line; all other local tools receive an honest permissive object schema instead of prose-guessed types.
- Native local-tool descriptions use the matching compact multi-tool declaration or the first prose line, not a duplicate copy of the full tool manual or fenced examples.
- Native local-tool descriptions reuse the tool catalog's compact prompt
description; Responses retains native-name mapping, schema derivation, and
provider description limits.
- Preserve original Agent Zero tool names through the native Responses name map.
- Keep MCP tool schemas merged after local prompt-derived tools.
- Apply `helpers.tool_policy` before emitting local or MCP schemas; a blocked
capability is absent from provider-native tool definitions. Vision remains
controlled solely by the active chat model configuration.
- Connector remote tools are advertised only when `_a0_connector` runtime metadata says the matching connected CLI capability is currently available.
## Work Guidance

323
helpers/tool_policy.py Normal file
View file

@ -0,0 +1,323 @@
from __future__ import annotations
from dataclasses import dataclass
import os
import re
from typing import Any
from helpers import files, plugins, subagents
from helpers.errors import RepairableException
PLUGIN_NAME = "_tool_access"
PROMPT_PREFIX = "agent.system.tool."
PROMPT_SUFFIX = ".md"
NON_CONFIGURABLE_TOOLS = frozenset({"response", "vision_load"})
@dataclass(frozen=True)
class ToolPolicyDecision:
allowed: bool
tool_id: str
source: str
mode: str
reason: str = ""
def normalize_policy(config: Any) -> dict[str, Any]:
raw = dict(config) if isinstance(config, dict) else {}
mode = str(raw.get("mode") or "inherit").strip().lower()
default = str(raw.get("default") or "allow").strip().lower()
raw["mode"] = "custom" if mode == "custom" else "inherit"
raw["default"] = "block" if default == "block" else "allow"
raw["allowed"] = _normalize_ids(raw.get("allowed"))
raw["blocked"] = _normalize_ids(raw.get("blocked"))
return raw
def get_policy(agent: Any) -> dict[str, Any]:
return normalize_policy(plugins.get_plugin_config(PLUGIN_NAME, agent=agent))
def get_tool_catalog(agent: Any) -> list[dict[str, Any]]:
tool_paths = _local_tool_paths(agent)
descriptions = _tool_descriptions(agent, set(tool_paths))
catalog: list[dict[str, Any]] = []
seen: set[str] = set()
for name, tool_path in tool_paths.items():
if name in NON_CONFIGURABLE_TOOLS:
continue
tool_id, origin = _canonical_from_path(tool_path, name)
if tool_id in seen:
continue
seen.add(tool_id)
catalog.append(
{
"id": tool_id,
"name": name,
"label": name.replace("_", " ").title(),
"origin": origin,
"description": descriptions.get(name, ""),
"available": True,
}
)
try:
from helpers.mcp_handler import MCPConfig
for item in MCPConfig.get_for_agent(agent).get_tools():
qualified, tool = next(iter(item.items()))
tool_id = canonical_mcp_id(qualified)
if tool_id in seen:
continue
seen.add(tool_id)
catalog.append(
{
"id": tool_id,
"name": qualified,
"label": str(tool.get("name") or qualified),
"description": str(tool.get("description") or ""),
"origin": f"MCP · {str(tool.get('server') or '').strip()}",
"available": True,
}
)
except Exception:
pass
policy = get_policy(agent)
for tool_id in [*policy["allowed"], *policy["blocked"]]:
if (
tool_id in seen
or _tool_name_from_id(tool_id) in NON_CONFIGURABLE_TOOLS
):
continue
seen.add(tool_id)
name = _tool_name_from_id(tool_id)
catalog.append(
{
"id": tool_id,
"name": name,
"label": name.replace("_", " ").title(),
"description": "",
"origin": "Unavailable",
"available": False,
}
)
catalog.sort(key=lambda item: (item["label"].casefold(), item["id"]))
return catalog
def canonical_mcp_id(tool_name: str) -> str:
server, separator, name = str(tool_name or "").partition(".")
return f"mcp:{server}:{name}" if separator and server and name else ""
def _canonical_tool_id(agent: Any, tool_name: str) -> str:
if mcp_id := canonical_mcp_id(tool_name):
try:
from helpers.mcp_handler import MCPConfig
if MCPConfig.get_for_agent(agent).has_tool(tool_name):
return mcp_id
except Exception:
pass
paths = subagents.get_paths(agent, "tools", f"{tool_name}.py")
path = next((candidate for candidate in paths if files.exists(candidate)), "")
return _canonical_from_path(path, tool_name)[0] if path else f"local:{tool_name}"
def resolve_tool(
agent: Any,
tool_name: str,
*,
canonical_id: str = "",
) -> ToolPolicyDecision:
tool_id = canonical_id or _canonical_tool_id(agent, tool_name)
requested = str(tool_name or "").strip()
name = _tool_name_from_id(tool_id) if requested == tool_id else requested
if name in NON_CONFIGURABLE_TOOLS:
source = "framework-required" if name == "response" else "runtime-config"
return ToolPolicyDecision(True, tool_id, source, "invariant")
policy = get_policy(agent)
if policy["mode"] != "custom":
return ToolPolicyDecision(True, tool_id, "inherited", "inherit")
if tool_id in policy["blocked"]:
return ToolPolicyDecision(
False, tool_id, "scoped-policy", "custom", "blocked explicitly"
)
if tool_id in policy["allowed"]:
return ToolPolicyDecision(True, tool_id, "scoped-policy", "custom")
is_allowed = policy["default"] == "allow"
return ToolPolicyDecision(
is_allowed,
tool_id,
"scoped-default",
"custom",
"blocked by default" if not is_allowed else "",
)
def ensure_tool_allowed(agent: Any, tool_name: str) -> ToolPolicyDecision:
decision = resolve_tool(
agent,
tool_name,
canonical_id=canonical_mcp_id(tool_name),
)
if decision.allowed:
return decision
profile = str(getattr(getattr(agent, "config", None), "profile", "") or "default")
raise RepairableException(
f'Tool "{tool_name}" is blocked for agent profile "{profile}".'
)
def filter_tool_prompt(agent: Any, prompt_file: str, prompt: str) -> str:
known_names = _policy_tool_names(agent)
names = _prompt_tool_names(prompt_file, prompt, known_names)
if names and not any(resolve_tool(agent, name).allowed for name in names):
return ""
blocked_names = {
name
for name in known_names
if not resolve_tool(agent, name).allowed
}
if not blocked_names:
return prompt
patterns = [
re.compile(
rf"(?:`{re.escape(name)}`|[\"']{re.escape(name)}[\"']|"
rf"(?<![A-Za-z0-9_-]){re.escape(name)}\s+tool\b)",
re.IGNORECASE,
)
for name in sorted(blocked_names, key=len, reverse=True)
]
return "".join(
line
for line in prompt.splitlines(keepends=True)
if not any(pattern.search(line) for pattern in patterns)
)
def _local_tool_paths(agent: Any) -> dict[str, str]:
result: dict[str, str] = {}
for path in files.get_unique_filenames_in_dirs(
subagents.get_paths(agent, "tools"), "*.py"
):
name = os.path.splitext(os.path.basename(path))[0]
if name not in {"__init__", "unknown"}:
result[name] = path
return result
def _policy_tool_names(agent: Any) -> set[str]:
names = set(_local_tool_paths(agent))
policy = get_policy(agent)
names.update(
_tool_name_from_id(tool_id)
for tool_id in [*policy["allowed"], *policy["blocked"]]
if not tool_id.startswith("mcp:")
)
return names
def _prompt_tool_names(
prompt_file: str, prompt: str, known_names: set[str]
) -> list[str]:
fallback = _prompt_name(prompt_file)
declared = [
name for name in sorted(known_names) if _prompt_declares_tool(prompt, name)
]
if fallback in known_names:
return list(dict.fromkeys([fallback, *declared]))
return declared or ([fallback] if fallback else [])
def _prompt_declares_tool(prompt: str, name: str) -> bool:
escaped = re.escape(name)
return bool(
re.search(
rf"^\s{{0,3}}#{{1,6}}\s+`?{escaped}`?(?:\s|:|$)",
prompt or "",
re.IGNORECASE | re.MULTILINE,
)
or re.search(
rf"^\s*-\s+`{escaped}`\s*:",
prompt or "",
re.IGNORECASE | re.MULTILINE,
)
)
def _prompt_name(prompt_file: str) -> str:
basename = os.path.basename(prompt_file)
if basename.startswith(PROMPT_PREFIX) and basename.endswith(PROMPT_SUFFIX):
return basename[len(PROMPT_PREFIX) : -len(PROMPT_SUFFIX)]
return ""
def _tool_descriptions(agent: Any, tool_names: set[str]) -> dict[str, str]:
descriptions: dict[str, str] = {}
prompt_files = files.get_unique_filenames_in_dirs(
subagents.get_paths(agent, "prompts"), f"{PROMPT_PREFIX}*{PROMPT_SUFFIX}"
)
for prompt_file in prompt_files:
try:
prompt = agent.read_prompt(os.path.basename(prompt_file))
except Exception:
continue
for name in _prompt_tool_names(prompt_file, prompt, tool_names):
if name in tool_names and name not in descriptions:
descriptions[name] = tool_prompt_description(prompt, name)[:512]
return descriptions
def _canonical_from_path(path: str, name: str) -> tuple[str, str]:
if plugin_id := plugins.get_plugin_name_from_path(path):
return f"plugin:{plugin_id}:{name}", f"Plugin · {plugin_id}"
return f"local:{name}", "Agent Zero"
def _normalize_ids(raw: Any) -> list[str]:
if not isinstance(raw, list):
return []
result: list[str] = []
for value in raw:
tool_id = str(value or "").strip()
if tool_id and tool_id not in result:
result.append(tool_id)
return result
def tool_prompt_description(
prompt: str,
name: str,
*,
fallback: str = "",
) -> str:
declaration = re.search(
rf"^\s*-\s+`{re.escape(name)}`:\s+(args?\b.*)$",
prompt or "",
re.IGNORECASE | re.MULTILINE,
)
if declaration:
return declaration.group(1).strip()
in_fence = False
for raw_line in (prompt or "").splitlines():
line = raw_line.strip()
if line.startswith(("```", "~~~")):
in_fence = not in_fence
continue
if in_fence or not line or line.startswith("#"):
continue
return line
return fallback or name.replace("_", " ").strip().capitalize()
def _tool_name_from_id(tool_id: str) -> str:
return str(tool_id or "").rsplit(":", 1)[-1]

View file

@ -0,0 +1,52 @@
# tool_policy.py DOX
## Purpose
- Own the single project/profile-aware tool policy used by catalogs, prompts, native
schemas, local execution, MCP invocation, and delegated agents.
## Ownership
- `normalize_policy` owns the sparse allow/block configuration shape.
- `get_tool_catalog` owns canonical local, plugin, and MCP identities plus
unavailable-policy retention; local entries come from executable `tools/*.py`
files in the runtime path hierarchy. Catalog entries describe tools; the
editor applies the current draft policy instead of receiving duplicated
allowed/required flags from the backend.
- `tool_prompt_description` owns the shared compact description extracted for
the editor catalog and provider-native schemas; transport-specific names and
schemas remain with their transports.
- `resolve_tool` returns the effective decision and provenance.
- `ensure_tool_allowed` raises the stable repairable runtime policy error.
- `filter_tool_prompt` removes denied local capabilities from the text protocol
without taking ownership of provider-native naming rules.
## Runtime Contracts
- Scoped config resolution is delegated to `helpers.plugins`: active project
profile, active project, user profile, bundled/plugin profile, then default.
- Missing policy inherits standard access; custom policy always records whether
future tools default to allowed or blocked.
- The `response` capability is a framework-required invariant: profile policy
cannot disable it, and the editor does not list it as a configurable tool.
- `vision_load` remains owned by the active chat model's vision configuration;
it is not exposed as a profile-policy choice and legacy policy IDs cannot
suppress the chat-configured capability.
- Policy IDs are namespaced as `local:`, `plugin:<id>:`, or `mcp:<server>:`.
- Plugin IDs are derived relative to the canonical roots from `helpers.plugins`,
not by independently parsing repository-relative path strings.
- Each executable local tool has its own policy identity, including tools that
share one Markdown prompt.
- Catalog descriptions call the supplied agent's prompt loader instead of
opening prompt files through a parallel path; the editor agent intentionally
keeps its existing raw, no-processor implementation.
- Unknown policy IDs remain in the catalog as unavailable entries.
- Resolution performs no model calls and logs no secrets.
## Verification
- Run `tests/test_tool_policy.py` and the prompt/Responses/MCP focused tests.
## Child DOX Index
No child DOX files.

View file

@ -99,6 +99,7 @@ Direct child DOX files:
| [_telegram_integration/AGENTS.md](_telegram_integration/AGENTS.md) | Telegram bot integration and per-user chat sessions. |
| [_text_editor/AGENTS.md](_text_editor/AGENTS.md) | Native text read, write, and patch tool. |
| [_time_travel/AGENTS.md](_time_travel/AGENTS.md) | Workspace history, diff, travel, snapshot, and revert flows. |
| [_tool_access/AGENTS.md](_tool_access/AGENTS.md) | Always-on project/profile tool-policy execution gate. |
| [_whatsapp_integration/AGENTS.md](_whatsapp_integration/AGENTS.md) | WhatsApp Baileys bridge integration. |
| [_whats_new/AGENTS.md](_whats_new/AGENTS.md) | Version-gated What's New showcase modal, card list, and startup trigger. |
| [_whisper_stt/AGENTS.md](_whisper_stt/AGENTS.md) | Whisper speech-to-text integration. |

View file

@ -20,6 +20,8 @@
prompts, remote file metadata enables `text_editor_remote`, F4-enabled remote
execution metadata enables `code_execution_remote`, and supported enabled
Computer Use that does not need re-arming enables `computer_use_remote`.
- Never re-add a connector prompt that the effective project/profile tool policy
blocks.
- Do not bypass WebSocket authentication or leak connector session data.
- Advertise Launcher gateways additively through HTTP capability
`launcher_gateway` and WebSocket feature `launcher_gateway_control`. Older

View file

@ -4,17 +4,13 @@ import re
from typing import Any
from helpers.extension import Extension
from helpers.tool_policy import resolve_tool
from plugins._a0_connector.helpers.remote_tool_prompts import (
REMOTE_TOOL_PROMPTS,
remote_tool_prompt_availability,
)
_TOOL_MARKERS = {
tool_name: f'"tool_name": "{tool_name}"' for tool_name in REMOTE_TOOL_PROMPTS
}
class IncludeRemoteToolStubs(Extension):
def execute(self, data: dict[str, Any] = {}, **kwargs: Any) -> None:
if self.agent is None:
@ -40,8 +36,8 @@ class IncludeRemoteToolStubs(Extension):
if not prompt:
continue
if available.get(tool_name):
marker = _TOOL_MARKERS[tool_name]
if available.get(tool_name) and resolve_tool(self.agent, tool_name).allowed:
marker = f'"tool_name": "{tool_name}"'
if marker not in result:
result = f"{result.rstrip()}\n\n{prompt}"
continue
@ -55,12 +51,4 @@ def _remove_prompt(result: str, prompt: str) -> str:
if prompt not in result:
return result
for needle, replacement in (
(f"\n\n{prompt}\n\n", "\n\n"),
(f"\n\n{prompt}", ""),
(f"{prompt}\n\n", ""),
(prompt, ""),
):
result = result.replace(needle, replacement)
return re.sub(r"\n{3,}", "\n\n", result).rstrip()
return re.sub(r"\n{3,}", "\n\n", result.replace(prompt, "")).rstrip()

View file

@ -0,0 +1,27 @@
# Tool Access Plugin DOX
## Purpose
- Own the always-enabled project/profile tool-policy configuration and execution gate.
## Ownership
- `helpers/tool_policy.py` owns shared resolution and catalog behavior.
- `hooks.py` normalizes scoped configuration.
- `extensions/python/tool_execute_before/` rejects blocked execution.
## Local Contracts
- This plugin has no independent settings UI; the Agent Editor writes sparse
profile `config.json` files, projects may own project or project-profile
configs through the standard plugin scope paths, and the runtime remains
authoritative.
- Required final-response capability is never disabled.
## Verification
- Run `tests/test_tool_policy.py`.
## Child DOX Index
No child DOX files.

View file

@ -0,0 +1,11 @@
# Tool Access
Tool Access is the always-enabled runtime owner for Agent Editor tool policy.
Policies use the standard plugin precedence: active project profile, active
project, user profile, bundled/plugin profile, then the default. Sparse project
policy lives under `.a0proj/plugins/_tool_access/config.json`; profile policy
lives under `usr/agents/<profile>/plugins/_tool_access/config.json`.
One resolver filters textual prompts and provider-native schemas, rejects local
and MCP execution, and keeps delegated agents bound to their own effective scope.
The required final-response capability is always available.

View file

@ -0,0 +1,4 @@
mode: inherit
default: allow
allowed: []
blocked: []

View file

@ -0,0 +1,8 @@
from helpers.extension import Extension
from helpers.tool_policy import ensure_tool_allowed
class EnforceToolPolicy(Extension):
async def execute(self, tool_name: str = "", **kwargs) -> None:
if self.agent and tool_name:
ensure_tool_allowed(self.agent, tool_name)

View file

@ -0,0 +1,9 @@
from helpers.tool_policy import normalize_policy
def get_plugin_config(default=None, **kwargs):
return normalize_policy(default)
def save_plugin_config(settings=None, **kwargs):
return normalize_policy(settings)

View file

@ -0,0 +1,7 @@
name: _tool_access
title: Tool Access
description: Enforces sparse project and agent tool visibility and execution policy.
version: 1.0.0
always_enabled: true
per_project_config: true
per_agent_config: true

View file

@ -3,6 +3,7 @@ import sys
import time
import uuid
from pathlib import Path
from types import SimpleNamespace
import yaml
@ -13,7 +14,11 @@ if str(PROJECT_ROOT) not in sys.path:
def _restore_real_helpers_package() -> None:
helpers_module = sys.modules.get("helpers")
if helpers_module is None or getattr(helpers_module, "__file__", ""):
if (
helpers_module is None
or getattr(helpers_module, "__file__", "")
or list(getattr(helpers_module, "__path__", []))
):
return
for name in list(sys.modules):
@ -65,10 +70,14 @@ class FakeContext:
def __init__(self, context_id: str):
self.id = context_id
def get_data(self, key: str, recursive: bool = True):
return None
class FakeAgent:
def __init__(self, context_id: str):
self.context = FakeContext(context_id)
self.config = SimpleNamespace(profile="default")
def read_prompt(self, file: str, **kwargs) -> str:
text = (PROMPT_ROOT / file).read_text(encoding="utf-8")
@ -218,6 +227,24 @@ def test_remote_tool_gate_appends_available_prompt_when_standard_prompt_missing(
_assert_remote_tool_absent(prompt, "computer_use_remote")
def test_remote_tool_gate_does_not_readd_a_policy_blocked_prompt(monkeypatch):
context_id = _context_id()
sid = _sid()
monkeypatch.setitem(
IncludeRemoteToolStubs.execute.__globals__,
"resolve_tool",
lambda _agent, name: SimpleNamespace(allowed=name != "text_editor_remote"),
)
ws_runtime.register_sid(sid)
ws_runtime.store_sid_remote_file_metadata(sid, {"enabled": True})
try:
prompt = _apply_gate(context_id, include_standard_remote_prompts=False)
finally:
ws_runtime.unregister_sid(sid)
_assert_remote_tool_absent(prompt, "text_editor_remote")
def test_responses_function_tools_follow_remote_prompt_gate(monkeypatch):
from helpers import responses_tools

View file

@ -1,18 +1,21 @@
import sys
from pathlib import Path
from types import SimpleNamespace
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from helpers import responses_tools
from helpers import responses_tools, tool_policy
class FakeAgent:
def __init__(self, prompt_root: Path, data=None):
self.prompt_root = prompt_root
self.data = data or {}
self.config = SimpleNamespace(profile="default")
self.context = SimpleNamespace(get_data=lambda *args, **kwargs: None)
def read_prompt(self, file: str, **kwargs) -> str:
prompt = (self.prompt_root / file).read_text(encoding="utf-8")
@ -170,7 +173,11 @@ def test_response_tool_native_contract_omits_wrapper_and_exposes_text():
encoding="utf-8"
)
description = responses_tools._description_from_prompt(prompt, fallback="response")
description = tool_policy.tool_prompt_description(
prompt,
"response",
fallback="response",
)
schema = responses_tools._schema_from_prompt(prompt)
assert description == "final answer to user"
@ -259,3 +266,14 @@ def test_local_tool_prompts_use_registered_render_kwargs(monkeypatch, tmp_path):
assert "{{default_line_count}}" not in prompts["text_editor"]
assert "read 200 lines by default" in prompts["text_editor"]
def test_explicit_tool_name_precedes_a_generic_heading():
prompt = """## memory tools
durable memory operations
{"tool_name": "memory_load", "tool_args": {}}
"""
assert responses_tools._tool_names_from_prompt(
prompt, fallback="memory"
) == ["memory_load"]

550
tests/test_tool_policy.py Normal file
View file

@ -0,0 +1,550 @@
from __future__ import annotations
from pathlib import Path
from types import SimpleNamespace
import pytest
from extensions.python.system_prompt import _11_tools_prompt
from helpers import mcp_handler, responses_tools, tool_policy
from helpers.errors import RepairableException
from plugins._tool_access.extensions.python.tool_execute_before._10_enforce_tool_policy import (
EnforceToolPolicy,
)
class _Context:
def get_data(self, key: str, recursive: bool = True):
return None
class _Agent:
def __init__(self, prompt_root: Path, profile: str = "researcher") -> None:
self.prompt_root = prompt_root
self.config = SimpleNamespace(profile=profile)
self.context = _Context()
self.data: dict = {}
def read_prompt(self, basename: str, **kwargs) -> str:
content = (self.prompt_root / basename).read_text(encoding="utf-8")
for key, value in kwargs.items():
content = content.replace("{{" + key + "}}", str(value))
return content
def get_data(self, key: str):
return self.data.get(key)
class _NoMCPTools:
def get_tools(self):
return []
def _write_prompt(root: Path, basename: str, content: str) -> None:
(root / basename).write_text(content.strip() + "\n", encoding="utf-8")
def _prompt_paths(root: Path):
def get_paths(agent, *parts, **kwargs):
return [str(root)] if parts and parts[0] == "prompts" else []
return get_paths
def _custom_policy(*, default: str, allowed=(), blocked=()):
return {
"mode": "custom",
"default": default,
"allowed": list(allowed),
"blocked": list(blocked),
}
@pytest.fixture
def local_prompt_agent(monkeypatch, tmp_path: Path) -> _Agent:
_write_prompt(tmp_path, "agent.system.tools.md", "TOOLS\n{{tools}}")
_write_prompt(
tmp_path,
"agent.system.tool.allowed.md",
"""### allowed
Allowed description
Keyboard input remains documented.
Do not call the `blocked` tool from here.
{"tool_name":"allowed","tool_args":{}}""",
)
_write_prompt(
tmp_path,
"agent.system.tool.blocked.md",
'### blocked\nBlocked description\n{"tool_name":"blocked","tool_args":{}}',
)
monkeypatch.setattr(tool_policy.subagents, "get_paths", _prompt_paths(tmp_path))
monkeypatch.setattr(
"plugins._model_config.helpers.model_config.get_chat_model_config",
lambda agent: {"vision": False},
)
monkeypatch.setattr(responses_tools, "_mcp_tools", lambda agent: [])
return _Agent(tmp_path)
@pytest.mark.asyncio
async def test_text_tool_prompt_omits_blocked_tool_and_description(
monkeypatch, local_prompt_agent: _Agent
) -> None:
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: _custom_policy(default="allow", blocked=["local:blocked"]),
)
prompt = await _11_tools_prompt.build_prompt(local_prompt_agent)
assert "Allowed description" in prompt
assert "Keyboard input remains documented." in prompt
assert "Do not call" not in prompt
assert "blocked" not in prompt.lower()
assert "Blocked description" not in prompt
def test_provider_native_schemas_omit_blocked_local_tool(
monkeypatch, local_prompt_agent: _Agent
) -> None:
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: _custom_policy(default="allow", blocked=["local:blocked"]),
)
tools, _name_map = responses_tools.build_responses_function_tools(
local_prompt_agent
)
assert [tool["name"] for tool in tools] == ["allowed"]
def test_required_response_survives_default_block(monkeypatch, tmp_path: Path) -> None:
_write_prompt(
tmp_path,
"agent.system.tool.response.md",
'### response\nfinal answer\n{"tool_name":"response","tool_args":{"text":"done"}}',
)
monkeypatch.setattr(tool_policy.subagents, "get_paths", _prompt_paths(tmp_path))
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: _custom_policy(default="block", blocked=["local:response"]),
)
monkeypatch.setattr(
mcp_handler.MCPConfig,
"get_for_agent",
lambda agent: _NoMCPTools(),
)
agent = _Agent(tmp_path)
decision = tool_policy.resolve_tool(agent, "response")
assert decision.allowed is True
assert decision.source == "framework-required"
assert tool_policy.get_tool_catalog(agent) == []
def test_catalog_comes_from_executable_tools_not_prompt_names(
monkeypatch, tmp_path: Path
) -> None:
prompt_root = tmp_path / "prompts"
tool_root = tmp_path / "tools"
prompt_root.mkdir()
tool_root.mkdir()
_write_prompt(
prompt_root,
"agent.system.tool.actual.md",
"### actual\nActual description",
)
_write_prompt(
prompt_root,
"agent.system.tool.prompt_only.md",
"### prompt_only\nNo executable implementation",
)
(tool_root / "actual.py").write_text("class Actual: pass\n", encoding="utf-8")
(tool_root / "response.py").write_text("class Response: pass\n", encoding="utf-8")
def get_paths(agent, *parts, **kwargs):
if parts[0] == "prompts":
return [str(prompt_root)]
if len(parts) == 1:
return [str(tool_root)]
return [str(tool_root / parts[1])]
monkeypatch.setattr(tool_policy.subagents, "get_paths", get_paths)
monkeypatch.setattr(
mcp_handler.MCPConfig,
"get_for_agent",
lambda agent: _NoMCPTools(),
)
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: {
"mode": "inherit",
"default": "allow",
"allowed": [],
"blocked": [],
},
)
catalog = tool_policy.get_tool_catalog(_Agent(prompt_root))
assert [item["id"] for item in catalog] == ["local:actual"]
assert catalog[0]["description"] == "Actual description"
def test_tool_prompt_description_skips_fenced_examples() -> None:
prompt = """### example
~~~json
{"tool_name":"example","tool_args":{}}
~~~
Visible summary
"""
assert tool_policy.tool_prompt_description(prompt, "example") == "Visible summary"
def test_plugin_tool_identity_uses_canonical_plugin_roots(
monkeypatch, tmp_path: Path
) -> None:
plugin_root = tmp_path / "plugins" / "_example"
plugin_tool = plugin_root / "tools" / "actual.py"
plugin_tool.parent.mkdir(parents=True)
plugin_tool.write_text("class Actual: pass\n", encoding="utf-8")
monkeypatch.setattr(
tool_policy.plugins,
"get_plugin_roots",
lambda: [str(tmp_path / "usr" / "plugins"), str(tmp_path / "plugins")],
)
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: {
"mode": "inherit",
"default": "allow",
"allowed": [],
"blocked": [],
},
)
agent = _Agent(tmp_path)
monkeypatch.setattr(
tool_policy.subagents,
"get_paths",
lambda *args, **kwargs: [str(plugin_tool)],
)
assert tool_policy.resolve_tool(agent, "actual").tool_id == "plugin:_example:actual"
lookalike = tmp_path / "work" / "plugins" / "_example" / "tools" / "actual.py"
lookalike.parent.mkdir(parents=True)
lookalike.write_text("class Actual: pass\n", encoding="utf-8")
monkeypatch.setattr(
tool_policy.subagents,
"get_paths",
lambda *args, **kwargs: [str(lookalike)],
)
assert tool_policy.resolve_tool(agent, "actual").tool_id == "local:actual"
def test_legacy_response_and_vision_policy_ids_stay_out_of_catalog(
monkeypatch, tmp_path: Path
) -> None:
monkeypatch.setattr(tool_policy.subagents, "get_paths", lambda *args, **kwargs: [])
monkeypatch.setattr(
mcp_handler.MCPConfig,
"get_for_agent",
lambda agent: _NoMCPTools(),
)
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: _custom_policy(
default="allow",
blocked=[
"response",
"local:response",
"plugin:legacy:response",
"vision_load",
"local:vision_load",
"plugin:legacy:vision_load",
],
),
)
assert tool_policy.get_tool_catalog(_Agent(tmp_path)) == []
@pytest.mark.asyncio
async def test_vision_tool_follows_chat_config_not_profile_policy(
monkeypatch, tmp_path: Path
) -> None:
_write_prompt(tmp_path, "agent.system.tools.md", "TOOLS\n{{tools}}")
_write_prompt(
tmp_path,
"agent.system.tools_vision.md",
'### vision_load\nload images\n{"tool_name":"vision_load","tool_args":{"paths":[]}}',
)
monkeypatch.setattr(tool_policy.subagents, "get_paths", _prompt_paths(tmp_path))
monkeypatch.setattr(
"plugins._model_config.helpers.model_config.get_chat_model_config",
lambda agent: {"vision": True},
)
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: _custom_policy(
default="block", blocked=["local:vision_load"]
),
)
monkeypatch.setattr(responses_tools, "_mcp_tools", lambda agent: [])
agent = _Agent(tmp_path)
prompt = await _11_tools_prompt.build_prompt(agent)
schemas, _name_map = responses_tools.build_responses_function_tools(agent)
assert "vision_load" in prompt
assert [schema["name"] for schema in schemas] == ["vision_load"]
assert tool_policy.resolve_tool(agent, "vision_load").source == "runtime-config"
def test_mcp_prompt_and_native_schema_omit_blocked_tool(
monkeypatch, tmp_path: Path
) -> None:
class Server:
name = "docs"
description = "Documentation"
def get_tools(self):
return [
{
"name": "read",
"description": "Read docs",
"input_schema": {"type": "object"},
},
{
"name": "write",
"description": "Write docs",
"input_schema": {"type": "object"},
},
]
config = mcp_handler.MCPConfig(servers_list=[])
config.servers = [Server()]
agent = _Agent(tmp_path)
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: _custom_policy(
default="allow", blocked=["mcp:docs:write"]
),
)
monkeypatch.setattr(
responses_tools,
"_mcp_tools",
lambda agent: [
("docs.read", Server().get_tools()[0]),
("docs.write", Server().get_tools()[1]),
],
)
monkeypatch.setattr(tool_policy.subagents, "get_paths", lambda *args: [])
monkeypatch.setattr(responses_tools, "_vision_tool_prompt", lambda agent: "")
prompt = config.get_tools_prompt(agent=agent)
schemas, name_map = responses_tools.build_responses_function_tools(agent)
assert "docs.read" in prompt
assert "docs.write" not in prompt
assert len(schemas) == 1
assert name_map[schemas[0]["name"]] == "docs.read"
@pytest.mark.asyncio
async def test_local_execution_gate_returns_stable_profile_error(
monkeypatch, tmp_path: Path
) -> None:
agent = _Agent(tmp_path, profile="researcher")
monkeypatch.setattr(tool_policy.subagents, "get_paths", lambda *args, **kwargs: [])
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: _custom_policy(default="block"),
)
with pytest.raises(
RepairableException,
match='Tool "shell" is blocked for agent profile "researcher"',
):
await EnforceToolPolicy(agent).execute(tool_name="shell")
@pytest.mark.asyncio
async def test_mcp_invocation_rechecks_policy_before_server_call(
monkeypatch, tmp_path: Path
) -> None:
agent = _Agent(tmp_path, profile="researcher")
called = False
class Config:
async def call_tool(self, name, kwargs):
nonlocal called
called = True
raise AssertionError("blocked MCP call reached the server")
monkeypatch.setattr(mcp_handler.MCPConfig, "get_for_agent", lambda agent: Config())
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: _custom_policy(
default="allow", blocked=["mcp:docs:write"]
),
)
tool = mcp_handler.MCPTool(
agent=agent,
name="docs.write",
method=None,
args={},
message="",
loop_data=None,
)
with pytest.raises(RepairableException, match='Tool "docs.write" is blocked'):
await tool.execute()
assert called is False
@pytest.mark.asyncio
async def test_delegated_agent_uses_its_own_profile_policy_at_execution_gate(
monkeypatch, tmp_path: Path
) -> None:
parent = _Agent(tmp_path, profile="agent0")
child = _Agent(tmp_path, profile="researcher")
monkeypatch.setattr(tool_policy.subagents, "get_paths", lambda *args, **kwargs: [])
def config_for_profile(plugin_name, agent=None, **kwargs):
if agent.config.profile == "researcher":
return _custom_policy(default="block")
return {"mode": "inherit"}
monkeypatch.setattr(tool_policy.plugins, "get_plugin_config", config_for_profile)
assert tool_policy.resolve_tool(parent, "shell").allowed is True
assert tool_policy.resolve_tool(child, "shell").allowed is False
await EnforceToolPolicy(parent).execute(tool_name="shell")
with pytest.raises(
RepairableException,
match='Tool "shell" is blocked for agent profile "researcher"',
):
await EnforceToolPolicy(child).execute(tool_name="shell")
def test_project_policy_precedes_profile_policy(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
class ProjectContext:
def get_data(self, key: str, recursive: bool = True):
return "demo" if key == "project" else None
monkeypatch.setattr(tool_policy.files, "_base_dir", str(tmp_path))
monkeypatch.setattr(tool_policy.subagents, "get_paths", lambda *args, **kwargs: [])
monkeypatch.setattr(
tool_policy.plugins,
"call_plugin_hook",
lambda _plugin, _hook, default=None, **_kwargs: default,
)
agent = _Agent(tmp_path)
agent.context = ProjectContext()
tool_policy.plugins.save_plugin_config(
tool_policy.PLUGIN_NAME,
"",
"researcher",
_custom_policy(default="block"),
)
tool_policy.plugins.save_plugin_config(
tool_policy.PLUGIN_NAME,
"demo",
"",
_custom_policy(default="allow"),
)
tool_policy.plugins.save_plugin_config(
tool_policy.PLUGIN_NAME,
"demo",
"researcher",
_custom_policy(default="allow", blocked=["local:shell"]),
)
profile_path = Path(
tool_policy.plugins.determine_plugin_asset_path(
tool_policy.PLUGIN_NAME,
"",
"researcher",
tool_policy.plugins.CONFIG_FILE_NAME,
)
)
project_path = Path(
tool_policy.plugins.determine_plugin_asset_path(
tool_policy.PLUGIN_NAME,
"demo",
"",
tool_policy.plugins.CONFIG_FILE_NAME,
)
)
project_profile_path = Path(
tool_policy.plugins.determine_plugin_asset_path(
tool_policy.PLUGIN_NAME,
"demo",
"researcher",
tool_policy.plugins.CONFIG_FILE_NAME,
)
)
decision = tool_policy.resolve_tool(agent, "shell")
assert decision.allowed is False
assert decision.source == "scoped-policy"
project_profile_path.unlink()
decision = tool_policy.resolve_tool(agent, "shell")
assert decision.allowed is True
assert decision.source == "scoped-default"
project_path.unlink()
decision = tool_policy.resolve_tool(agent, "shell")
assert decision.allowed is False
assert decision.source == "scoped-default"
assert profile_path.is_file()
def test_unknown_policy_ids_are_retained_as_unavailable(
monkeypatch, tmp_path: Path
) -> None:
agent = _Agent(tmp_path)
monkeypatch.setattr(tool_policy.subagents, "get_paths", lambda *args, **kwargs: [])
monkeypatch.setattr(
mcp_handler.MCPConfig,
"get_for_agent",
lambda agent: _NoMCPTools(),
)
monkeypatch.setattr(
tool_policy,
"get_policy",
lambda agent: _custom_policy(
default="allow", blocked=["plugin:missing:ghost"]
),
)
catalog = tool_policy.get_tool_catalog(agent)
assert catalog == [
{
"id": "plugin:missing:ghost",
"name": "ghost",
"label": "Ghost",
"description": "",
"origin": "Unavailable",
"available": False,
}
]