feat(strata): add read-only action scope

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/)** 👈
-

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

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)

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},
+        }
+    ]