Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions src/strands/hooks/events.py
Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,12 @@ class BeforeToolCallEvent(HookEvent, _Interruptible):
cancel_tool: A user defined message that when set, will cancel the tool call.
The message will be placed into a tool result with an error status. If set to `True`, Strands will cancel
the tool call and use a default cancel message.
tool_is_read_only: Convenience property. True if the selected tool is read-only, False otherwise
(including when selected_tool is None).
tool_is_destructive: Convenience property. True if the selected tool is destructive, False otherwise
(including when selected_tool is None).
tool_requires_confirmation: Convenience property. True if the selected tool requires confirmation,
False otherwise (including when selected_tool is None).
"""

selected_tool: AgentTool | None
Expand All @@ -157,6 +163,21 @@ class BeforeToolCallEvent(HookEvent, _Interruptible):
def _can_write(self, name: str) -> bool:
return name in ["cancel_tool", "selected_tool", "tool_use"]

@property
def tool_is_read_only(self) -> bool:
"""Whether the selected tool only reads state. False when selected_tool is None."""
return self.selected_tool is not None and self.selected_tool.is_read_only

@property
def tool_is_destructive(self) -> bool:
"""Whether the selected tool performs irreversible actions. False when selected_tool is None."""
return self.selected_tool is not None and self.selected_tool.is_destructive

@property
def tool_requires_confirmation(self) -> bool:
"""Whether the selected tool requires user confirmation. False when selected_tool is None."""
return self.selected_tool is not None and self.selected_tool.requires_confirmation

@override
def _interrupt_id(self, name: str) -> str:
"""Unique id for the interrupt.
Expand Down
64 changes: 62 additions & 2 deletions src/strands/tools/decorator.py
Original file line number Diff line number Diff line change
Expand Up @@ -459,6 +459,10 @@ def __init__(
tool_spec: ToolSpec,
tool_func: Callable[P, R],
metadata: FunctionToolMetadata,
*,
read_only: bool = False,
destructive: bool = False,
requires_confirmation: bool = False,
):
"""Initialize the decorated function tool.

Expand All @@ -467,13 +471,19 @@ def __init__(
tool_spec: The tool specification containing metadata for Agent integration.
tool_func: The original function being decorated.
metadata: The FunctionToolMetadata object with extracted function information.
read_only: Whether this tool only reads state without modification.
destructive: Whether this tool performs irreversible actions.
requires_confirmation: Whether this tool should require user confirmation before execution.
"""
super().__init__()

self._tool_name = tool_name
self._tool_spec = tool_spec
self._tool_func = tool_func
self._metadata = metadata
self._read_only = read_only
self._destructive = destructive
self._requires_confirmation = requires_confirmation

functools.update_wrapper(wrapper=self, wrapped=self._tool_func)

Expand Down Expand Up @@ -506,7 +516,15 @@ def my_tool():
if instance is not None and not inspect.ismethod(self._tool_func):
# Create a bound method
tool_func = self._tool_func.__get__(instance, instance.__class__)
return DecoratedFunctionTool(self._tool_name, self._tool_spec, tool_func, self._metadata)
return DecoratedFunctionTool(
self._tool_name,
self._tool_spec,
tool_func,
self._metadata,
read_only=self._read_only,
destructive=self._destructive,
requires_confirmation=self._requires_confirmation,
)

return self

Expand Down Expand Up @@ -577,6 +595,24 @@ def tool_type(self) -> str:
"""
return "function"

@property
@override
def is_read_only(self) -> bool:
"""Whether this tool only reads state without modification."""
return self._read_only

@property
@override
def is_destructive(self) -> bool:
"""Whether this tool performs irreversible actions."""
return self._destructive

@property
@override
def requires_confirmation(self) -> bool:
"""Whether this tool should require user confirmation before execution."""
return self._requires_confirmation

@override
async def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
"""Stream the tool with a tool use specification.
Expand Down Expand Up @@ -725,6 +761,9 @@ def tool(
inputSchema: JSONSchema | None = None,
name: str | None = None,
context: bool | str = False,
read_only: bool = False,
destructive: bool = False,
requires_confirmation: bool = False,
) -> Callable[[Callable[P, R]], DecoratedFunctionTool[P, R]]: ...
# Suppressing the type error because we want callers to be able to use both `tool` and `tool()` at the
# call site, but the actual implementation handles that and it's not representable via the type-system
Expand All @@ -734,6 +773,9 @@ def tool( # type: ignore
inputSchema: JSONSchema | None = None,
name: str | None = None,
context: bool | str = False,
read_only: bool = False,
destructive: bool = False,
requires_confirmation: bool = False,
) -> DecoratedFunctionTool[P, R] | Callable[[Callable[P, R]], DecoratedFunctionTool[P, R]]:
"""Decorator that transforms a Python function into a Strands tool.

Expand Down Expand Up @@ -762,6 +804,10 @@ def tool( # type: ignore
context: When provided, places an object in the designated parameter. If True, the param name
defaults to 'tool_context', or if an override is needed, set context equal to a string to designate
the param name.
read_only: Whether this tool only reads state without modification. Defaults to False.
destructive: Whether this tool performs irreversible actions. Defaults to False.
requires_confirmation: Whether this tool should require user confirmation before execution.
Defaults to False.

Returns:
An AgentTool that also mimics the original function when invoked
Expand Down Expand Up @@ -816,13 +862,27 @@ def decorator(f: T) -> "DecoratedFunctionTool[P, R]":
tool_spec["description"] = description
if inputSchema is not None:
tool_spec["inputSchema"] = inputSchema
if read_only:
tool_spec["readOnly"] = True
if destructive:
tool_spec["destructive"] = True
if requires_confirmation:
tool_spec["requiresConfirmation"] = True

tool_name = tool_spec.get("name", f.__name__)

if not isinstance(tool_name, str):
raise ValueError(f"Tool name must be a string, got {type(tool_name)}")

return DecoratedFunctionTool(tool_name, tool_spec, f, tool_meta)
return DecoratedFunctionTool(
tool_name,
tool_spec,
f,
tool_meta,
read_only=read_only,
destructive=destructive,
requires_confirmation=requires_confirmation,
)

# Handle both @tool and @tool() syntax
if func is None:
Expand Down
31 changes: 31 additions & 0 deletions src/strands/tools/mcp/mcp_agent_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,10 @@ def __init__(
mcp_client: "MCPClient",
name_override: str | None = None,
timeout: timedelta | None = None,
*,
read_only: bool | None = None,
destructive: bool | None = None,
requires_confirmation: bool | None = None,
) -> None:
"""Initialize a new MCPAgentTool instance.

Expand All @@ -44,13 +48,22 @@ def __init__(
name_override: Optional name to use for the agent tool (for disambiguation)
If None, uses the original MCP tool name
timeout: Optional timeout duration for tool execution
read_only: Override for read-only classification. When None, falls back to the
tool spec's ``readOnly`` field if present, otherwise False.
destructive: Override for destructive classification. When None, falls back to the
tool spec's ``destructive`` field if present, otherwise False.
requires_confirmation: Override for confirmation requirement. When None, falls back
to the tool spec's ``requiresConfirmation`` field if present, otherwise False.
"""
super().__init__()
logger.debug("tool_name=<%s> | creating mcp agent tool", mcp_tool.name)
self.mcp_tool = mcp_tool
self.mcp_client = mcp_client
self._agent_tool_name = name_override or mcp_tool.name
self.timeout = timeout
self._read_only_override = read_only
self._destructive_override = destructive
self._requires_confirmation_override = requires_confirmation

@property
def tool_name(self) -> str:
Expand Down Expand Up @@ -93,6 +106,24 @@ def tool_type(self) -> str:
"""
return "python"

@property
@override
def is_read_only(self) -> bool:
"""Whether this tool only reads state. Set via constructor override."""
return self._read_only_override if self._read_only_override is not None else False

@property
@override
def is_destructive(self) -> bool:
"""Whether this tool performs irreversible actions. Set via constructor override."""
return self._destructive_override if self._destructive_override is not None else False

@property
@override
def requires_confirmation(self) -> bool:
"""Whether this tool requires user confirmation. Set via constructor override."""
return self._requires_confirmation_override if self._requires_confirmation_override is not None else False

@override
async def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
"""Stream the MCP tool.
Expand Down
22 changes: 22 additions & 0 deletions src/strands/tools/registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -270,6 +270,8 @@ def register_tool(self, tool: AgentTool) -> None:
" Cannot add a duplicate tool which differs by a '-' or '_'"
)

self._validate_security_metadata(tool)

# Register in main registry
self.registry[tool.tool_name] = tool

Expand All @@ -288,6 +290,24 @@ def register_tool(self, tool: AgentTool) -> None:
list(self.dynamic_tools.keys()),
)

def _validate_security_metadata(self, tool: AgentTool) -> None:
"""Validate that a tool's security metadata is internally consistent.

Args:
tool: The tool to validate.

Raises:
ValueError: If the tool has contradictory security metadata.
"""
if tool.is_read_only and tool.is_destructive:
raise ValueError(f"Tool '{tool.tool_name}' cannot be both read_only and destructive")

if tool.is_destructive and not tool.requires_confirmation:
logger.warning(
"tool_name=<%s> | tool is marked destructive but does not require confirmation",
tool.tool_name,
)

def replace(self, new_tool: AgentTool) -> None:
"""Replace an existing tool with a new implementation.

Expand All @@ -305,6 +325,8 @@ def replace(self, new_tool: AgentTool) -> None:
if tool_name not in self.registry:
raise ValueError(f"Cannot replace tool '{tool_name}' - tool does not exist")

self._validate_security_metadata(new_tool)

# Update main registry
self.registry[tool_name] = new_tool

Expand Down
18 changes: 18 additions & 0 deletions src/strands/tools/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -240,6 +240,24 @@ def tool_type(self) -> str:
"""
return "python"

@property
@override
def is_read_only(self) -> bool:
"""Whether this tool only reads state, derived from its ToolSpec."""
return self._tool_spec.get("readOnly") is True

@property
@override
def is_destructive(self) -> bool:
"""Whether this tool performs irreversible actions, derived from its ToolSpec."""
return self._tool_spec.get("destructive") is True

@property
@override
def requires_confirmation(self) -> bool:
"""Whether this tool requires user confirmation, derived from its ToolSpec."""
return self._tool_spec.get("requiresConfirmation") is True

@override
async def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
"""Stream the Python function with the given tool use request.
Expand Down
34 changes: 34 additions & 0 deletions src/strands/types/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,12 +30,19 @@ class ToolSpec(TypedDict):
outputSchema: Optional JSON Schema defining the expected output format.
Note: Not all model providers support this field. Providers that don't
support it should filter it out before sending to their API.
readOnly: Optional flag indicating the tool only reads state without modification.
destructive: Optional flag indicating the tool performs irreversible actions.
requiresConfirmation: Optional flag indicating the tool should require user
confirmation before execution.
"""

description: str
inputSchema: JSONSchema
name: str
outputSchema: NotRequired[JSONSchema]
readOnly: NotRequired[bool]
destructive: NotRequired[bool]
requiresConfirmation: NotRequired[bool]


class Tool(TypedDict):
Expand Down Expand Up @@ -255,6 +262,33 @@ def supports_hot_reload(self) -> bool:
"""
return False

@property
def is_read_only(self) -> bool:
"""Whether this tool only reads state without modification.

Returns:
False by default. Override in subclasses or set via @tool(read_only=True).
"""
return False

@property
def is_destructive(self) -> bool:
"""Whether this tool performs irreversible actions.

Returns:
False by default. Override in subclasses or set via @tool(destructive=True).
"""
return False

@property
def requires_confirmation(self) -> bool:
"""Whether this tool should require user confirmation before execution.

Returns:
False by default. Override in subclasses or set via @tool(requires_confirmation=True).
"""
return False

@abstractmethod
# pragma: no cover
def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
Expand Down
3 changes: 3 additions & 0 deletions tests/strands/tools/mcp/test_mcp_client_tool_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,9 @@ def create_mock_tool(tool_name: str, mcp_tool_name: str | None = None) -> MagicM
tool.mcp_tool = MagicMock(spec=MCPTool)
tool.mcp_tool.name = mcp_tool_name or tool_name
tool.mcp_tool.description = f"Description for {tool_name}"
tool.is_read_only = False
tool.is_destructive = False
tool.requires_confirmation = False
return tool


Expand Down
Loading