Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/contrib.md
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ The helpers under `acp.contrib` package recurring patterns we saw in integration

- `ToolCallTracker.start()/progress()/append_stream_text()` emits canonical `ToolCallStart` / `ToolCallProgress` updates and keeps an in-memory view via `view()` or `tool_call_model()`.
- `PermissionBroker.request_for()` wraps `requestPermission` RPCs. It reuses tracker state (or a provided `ToolCall`), lets you append extra content, and defaults to Approve / Approve for session / Reject options.
- Omit `options` or pass `None` to use broker defaults. An explicit empty option list raises `MissingPermissionOptionsError` before the permission request is sent; the same applies to empty `default_options` when no per-request options are supplied.
- `default_permission_options()` exposes that canonical option triple if you need to customise it.

> Tip: Keep one tracker near the agent event loop. Emit notifications through it and share the tracker with `PermissionBroker` so permission prompts always match the latest tool call state.
Expand Down
7 changes: 5 additions & 2 deletions src/acp/contrib/permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,8 @@ def __init__(
self._requester = requester
self._tracker = tracker
self._default_options = tuple(
option.model_copy(deep=True) for option in (default_options or default_permission_options())
option.model_copy(deep=True)
for option in (default_permission_options() if default_options is None else default_options)
)

async def request_for(
Expand Down Expand Up @@ -84,7 +85,9 @@ async def request_for(
existing.append(ContentToolCallContent(content=TextContentBlock(text=description)))
tool_call.content = existing

option_set = tuple(option.model_copy(deep=True) for option in (options or self._default_options))
option_set = tuple(
option.model_copy(deep=True) for option in (self._default_options if options is None else options)
)
if not option_set:
raise MissingPermissionOptionsError()

Expand Down
56 changes: 55 additions & 1 deletion tests/contrib/test_contrib_permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import pytest

from acp.contrib.permissions import PermissionBroker, default_permission_options
from acp.contrib.permissions import MissingPermissionOptionsError, PermissionBroker, default_permission_options
from acp.contrib.tool_calls import ToolCallTracker
from acp.schema import (
AllowedOutcome,
Expand Down Expand Up @@ -58,6 +58,60 @@ async def requester(request: RequestPermissionRequest):
assert recorded == ["allow"]


@pytest.mark.asyncio
async def test_permission_broker_none_uses_standard_options():
tracker = ToolCallTracker(id_factory=lambda: "standard")
tracker.start("external", title="Standard options")
recorded: list[list[str]] = []

async def requester(request: RequestPermissionRequest):
recorded.append([option.option_id for option in request.options])
return RequestPermissionResponse(outcome=AllowedOutcome(option_id="approve", outcome="selected"))

broker = PermissionBroker("session", requester, tracker=tracker, default_options=None)
await broker.request_for("external", options=None)
assert recorded == [["approve", "approve_for_session", "reject"]]


@pytest.mark.asyncio
async def test_permission_broker_custom_default_and_override():
tracker = ToolCallTracker(id_factory=lambda: "reject-only")
tracker.start("external", title="Reject only")
reject = PermissionOption(option_id="reject", name="Reject", kind="reject_once")
allow = PermissionOption(option_id="allow", name="Allow", kind="allow_once")
recorded: list[list[str]] = []

async def requester(request: RequestPermissionRequest):
recorded.append([option.option_id for option in request.options])
return RequestPermissionResponse(
outcome=AllowedOutcome(option_id=request.options[0].option_id, outcome="selected")
)

broker = PermissionBroker("session", requester, tracker=tracker, default_options=[reject])
await broker.request_for("external")
await broker.request_for("external", options=[allow])
assert recorded == [["reject"], ["allow"]]


@pytest.mark.asyncio
@pytest.mark.parametrize(("default_options", "options"), [(None, []), ([], None)])
async def test_permission_broker_rejects_empty_option_sources(default_options, options):
tracker = ToolCallTracker(id_factory=lambda: "empty")
tracker.start("external", title="No options")
recorded: list[RequestPermissionRequest] = []

async def requester(request: RequestPermissionRequest):
recorded.append(request)
return RequestPermissionResponse(
outcome=AllowedOutcome(option_id=request.options[0].option_id, outcome="selected")
)

broker = PermissionBroker("session", requester, tracker=tracker, default_options=default_options)
with pytest.raises(MissingPermissionOptionsError, match="requires at least one permission option"):
await broker.request_for("external", options=options)
assert recorded == []


def test_default_permission_options_shape():
options = default_permission_options()
assert len(options) == 3
Expand Down