feat(mcp): Act as the task's user and always confirm write tools

.github/workflows/image-publish.yml

@@ -0,0 +1,70 @@
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+# Publish the ExApp image of this repository to ghcr.io/<owner>/context_agent when a
+# release is published, tagged with the release tag without its leading "v". A pull
+# request that changes this workflow or the Dockerfile builds the image without pushing.
+name: Publish image
+
+on:
+  release:
+    types: [published]
+  pull_request:
+    paths:
+      - .github/workflows/image-publish.yml
+      - Dockerfile
+
+permissions:
+  contents: read
+
+concurrency:
+  group: image-publish-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+jobs:
+  image:
+    runs-on: ubuntu-latest
+
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+
+      - name: Image name
+        id: image
+        env:
+          TAG: ${{ github.event.release.tag_name }}
+        run: |
+          # registry names are lowercase, owners need not be
+          tag="${TAG#v}"
+          echo "name=ghcr.io/${GITHUB_REPOSITORY,,}:${tag:-pr}" >> "$GITHUB_OUTPUT"
+
+      - uses: docker/setup-qemu-action@99012661954931238ded8c8b007157a8430204e1 # v4.4.0
+      - uses: docker/setup-buildx-action@f87e5991a6d7451dcb8d9637bfbc97413f497069 # v4.4.1
+
+      - name: Log in to ghcr.io
+        if: ${{ github.event_name == 'release' }}
+        uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Build, and push on a release
+        id: build
+        uses: docker/build-push-action@c3c9e263c25d99ce0380d002d59b67737d91b0dc # v7.4.0
+        with:
+          context: .
+          platforms: linux/amd64,linux/arm64
+          push: ${{ github.event_name == 'release' }}
+          tags: ${{ steps.image.outputs.name }}
+          labels: org.opencontainers.image.source=${{ github.server_url }}/${{ github.repository }}
+
+      - name: Report the digest
+        if: ${{ github.event_name == 'release' }}
+        run: echo "Pushed ${{ steps.image.outputs.name }}@${{ steps.build.outputs.digest }}" >> "$GITHUB_STEP_SUMMARY"

.github/workflows/reuse.yml

@@ -16,7 +16,7 @@ permissions:
 
 jobs:
   reuse-compliance-check:
-    runs-on: ubuntu-latest-low
+    runs-on: ubuntu-latest
     steps:
       - name: Checkout
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

ex_app/lib/all_tools/lib/impulse.py

@@ -138,7 +138,8 @@ class Reversibility(IntEnum):
 ITEM_COUNT_THRESHOLD_SETTING_ID = 'item_count_threshold'
 
 # MCP tools the admin vouches for, so they are never confirmed: a comma or newline
-# separated list of tool names, or '*' for all of them.
+# separated list of tool names, or '*' for all of them. A tool its server marks as
+# writing is left out (see ex_app.lib.all_tools.mcp.is_write).
 MCP_ALWAYS_ALLOW_SETTING_ID = 'mcp_always_allow'
 
 _ATTR = 'impulse_hook'

ex_app/lib/all_tools/mcp.py

@@ -1,5 +1,6 @@
 # SPDX-FileCopyrightText: 2025 Nextcloud GmbH and Nextcloud contributors
 # SPDX-License-Identifier: AGPL-3.0-or-later
+from base64 import b64encode
 from json import JSONDecodeError
 
 from langchain_mcp_adapters.client import MultiServerMCPClient
@@ -22,12 +23,12 @@ async def get_tools(nc: AsyncNextcloudApp):
 		mcp_config = {}
 	always_allowed = parse_always_allow(await nc.appconfig_ex.get_value(MCP_ALWAYS_ALLOW_SETTING_ID, default=""))
 	try:
-		server = MultiServerMCPClient(mcp_config)
+		server = MultiServerMCPClient(await sign_nextcloud_servers(nc, mcp_config))
 		tools = await asyncio.wait_for(server.get_tools(), timeout=120)
 		for tool in tools:
 			# MCP tools cannot be classified, so they are confirmed as the worst case
-			# unless the admin vouched for them
-			if "*" in always_allowed or tool.name in always_allowed:
+			# unless the admin vouched for them; a tool the server says writes never is
+			if not is_write(tool) and ("*" in always_allowed or tool.name in always_allowed):
 				always_allow(tool.coroutine)
 			tool.name = "mcp_" + tool.name
 		return tools
@@ -37,6 +38,39 @@ async def get_tools(nc: AsyncNextcloudApp):
 	return []
 
 
+async def sign_nextcloud_servers(nc: AsyncNextcloudApp, mcp_config: dict) -> dict:
+	"""Sign the requests to MCP servers on this Nextcloud for the user of the task.
+
+	They carry the same AppAPI headers as this app's own requests to Nextcloud, so a
+	server behind the AppAPI proxy runs each call as that user. Servers elsewhere get
+	no headers: the app secret never leaves this Nextcloud.
+	"""
+	cfg = nc.app_cfg
+	headers = {
+		"AA-VERSION": cfg.aa_version,
+		"EX-APP-ID": cfg.app_name,
+		"EX-APP-VERSION": cfg.app_version,
+		"AUTHORIZATION-APP-API": b64encode(f"{await nc.user}:{cfg.app_secret}".encode()).decode(),
+	}
+	signed = {}
+	for name, connection in mcp_config.items():
+		if str(connection.get("url", "")).startswith(cfg.endpoint + "/"):
+			connection = {**connection, "headers": {**(connection.get("headers") or {}), **headers}}
+		signed[name] = connection
+	return signed
+
+
+def is_write(tool) -> bool:
+	"""Whether the MCP server says this tool can change something.
+
+	langchain-mcp-adapters puts every field of the tool's MCP annotations in its
+	metadata. readOnlyHint defaults to false, so an annotated tool writes unless it is
+	marked read-only; a tool without annotations is not classified by its server.
+	"""
+	annotations = tool.metadata or {}
+	return "readOnlyHint" in annotations and annotations["readOnlyHint"] is not True
+
+
 def parse_always_allow(value) -> set[str]:
 	"""Turn the admin's list of always allowed MCP tools into a set of names.
 

ex_app/lib/main.py

@@ -273,7 +273,8 @@ async def exapp_lifespan(app: FastAPI):
             description=_(
                 "Context Agent cannot tell what an MCP tool does, so it asks the user to confirm every"
                 " call to one. List the names of MCP tools you trust, separated by commas, to let them"
-                " run without asking, or enter * to allow all MCP tools."
+                " run without asking, or enter * to allow all MCP tools. A tool its MCP server marks as"
+                " changing something is still confirmed on every call."
             ),
             type=SettingsFieldType.TEXT,
             default="",

tests/test_mcp_tools.py

@@ -0,0 +1,113 @@
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""MCP tools act as the task's user, and a tool the server says writes is always confirmed.
+
+Runnable standalone (``python tests/test_mcp_tools.py``) or via pytest.
+No network: the MCP client is replaced by one that records the connections it is
+given and lists tools converted the way langchain-mcp-adapters converts them.
+"""
+import asyncio
+import base64
+import json
+import os
+import sys
+import tempfile
+
+os.environ.setdefault("APP_ID", "context_agent")
+os.environ.setdefault("APP_HOST", "127.0.0.1")
+os.environ.setdefault("APP_PORT", "9081")
+os.environ.setdefault("APP_SECRET", "test-secret")
+os.environ.setdefault("APP_VERSION", "2.10.0")
+os.environ.setdefault("NEXTCLOUD_URL", "http://127.0.0.1:8080")
+os.environ.setdefault("APP_PERSISTENT_STORAGE", tempfile.mkdtemp())
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from langchain_mcp_adapters.tools import convert_mcp_tool_to_langchain_tool  # noqa: E402
+from mcp.types import Tool, ToolAnnotations  # noqa: E402
+from nc_py_api import AsyncNextcloudApp  # noqa: E402
+
+from ex_app.lib.all_tools import mcp  # noqa: E402
+from ex_app.lib.all_tools.lib.impulse import MCP_ALWAYS_ALLOW_SETTING_ID, is_always_allowed  # noqa: E402
+
+ON_NEXTCLOUD = {
+	"url": os.environ["NEXTCLOUD_URL"] + "/index.php/apps/app_api/proxy/mcp_connector/mcp",
+	"transport": "streamable_http",
+}
+ELSEWHERE = {"url": "https://mcp.example.com/mcp", "transport": "streamable_http"}
+
+
+def _server_tool(name, annotations=None):
+	return Tool(name=name, inputSchema={"type": "object", "properties": {}}, annotations=annotations)
+
+
+SERVER_TOOLS = [
+	_server_tool("read_note", ToolAnnotations(readOnlyHint=True)),
+	_server_tool("create_note", ToolAnnotations(readOnlyHint=False, destructiveHint=False)),
+	# readOnlyHint defaults to false: a server that annotates a tool without it says it writes
+	_server_tool("update_note", ToolAnnotations(destructiveHint=True)),
+	# not annotated at all: the server does not classify it
+	_server_tool("get_forecast"),
+]
+
+
+class FakeMCPClient:
+	"""Records the connections it is given, lists SERVER_TOOLS."""
+
+	connections = None
+
+	def __init__(self, connections):
+		FakeMCPClient.connections = connections
+
+	async def get_tools(self):
+		return [convert_mcp_tool_to_langchain_tool(None, t, connection=ON_NEXTCLOUD) for t in SERVER_TOOLS]
+
+
+class FakeAppConfig:
+	def __init__(self, values):
+		self.values = values
+
+	async def get_value(self, key, default=None):
+		return self.values.get(key, default)
+
+
+def _load_tools(user, mcp_config, always_allow=""):
+	async def run():
+		nc = AsyncNextcloudApp(user=user)
+		nc.appconfig_ex = FakeAppConfig({
+			"mcp_config": json.dumps(mcp_config),
+			MCP_ALWAYS_ALLOW_SETTING_ID: always_allow,
+		})
+		return await mcp.get_tools(nc)
+
+	real_client, mcp.MultiServerMCPClient = mcp.MultiServerMCPClient, FakeMCPClient
+	try:
+		return asyncio.run(run()), FakeMCPClient.connections
+	finally:
+		mcp.MultiServerMCPClient = real_client
+
+
+def test_calls_to_this_nextcloud_are_signed_for_the_tasks_user():
+	for user in ("alice", "bob"):
+		_, connections = _load_tools(user, {"nextcloud": ON_NEXTCLOUD, "elsewhere": ELSEWHERE})
+		headers = connections["nextcloud"]["headers"]
+		assert base64.b64decode(headers["AUTHORIZATION-APP-API"]).decode() == f"{user}:{os.environ['APP_SECRET']}"
+		assert headers["EX-APP-ID"] == os.environ["APP_ID"]
+		assert headers["EX-APP-VERSION"] == os.environ["APP_VERSION"]
+		assert headers["AA-VERSION"]
+		# the app secret never leaves this Nextcloud
+		assert connections["elsewhere"] == ELSEWHERE
+
+
+def test_a_tool_the_server_says_writes_is_never_always_allowed():
+	tools, _ = _load_tools("alice", {"nextcloud": ON_NEXTCLOUD}, always_allow="*")
+	assert {t.name for t in tools if is_always_allowed(t)} == {"mcp_read_note", "mcp_get_forecast"}
+
+	tools, _ = _load_tools("alice", {"nextcloud": ON_NEXTCLOUD}, always_allow="read_note, mcp_create_note, update_note")
+	assert {t.name for t in tools if is_always_allowed(t)} == {"mcp_read_note"}
+
+
+if __name__ == "__main__":
+	for name, fn in list(globals().items()):
+		if name.startswith("test_") and callable(fn):
+			fn()
+			print(f"ok {name}")