@@ -130,6 +130,11 @@ async def release_terminal(self, terminal_id: str) -> None:
     async def kill_terminal(self, terminal_id: str) -> None:
         await self._require_conn().kill_terminal(session_id=self._active_session_id(), terminal_id=terminal_id)
 
+    async def request_permission(self, *, options: list[Any], tool_call: Any) -> Any:
+        return await self._require_conn().request_permission(
+            options=options, session_id=self._active_session_id(), tool_call=tool_call
+        )
+
 
 class ACPZedServerAgent(ACPBaseAgent):
     """Wrapper that exposes a BeeAI agent over the Zed Agent Client Protocol (stdio).
@@ -211,13 +216,21 @@ async def prompt(
             with contextlib.suppress(asyncio.CancelledError):
                 await existing
 
+        # Lazy import to avoid an import cycle (io.py depends on FsBridge).
+        from beeai_framework.adapters.acp_zed.serve.io import ACPZedIOContext
+
         token = _active_session.set(session_id)
-        task = asyncio.create_task(self._run_turn(session_id, prompt, self._conn, self._sessions[session_id]))
-        self._active_tasks[session_id] = task
         try:
-            stop_reason = await task
-        except asyncio.CancelledError:
-            stop_reason = "cancelled"
+            with ACPZedIOContext(self._bridge):
+                # Create the task *inside* the io context so the ContextVar propagates
+                # to `run_turn` and any `io_confirm` calls made by tools / requirements
+                # route through `session/request_permission`.
+                task = asyncio.create_task(self._run_turn(session_id, prompt, self._conn, self._sessions[session_id]))
+                self._active_tasks[session_id] = task
+                try:
+                    stop_reason = await task
+                except asyncio.CancelledError:
+                    stop_reason = "cancelled"
         finally:
             _active_session.reset(token)
             self._active_tasks.pop(session_id, None)
 
@@ -0,0 +1,75 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any, Self
+
+from typing_extensions import Unpack
+
+try:
+    from acp.schema import AllowedOutcome, PermissionOption, ToolCallUpdate
+except ModuleNotFoundError as e:
+    raise ModuleNotFoundError(
+        "Optional module [acp_zed] not found.\nRun 'pip install \"beeai-framework[acp-zed]\"' to install."
+    ) from e
+
+from beeai_framework.utils.io import IOConfirmKwargs, setup_io_context
+
+if TYPE_CHECKING:
+    from beeai_framework.adapters.acp_zed.serve.agent import FsBridge
+
+
+class ACPZedIOContext:
+    """Route `io_confirm` through ACP `session/request_permission` for the life of a turn.
+
+    Mirrors the A2A / IBM-ACP `setup_io_context` pattern: when this context is active,
+    any agent code calling `io_confirm(...)` (e.g. `AskPermissionRequirement`) gets the
+    prompt delivered to the editor user via `session/request_permission` instead of
+    stdin, and the user's choice flows back as a `bool`.
+
+    `io_read` is overridden to raise — ACP has no free-form text input method; calling
+    it under the adapter is a programmer error worth surfacing loudly rather than
+    hanging on a stdin that's already owned by the JSON-RPC transport.
+    """
+
+    def __init__(self, bridge: FsBridge) -> None:
+        self._bridge = bridge
+        self._cleanup: Callable[[], None] = lambda: None
+
+    def __enter__(self) -> Self:
+        # pyrefly: ignore [bad-argument-type]
+        self._cleanup = setup_io_context(read=self._read, confirm=self._confirm)
+        return self
+
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        self._cleanup()
+        self._cleanup = lambda: None
+
+    async def _read(self, prompt: str) -> str:
+        raise RuntimeError(
+            "io_read is not supported under the ACPZed adapter — ACP has no free-form "
+            "text-input method. Use a tool call or session/request_permission instead."
+        )
+
+    async def _confirm(self, prompt: str, **kwargs: Unpack[IOConfirmKwargs]) -> bool:
+        title = kwargs.get("title", prompt)
+        data = kwargs.get("data", {})
+        submit_label = kwargs.get("submit_label", "Allow")
+        cancel_label = kwargs.get("cancel_label", "Deny")
+
+        tool_call = ToolCallUpdate(
+            tool_call_id=f"confirm-{id(data)}",
+            title=title,
+            raw_input=data if isinstance(data, dict) else {"data": data},
+            status="pending",
+        )
+        response = await self._bridge.request_permission(
+            options=[
+                PermissionOption(option_id="allow", name=submit_label, kind="allow_once"),
+                PermissionOption(option_id="deny", name=cancel_label, kind="reject_once"),
+            ],
+            tool_call=tool_call,
+        )
+        return isinstance(response.outcome, AllowedOutcome) and response.outcome.option_id == "allow"
@@ -0,0 +1,103 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+pytest.importorskip("acp", reason="Optional module [acp_zed] not installed.")
+
+from acp.schema import (
+    AllowedOutcome,
+    DeniedOutcome,
+    RequestPermissionResponse,
+)
+
+from beeai_framework.adapters.acp_zed.serve.agent import FsBridge, _active_session
+from beeai_framework.adapters.acp_zed.serve.io import ACPZedIOContext
+from beeai_framework.utils.io import io_confirm, io_read
+
+
+class _StubConn:
+    def __init__(self, *, allow: bool) -> None:
+        self.allow = allow
+        self.calls: list[dict[str, Any]] = []
+
+    async def request_permission(self, **kwargs: Any) -> RequestPermissionResponse:
+        self.calls.append(kwargs)
+        if self.allow:
+            return RequestPermissionResponse(
+                outcome=AllowedOutcome(outcome="selected", option_id="allow")  # type: ignore[call-arg]
+            )
+        return RequestPermissionResponse(outcome=DeniedOutcome(outcome="cancelled"))
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_confirm_allow_routes_through_request_permission() -> None:
+    bridge = FsBridge()
+    conn = _StubConn(allow=True)
+    bridge.bind(conn)  # type: ignore[arg-type]
+
+    tok = _active_session.set("sess-1")
+    try:
+        with ACPZedIOContext(bridge):
+            result = await io_confirm("Are you sure?", data={"tool": "shell"})
+    finally:
+        _active_session.reset(tok)
+
+    assert result is True
+    assert len(conn.calls) == 1
+    call = conn.calls[0]
+    assert call["session_id"] == "sess-1"
+    assert [opt.option_id for opt in call["options"]] == ["allow", "deny"]
+    assert call["tool_call"].raw_input == {"tool": "shell"}
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_confirm_deny_returns_false() -> None:
+    bridge = FsBridge()
+    bridge.bind(_StubConn(allow=False))  # type: ignore[arg-type]
+    tok = _active_session.set("sess-1")
+    try:
+        with ACPZedIOContext(bridge):
+            result = await io_confirm("Run this?", data={})
+    finally:
+        _active_session.reset(tok)
+    assert result is False
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_read_raises_under_context() -> None:
+    bridge = FsBridge()
+    bridge.bind(_StubConn(allow=True))  # type: ignore[arg-type]
+    tok = _active_session.set("sess-1")
+    try:
+        with ACPZedIOContext(bridge), pytest.raises(RuntimeError, match="io_read is not supported"):
+            await io_read("anything: ")
+    finally:
+        _active_session.reset(tok)
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_context_restores_previous_handlers() -> None:
+    """After exiting the context, the previous (default) handlers are restored."""
+    bridge = FsBridge()
+    bridge.bind(_StubConn(allow=True))  # type: ignore[arg-type]
+    tok = _active_session.set("sess-1")
+    try:
+        with ACPZedIOContext(bridge):
+            assert await io_confirm("x", data={}) is True
+        # Outside the context, io_read would block on stdin; just confirm it's a
+        # different function now (the default one, not our _read).
+        from beeai_framework.utils.io import _storage
+
+        restored = _storage.get()
+        assert restored.read.__name__ != "_read"
+    finally:
+        _active_session.reset(tok)