diff --git a/open-strata/README.md b/open-strata/README.md index be722f5cb..c2a8c0b94 100644 --- a/open-strata/README.md +++ b/open-strata/README.md @@ -198,6 +198,13 @@ When running as a router, the following tools are exposed: - `search_documentation` - Search server documentation - `handle_auth_failure` - Handle authentication issues +`discover_server_actions`, `get_action_details`, `execute_action`, and +`search_documentation` accept an optional `scope` argument. Use +`"scope": "read_only"` to only discover, inspect, search, or execute actions +whose upstream MCP tool is annotated with `readOnlyHint: true`. This is useful +when connecting user-owned integrations in contexts where the agent should be +able to inspect data but not mutate it. + # Development ### Running Tests @@ -243,4 +250,3 @@ strata run --port 8080 --- 👉 **[Get Instant Access at Klavis AI (YC X25)](https://klavis.ai/)** 👈 - diff --git a/open-strata/src/strata/mcp_proxy/client.py b/open-strata/src/strata/mcp_proxy/client.py index 677c9c8a4..044a11363 100644 --- a/open-strata/src/strata/mcp_proxy/client.py +++ b/open-strata/src/strata/mcp_proxy/client.py @@ -87,6 +87,16 @@ async def list_tools(self, use_cache: bool = True) -> List[Dict[str, Any]]: tool_dict["title"] = tool.title if hasattr(tool, "outputSchema") and tool.outputSchema: tool_dict["outputSchema"] = tool.outputSchema + if hasattr(tool, "annotations") and tool.annotations: + annotations = tool.annotations + if hasattr(annotations, "model_dump"): + tool_dict["annotations"] = annotations.model_dump( + exclude_none=True + ) + elif hasattr(annotations, "dict"): + tool_dict["annotations"] = annotations.dict(exclude_none=True) + elif isinstance(annotations, dict): + tool_dict["annotations"] = annotations tools.append(tool_dict) self._tools_cache = tools diff --git a/open-strata/src/strata/tools.py b/open-strata/src/strata/tools.py index 903eeaf76..2655d480b 100644 --- a/open-strata/src/strata/tools.py +++ b/open-strata/src/strata/tools.py @@ -18,6 +18,49 @@ TOOL_SEARCH_DOCUMENTATION = "search_documentation" TOOL_HANDLE_AUTH_FAILURE = "handle_auth_failure" +ACTION_SCOPE_ALL = "all" +ACTION_SCOPE_READ_ONLY = "read_only" +ACTION_SCOPES = [ACTION_SCOPE_ALL, ACTION_SCOPE_READ_ONLY] + + +def _scope_property() -> dict: + return { + "type": "string", + "enum": ACTION_SCOPES, + "default": ACTION_SCOPE_ALL, + "description": ( + "Limit actions by safety scope. Use read_only to expose or execute only " + "actions annotated with readOnlyHint." + ), + } + + +def _tool_annotations(tool: dict) -> dict: + annotations = tool.get("annotations") or {} + if isinstance(annotations, dict): + return annotations + if hasattr(annotations, "model_dump"): + return annotations.model_dump(exclude_none=True) + if hasattr(annotations, "dict"): + return annotations.dict(exclude_none=True) + return {} + + +def is_read_only_tool(tool: dict) -> bool: + annotations = _tool_annotations(tool) + return annotations.get("readOnlyHint") is True + + +def filter_tools_by_scope(tools: List[dict] | None, scope: str | None) -> List[dict]: + tool_list = list(tools or []) + if not scope or scope == ACTION_SCOPE_ALL: + return tool_list + if scope == ACTION_SCOPE_READ_ONLY: + return [tool for tool in tool_list if is_read_only_tool(tool)] + raise ValueError( + f"Invalid scope '{scope}'. Expected one of: {', '.join(ACTION_SCOPES)}" + ) + def get_tool_definitions(user_available_servers: List[str]) -> List[types.Tool]: """Get tool definitions for the available servers.""" @@ -38,6 +81,7 @@ def get_tool_definitions(user_available_servers: List[str]) -> List[types.Tool]: "items": {"type": "string", "enum": user_available_servers}, "description": "List of server names to discover actions from.", }, + "scope": _scope_property(), }, }, ), @@ -57,6 +101,7 @@ def get_tool_definitions(user_available_servers: List[str]) -> List[types.Tool]: "type": "string", "description": "The name of the action/operation", }, + "scope": _scope_property(), }, }, ), @@ -89,6 +134,7 @@ def get_tool_definitions(user_available_servers: List[str]) -> List[types.Tool]: "description": "JSON string containing request body", "default": "{}", }, + "scope": _scope_property(), }, }, ), @@ -115,6 +161,7 @@ def get_tool_definitions(user_available_servers: List[str]) -> List[types.Tool]: "maximum": 50, "default": 10, }, + "scope": _scope_property(), }, }, ), @@ -155,6 +202,7 @@ async def execute_tool( if name == TOOL_DISCOVER_SERVER_ACTIONS: user_query = arguments.get("user_query") server_names = arguments.get("server_names") + scope = arguments.get("scope", ACTION_SCOPE_ALL) # If no server names provided, use all available servers if not server_names: @@ -165,7 +213,7 @@ async def execute_tool( for server_name in server_names: try: client = client_manager.get_client(server_name) - tools = await client.list_tools() + tools = filter_tools_by_scope(await client.list_tools(), scope) # Filter tools based on user query if provided if user_query and tools: @@ -205,6 +253,7 @@ async def execute_tool( elif name == TOOL_GET_ACTION_DETAILS: server_name = arguments.get("server_name") action_name = arguments.get("action_name") + scope = arguments.get("scope", ACTION_SCOPE_ALL) if not server_name or not action_name: return [ @@ -216,7 +265,7 @@ async def execute_tool( try: client = client_manager.get_client(server_name) - tools = await client.list_tools() + tools = filter_tools_by_scope(await client.list_tools(), scope) action_found = None for tool in tools or []: @@ -249,6 +298,7 @@ async def execute_tool( path_params = arguments.get("path_params") query_params = arguments.get("query_params") body_schema = arguments.get("body_schema", "{}") + scope = arguments.get("scope", ACTION_SCOPE_ALL) if not server_name or not action_name: return [ @@ -260,6 +310,22 @@ async def execute_tool( try: client = client_manager.get_client(server_name) + tools = await client.list_tools() + scoped_tools = filter_tools_by_scope(tools, scope) + if scope != ACTION_SCOPE_ALL and not any( + tool["name"] == action_name for tool in scoped_tools + ): + result = { + "error": ( + f"Action '{action_name}' is not available in scope " + f"'{scope}' for server '{server_name}'" + ) + } + return [ + types.TextContent( + type="text", text=json.dumps(result, separators=(",", ":")) + ) + ] action_params = {} # Parse parameters if they're JSON strings @@ -295,6 +361,7 @@ async def execute_tool( query = arguments.get("query") server_name = arguments.get("server_name") max_results = arguments.get("max_results", 10) + scope = arguments.get("scope", ACTION_SCOPE_ALL) if not query or not server_name: return [ @@ -306,7 +373,7 @@ async def execute_tool( try: client = client_manager.get_client(server_name) - tools = await client.list_tools() + tools = filter_tools_by_scope(await client.list_tools(), scope) tools_map = {server_name: tools if tools else []} searcher = UniversalToolSearcher(tools_map) diff --git a/open-strata/tests/test_read_only_scope.py b/open-strata/tests/test_read_only_scope.py new file mode 100644 index 000000000..4eb120bc6 --- /dev/null +++ b/open-strata/tests/test_read_only_scope.py @@ -0,0 +1,165 @@ +"""Tests for Strata read-only action scoping.""" + +import json +from types import SimpleNamespace + +import pytest + +from strata.mcp_proxy.client import MCPClient +from strata.tools import ( + ACTION_SCOPE_READ_ONLY, + TOOL_DISCOVER_SERVER_ACTIONS, + TOOL_EXECUTE_ACTION, + TOOL_GET_ACTION_DETAILS, + execute_tool, + filter_tools_by_scope, + is_read_only_tool, +) + + +READ_TOOL = { + "name": "gmail_search", + "description": "Search messages", + "inputSchema": {"type": "object"}, + "annotations": {"readOnlyHint": True}, +} + +WRITE_TOOL = { + "name": "gmail_send", + "description": "Send a message", + "inputSchema": {"type": "object"}, + "annotations": {"readOnlyHint": False}, +} + + +class FakeClient: + def __init__(self): + self.called_tools = [] + + async def list_tools(self): + return [READ_TOOL, WRITE_TOOL] + + async def call_tool(self, action_name, action_params): + self.called_tools.append((action_name, action_params)) + return [] + + +class FakeManager: + def __init__(self, client): + self.active_clients = {"gmail": client} + self.client = client + + def get_client(self, server_name): + if server_name != "gmail": + raise KeyError(server_name) + return self.client + + +def parse_text_result(result): + return json.loads(result[0].text) + + +def test_read_only_detection_uses_mcp_annotations(): + assert is_read_only_tool(READ_TOOL) + assert not is_read_only_tool(WRITE_TOOL) + assert filter_tools_by_scope([READ_TOOL, WRITE_TOOL], ACTION_SCOPE_READ_ONLY) == [ + READ_TOOL + ] + + +@pytest.mark.asyncio +async def test_discover_server_actions_can_return_only_read_only_tools(): + client = FakeClient() + result = await execute_tool( + TOOL_DISCOVER_SERVER_ACTIONS, + {"server_names": ["gmail"], "scope": ACTION_SCOPE_READ_ONLY}, + FakeManager(client), + ) + + payload = parse_text_result(result) + assert payload["servers"]["gmail"] == { + "action_count": 1, + "actions": ["gmail_search"], + } + + +@pytest.mark.asyncio +async def test_get_action_details_hides_write_actions_in_read_only_scope(): + client = FakeClient() + result = await execute_tool( + TOOL_GET_ACTION_DETAILS, + { + "server_name": "gmail", + "action_name": "gmail_send", + "scope": ACTION_SCOPE_READ_ONLY, + }, + FakeManager(client), + ) + + payload = parse_text_result(result) + assert payload == {"error": "Action 'gmail_send' not found in server 'gmail'"} + + +@pytest.mark.asyncio +async def test_execute_action_rejects_write_actions_in_read_only_scope(): + client = FakeClient() + result = await execute_tool( + TOOL_EXECUTE_ACTION, + { + "server_name": "gmail", + "action_name": "gmail_send", + "scope": ACTION_SCOPE_READ_ONLY, + }, + FakeManager(client), + ) + + payload = parse_text_result(result) + assert payload == { + "error": ( + "Action 'gmail_send' is not available in scope 'read_only' " + "for server 'gmail'" + ) + } + assert client.called_tools == [] + + +class DumpableAnnotations: + def model_dump(self, exclude_none=False): + assert exclude_none is True + return {"readOnlyHint": True} + + +class FakeTransport: + def is_connected(self): + return True + + def get_session(self): + return self + + async def list_tools(self): + return SimpleNamespace( + tools=[ + SimpleNamespace( + name="read", + description="Read records", + inputSchema={"type": "object"}, + annotations=DumpableAnnotations(), + ) + ] + ) + + +@pytest.mark.asyncio +async def test_mcp_client_preserves_tool_annotations_for_router_scoping(): + client = MCPClient(FakeTransport()) + + tools = await client.list_tools(use_cache=False) + + assert tools == [ + { + "name": "read", + "description": "Read records", + "inputSchema": {"type": "object"}, + "annotations": {"readOnlyHint": True}, + } + ]