Fix hosted result payload handling

src/mcp_server_appwrite/http_app.py

@@ -147,7 +147,7 @@ async def health_endpoint(request: Request) -> PlainTextResponse:
 def build_app() -> Starlette:
     telemetry.init_telemetry("http", SERVER_VERSION)
     tools_manager = build_catalog_tools_manager()
-    operator = build_operator(tools_manager)
+    operator = build_operator(tools_manager, store_results=False)
     server = build_mcp_server(operator, transport="http")
 
     # Streamable HTTP with SSE responses (the MCP SDK/ecosystem default). Stateless,

src/mcp_server_appwrite/operator.py

@@ -104,13 +104,15 @@ def __init__(
         docs_search: DocsSearch | None = None,
         context_provider: ContextProvider | None = None,
         preview_threshold: int = PREVIEW_THRESHOLD,
+        store_results: bool = True,
         search_limit: int = SEARCH_LIMIT,
     ):
         self._tools_manager = tools_manager
         self._execute_tool = execute_tool
         self._docs_search = docs_search
         self._context_provider = context_provider
         self._preview_threshold = preview_threshold
+        self._store_results = store_results
         self._search_limit = search_limit
         self._result_store = ResultStore()
         self._catalog = self._build_catalog()
@@ -474,6 +476,9 @@ def _call_hidden_tool(self, raw_arguments: dict[str, Any]) -> list[ToolContent]:
     def _preview_or_store_result(
         self, tool_name: str, content: list[ToolContent]
     ) -> list[ToolContent]:
+        if not self._store_results:
+            return content
+
         if all(isinstance(item, types.TextContent) for item in content):
             full_text = "\n".join(
                 item.text for item in content if isinstance(item, types.TextContent)

src/mcp_server_appwrite/server.py

@@ -865,14 +865,19 @@ def _format_appwrite_error(exc: AppwriteException) -> str:
 
 
 def build_instructions(transport: str = "http") -> str:
+    result_handling = (
+        "Large results are stored as resources; read the URI returned by the tool."
+        if transport == "stdio"
+        else "Hosted HTTP returns tool results inline, including images and other binary payloads."
+    )
     common = (
         "Appwrite workflow: use appwrite_get_context to understand the current "
         "connection and available project resources, then use appwrite_search_tools "
         "and appwrite_call_tool for specific operations. "
         "Mutating hidden tools require confirm_write=true. "
         "For questions about Appwrite concepts, products, or guides, use "
         "appwrite_search_docs to search the documentation when available. "
-        "Large results are stored as resources; read the URI returned by the tool."
+        f"{result_handling}"
     )
 
     if transport == "stdio":
@@ -1003,7 +1008,10 @@ def _emit_initialize(server: Server) -> None:
 
 
 def build_operator(
-    tools_manager: ToolManager, client: Client | None = None
+    tools_manager: ToolManager,
+    client: Client | None = None,
+    *,
+    store_results: bool = True,
 ) -> Operator:
     """Wire the operator surface to the per-request execution path. The execution
     callback re-binds each call to a per-request client via `resolve_client` in
@@ -1032,6 +1040,7 @@ def build_operator(
         ),
         context_provider=lambda arguments: _get_context_for_request(arguments, client),
         docs_search=docs_search,
+        store_results=store_results,
     )
 
 

tests/unit/test_operator.py

@@ -277,6 +277,57 @@ def test_large_result_is_stored_as_resource(self):
         self.assertEqual(contents[0].mime_type, "application/json")
         self.assertIn('"type": "text"', contents[0].content)
 
+    def test_store_results_false_returns_large_result_inline(self):
+        manager = ToolManager()
+        manager.tools_registry = {
+            "tables_db_list": {
+                "definition": make_tool("tables_db_list", "List all databases."),
+                "function": object(),
+                "parameter_types": {},
+            },
+        }
+        runtime = Operator(
+            manager,
+            lambda name, arguments, *_: [
+                types.TextContent(type="text", text="x" * 1200)
+            ],
+            store_results=False,
+        )
+
+        result = runtime.execute_public_tool(
+            "appwrite_call_tool",
+            {"tool_name": "tables_db_list"},
+        )
+
+        self.assertEqual(result[0].text, "x" * 1200)
+        self.assertNotIn("appwrite://operator/results/", result[0].text)
+
+    def test_store_results_false_returns_image_inline(self):
+        manager = ToolManager()
+        manager.tools_registry = {
+            "avatars_get_qr": {
+                "definition": make_tool("avatars_get_qr", "Get a QR code."),
+                "function": object(),
+                "parameter_types": {},
+            },
+        }
+        runtime = Operator(
+            manager,
+            lambda name, arguments, *_: [
+                types.ImageContent(type="image", data="aW1hZ2U=", mimeType="image/png")
+            ],
+            store_results=False,
+        )
+
+        result = runtime.execute_public_tool(
+            "appwrite_call_tool",
+            {"tool_name": "avatars_get_qr"},
+        )
+
+        self.assertEqual(len(result), 1)
+        self.assertIsInstance(result[0], types.ImageContent)
+        self.assertEqual(result[0].mimeType, "image/png")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_server.py

@@ -103,6 +103,8 @@ def test_build_instructions_are_transport_specific(self):
         self.assertNotIn("Appwrite console", stdio)
         self.assertIn("Appwrite console", http)
         self.assertIn("project_id", http)
+        self.assertIn("Large results are stored as resources", stdio)
+        self.assertIn("returns tool results inline", http)
 
     def test_coerce_input_file_from_path(self):
         with tempfile.NamedTemporaryFile(suffix=".txt") as handle: