Python: Align GitHub Copilot provider function approval to use SDK on_pre_tool_use hook (#6750)
* Python: align GitHub Copilot approval to SDK on_pre_tool_use hook Replace the bespoke on_function_approval enforcement in the GitHub Copilot provider with the Copilot SDK's native on_pre_tool_use hook. When no caller hook is supplied, a default hook returns 'ask' for approval_mode='always_require' tools (routed to on_permission_request) and defers others; a caller-supplied on_pre_tool_use takes precedence and logs a warning for any unenforced approval tool. Fixes #6746 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix type-checker errors and restore load_dotenv in sample Use a complete PreToolUseHookInput in on_pre_tool_use hook tests so pyright/pyrefly/ty/zuban no longer report missing required TypedDict keys. Restore load_dotenv() in the function-approval sample for consistency with the other GitHub Copilot samples (PR review feedback). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Deprecate on_function_approval instead of removing it Per PR review feedback, keep the on_function_approval callback working (still enforced in the tool handler for approval_mode='always_require' tools) but emit a DeprecationWarning at construction, so existing users get a signal rather than a silent behavior change. The default on_pre_tool_use ask-hook is not installed when on_function_approval is set, avoiding double-gating. Precedence: user on_pre_tool_use > on_function_approval > default ask-hook. Adds tests for the deprecated path and documents it in the package README. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Make on_function_approval and on_pre_tool_use mutually exclusive Per automated review feedback, instead of a precedence ordering between the deprecated on_function_approval callback and the new on_pre_tool_use hook (which silently double-gated when both were set), raise ValueError if both are supplied - at construction (both in default_options) or per run (per-run on_pre_tool_use with a construction-time on_function_approval). This matches the repo convention for deprecated-vs-new params (see _workflows/_workflow.py) and removes the flag-threading. Updates tests and the package README. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -9,3 +9,53 @@ pip install agent-framework-github-copilot --pre
|
||||
## GitHub Copilot Agent
|
||||
|
||||
The GitHub Copilot agent enables integration with GitHub Copilot, allowing you to interact with Copilot's agentic capabilities through the Agent Framework.
|
||||
|
||||
## Tool approval (`approval_mode="always_require"`)
|
||||
|
||||
The GitHub Copilot SDK owns the tool-calling loop for this provider, so approval for
|
||||
custom function tools is enforced through the SDK's native pre-execution hook rather
|
||||
than the standard Agent Framework approval round-trip.
|
||||
|
||||
When you register a `FunctionTool` declared with `approval_mode="always_require"` and you
|
||||
do **not** supply your own `on_pre_tool_use` hook, `GitHubCopilotAgent` installs a default
|
||||
`on_pre_tool_use` hook that returns `"ask"` for that tool and defers (`None`) for all other
|
||||
tools. The `"ask"` decision routes to your `on_permission_request` handler, where you
|
||||
approve or deny the call:
|
||||
|
||||
```python
|
||||
from agent_framework import tool
|
||||
from agent_framework.github import GitHubCopilotAgent, GitHubCopilotOptions
|
||||
from copilot.session import PermissionHandler
|
||||
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def delete_file(path: str) -> str:
|
||||
"""Delete a file."""
|
||||
...
|
||||
|
||||
|
||||
agent = GitHubCopilotAgent(
|
||||
tools=[delete_file],
|
||||
# The "ask" decision is routed here; approve or deny the call.
|
||||
default_options=GitHubCopilotOptions(on_permission_request=PermissionHandler.approve_all),
|
||||
)
|
||||
```
|
||||
|
||||
> **⚠️ If you provide your own `on_pre_tool_use` hook**, it takes precedence and the agent
|
||||
> does **not** install its default approval hook. In that case **you are fully responsible**
|
||||
> for enforcing approval — including for any `approval_mode="always_require"` tool (e.g. by
|
||||
> returning a `"deny"` or `"ask"` decision). The agent logs a warning naming any
|
||||
> approval-required tool that your hook must handle.
|
||||
>
|
||||
> Note: with the default (deny-all) permission handler, an `always_require` tool is denied
|
||||
> unless you wire an approving `on_permission_request`.
|
||||
|
||||
### Deprecated: `on_function_approval`
|
||||
|
||||
The `on_function_approval` callback is **deprecated**. It still works (and is still enforced
|
||||
inside the tool handler for backward compatibility), but it emits a `DeprecationWarning` and
|
||||
will be removed in a future version. Migrate to the `on_pre_tool_use` + `on_permission_request`
|
||||
model described above. When `on_function_approval` is set, it gates `always_require` tools and
|
||||
the default ask-hook is not installed. It is **mutually exclusive** with `on_pre_tool_use` —
|
||||
setting both (whether at construction or per run) raises `ValueError`.
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@ import contextlib
|
||||
import inspect
|
||||
import logging
|
||||
import sys
|
||||
import warnings
|
||||
from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, MutableMapping, Sequence
|
||||
from typing import Any, ClassVar, Generic, Literal, TypedDict, overload
|
||||
|
||||
@@ -39,7 +40,15 @@ from agent_framework.observability import AgentTelemetryLayer
|
||||
try:
|
||||
from copilot import CopilotClient, CopilotSession, RuntimeConnection
|
||||
from copilot.generated.rpc import PermissionDecisionUserNotAvailable
|
||||
from copilot.session import MCPServerConfig, PermissionRequestResult, ProviderConfig, SystemMessageConfig
|
||||
from copilot.session import (
|
||||
MCPServerConfig,
|
||||
PermissionRequestResult,
|
||||
PreToolUseHandler,
|
||||
PreToolUseHookOutput,
|
||||
ProviderConfig,
|
||||
SessionHooks,
|
||||
SystemMessageConfig,
|
||||
)
|
||||
from copilot.session_events import PermissionRequest, SessionEvent, SessionEventType
|
||||
from copilot.tools import Tool as CopilotTool
|
||||
from copilot.tools import ToolInvocation, ToolResult
|
||||
@@ -65,23 +74,18 @@ PermissionHandlerType = Callable[
|
||||
|
||||
|
||||
FunctionApprovalCallback = Callable[[Content], "bool | Awaitable[bool]"]
|
||||
"""Callback invoked by the agent before executing a FunctionTool that requires approval.
|
||||
"""Deprecated approval callback for ``FunctionTool`` instances declared with
|
||||
``approval_mode="always_require"``.
|
||||
|
||||
.. deprecated::
|
||||
Use the SDK ``on_pre_tool_use`` hook together with ``on_permission_request``
|
||||
instead. The default ``on_pre_tool_use`` hook returns ``"ask"`` for
|
||||
``always_require`` tools and routes the decision to ``on_permission_request``.
|
||||
|
||||
The callback receives a ``FunctionCallContent`` describing the pending call
|
||||
(``name``, ``arguments``, and a synthetic ``call_id``) and must return ``True``
|
||||
to allow execution or ``False`` to deny it. Both synchronous and ``await``-able
|
||||
return values are supported.
|
||||
|
||||
The Copilot CLI manages its own tool-calling loop, so the framework cannot
|
||||
round-trip a ``FunctionApprovalRequestContent`` / ``FunctionApprovalResponseContent``
|
||||
pair the way the standard chat-client pipeline does. This callback is the
|
||||
agent-level enforcement point for tools declared with
|
||||
``approval_mode="always_require"``: when no callback is configured the agent
|
||||
denies these calls by default.
|
||||
|
||||
Note: this is independent of ``on_permission_request``, which gates the
|
||||
Copilot SDK's *built-in* shell/file actions; ``on_function_approval`` gates
|
||||
agent-framework ``FunctionTool`` calls.
|
||||
"""
|
||||
|
||||
|
||||
@@ -90,7 +94,7 @@ async def _resolve_function_approval(
|
||||
func_tool: FunctionTool,
|
||||
arguments: Mapping[str, Any] | None,
|
||||
) -> bool:
|
||||
"""Run the agent-level approval callback for a pending tool call.
|
||||
"""Run the deprecated agent-level approval callback for a pending tool call.
|
||||
|
||||
Returns ``True`` only when ``callback`` is configured and explicitly returns
|
||||
a truthy value. A missing callback or any callback failure is treated as a
|
||||
@@ -205,13 +209,36 @@ class GitHubCopilotOptions(TypedDict, total=False):
|
||||
base_directory: str
|
||||
"""Directory where the CLI stores session state, configuration, and other persistent data."""
|
||||
|
||||
on_pre_tool_use: PreToolUseHandler
|
||||
"""Pre-tool-use hook handler for the Copilot SDK.
|
||||
|
||||
Called by the Copilot SDK before any tool is executed. The handler receives a
|
||||
``PreToolUseHookInput`` and a context dict, and returns a ``PreToolUseHookOutput``
|
||||
(or ``None`` to defer). Returning ``{"permissionDecision": "ask"}`` routes the
|
||||
decision to ``on_permission_request``; ``"allow"`` / ``"deny"`` gate the call
|
||||
directly.
|
||||
|
||||
If you do **not** supply this hook, the agent installs a default ``on_pre_tool_use``
|
||||
hook that returns ``"ask"`` for ``FunctionTool`` instances declared with
|
||||
``approval_mode="always_require"`` (deferring all other tools), so those tools are
|
||||
gated through ``on_permission_request``. If you **do** supply your own hook, it
|
||||
takes precedence and **you** are responsible for enforcing approval for any
|
||||
``always_require`` tool; the agent logs a warning naming such tools."""
|
||||
|
||||
on_function_approval: FunctionApprovalCallback
|
||||
"""Approval callback for ``FunctionTool`` instances declared with
|
||||
``approval_mode="always_require"``. The callback is awaited (sync or async)
|
||||
inside the SDK tool-handler before the tool is executed; a falsy return
|
||||
value denies the call. If omitted, calls to such tools are denied with an
|
||||
explanatory message returned to the model. This is independent of
|
||||
``on_permission_request``, which gates the Copilot SDK's built-in actions."""
|
||||
"""Deprecated approval callback for ``FunctionTool`` instances declared with
|
||||
``approval_mode="always_require"``.
|
||||
|
||||
.. deprecated::
|
||||
Use ``on_pre_tool_use`` together with ``on_permission_request`` instead.
|
||||
When neither this callback nor ``on_pre_tool_use`` is set, the agent
|
||||
installs a default ``on_pre_tool_use`` hook that returns ``"ask"`` for
|
||||
``always_require`` tools and routes the decision to ``on_permission_request``.
|
||||
|
||||
When set, this callback is enforced inside the SDK tool-handler before the tool
|
||||
runs; a falsy return value denies the call. Setting it emits a
|
||||
``DeprecationWarning``. It is **mutually exclusive** with ``on_pre_tool_use`` —
|
||||
setting both raises ``ValueError``."""
|
||||
|
||||
|
||||
OptionsT = TypeVar(
|
||||
@@ -319,9 +346,27 @@ class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
|
||||
mcp_servers: dict[str, MCPServerConfig] | None = opts.pop("mcp_servers", None)
|
||||
provider: ProviderConfig | None = opts.pop("provider", None)
|
||||
instruction_directories: list[str] | None = opts.pop("instruction_directories", None)
|
||||
on_pre_tool_use: PreToolUseHandler | None = opts.pop("on_pre_tool_use", None)
|
||||
on_function_approval: FunctionApprovalCallback | None = opts.pop("on_function_approval", None)
|
||||
base_directory = opts.pop("base_directory", None)
|
||||
|
||||
if on_function_approval is not None and on_pre_tool_use is not None:
|
||||
raise ValueError(
|
||||
"on_function_approval and on_pre_tool_use cannot both be set. "
|
||||
"on_function_approval is deprecated; use on_pre_tool_use together with "
|
||||
"on_permission_request instead."
|
||||
)
|
||||
|
||||
if on_function_approval is not None:
|
||||
warnings.warn(
|
||||
"on_function_approval is deprecated and will be removed in a future version. "
|
||||
"Use the SDK 'on_pre_tool_use' hook together with 'on_permission_request' instead: "
|
||||
"the default 'on_pre_tool_use' hook returns 'ask' for approval_mode='always_require' "
|
||||
"tools and routes the decision to 'on_permission_request'.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
||||
self._settings = load_settings(
|
||||
GitHubCopilotSettings,
|
||||
env_prefix="GITHUB_COPILOT_",
|
||||
@@ -336,6 +381,7 @@ class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
|
||||
|
||||
self._tools = normalize_tools(tools)
|
||||
self._permission_handler = on_permission_request
|
||||
self._on_pre_tool_use: PreToolUseHandler | None = on_pre_tool_use
|
||||
self._function_approval_handler: FunctionApprovalCallback | None = on_function_approval
|
||||
self._mcp_servers = mcp_servers
|
||||
self._provider = provider
|
||||
@@ -522,6 +568,12 @@ class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
|
||||
"via default_options at agent construction time. It cannot be overridden "
|
||||
"per run."
|
||||
)
|
||||
if "on_pre_tool_use" in opts and self._function_approval_handler is not None:
|
||||
raise ValueError(
|
||||
"on_pre_tool_use cannot be combined with the deprecated on_function_approval "
|
||||
"(set via default_options). Remove on_function_approval and use on_pre_tool_use "
|
||||
"together with on_permission_request instead."
|
||||
)
|
||||
timeout = opts.get("timeout") or self._settings.get("timeout") or DEFAULT_TIMEOUT_SECONDS
|
||||
|
||||
input_messages = normalize_messages(messages)
|
||||
@@ -611,6 +663,12 @@ class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
|
||||
"via default_options at agent construction time. It cannot be overridden "
|
||||
"per run."
|
||||
)
|
||||
if "on_pre_tool_use" in opts and self._function_approval_handler is not None:
|
||||
raise ValueError(
|
||||
"on_pre_tool_use cannot be combined with the deprecated on_function_approval "
|
||||
"(set via default_options). Remove on_function_approval and use on_pre_tool_use "
|
||||
"together with on_permission_request instead."
|
||||
)
|
||||
|
||||
input_messages = normalize_messages(messages)
|
||||
|
||||
@@ -792,31 +850,32 @@ class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
|
||||
return copilot_tools
|
||||
|
||||
def _tool_to_copilot_tool(self, ai_func: FunctionTool) -> CopilotTool:
|
||||
"""Convert an FunctionTool to a Copilot SDK tool."""
|
||||
"""Convert an FunctionTool to a Copilot SDK tool.
|
||||
|
||||
Approval for tools declared with ``approval_mode="always_require"`` is normally
|
||||
enforced by the Copilot SDK's native ``on_pre_tool_use`` hook (see
|
||||
:meth:`_build_session_hooks`). When the deprecated ``on_function_approval``
|
||||
callback is configured instead, approval is enforced inside this handler for
|
||||
backward compatibility. (``on_function_approval`` and ``on_pre_tool_use`` are
|
||||
mutually exclusive, so only one mechanism is ever active.)
|
||||
"""
|
||||
approval_handler = self._function_approval_handler
|
||||
requires_approval = ai_func.approval_mode == "always_require"
|
||||
enforce = approval_handler is not None and ai_func.approval_mode == "always_require"
|
||||
|
||||
async def handler(invocation: ToolInvocation) -> ToolResult:
|
||||
args: dict[str, Any] = invocation.arguments or {}
|
||||
try:
|
||||
if requires_approval and not await _resolve_function_approval(approval_handler, ai_func, args):
|
||||
deny_text = (
|
||||
f"Tool '{ai_func.name}' requires human approval "
|
||||
"(approval_mode='always_require') and the request was denied."
|
||||
if approval_handler is not None
|
||||
else (
|
||||
f"Tool '{ai_func.name}' requires human approval "
|
||||
"(approval_mode='always_require') but no on_function_approval "
|
||||
"callback is configured on the agent; the request was denied."
|
||||
)
|
||||
)
|
||||
if enforce and not await _resolve_function_approval(approval_handler, ai_func, args):
|
||||
logger.info(
|
||||
"Denying execution of tool '%s' (approval_mode='always_require', %s)",
|
||||
"Denying execution of tool '%s' (approval_mode='always_require', "
|
||||
"on_function_approval callback denied).",
|
||||
ai_func.name,
|
||||
"callback denied" if approval_handler is not None else "no callback configured",
|
||||
)
|
||||
return ToolResult(
|
||||
text_result_for_llm=deny_text,
|
||||
text_result_for_llm=(
|
||||
f"Tool '{ai_func.name}' requires human approval "
|
||||
"(approval_mode='always_require') and the request was denied."
|
||||
),
|
||||
result_type="failure",
|
||||
error="approval_denied",
|
||||
)
|
||||
@@ -850,6 +909,80 @@ class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
|
||||
parameters=ai_func.parameters(),
|
||||
)
|
||||
|
||||
def _build_session_hooks(
|
||||
self,
|
||||
all_tools: Sequence[ToolTypes | CopilotTool],
|
||||
opts: Mapping[str, Any],
|
||||
) -> SessionHooks | None:
|
||||
"""Build the ``SessionHooks`` to pass to the Copilot SDK for this session.
|
||||
|
||||
Approval enforcement for ``FunctionTool`` instances declared with
|
||||
``approval_mode="always_require"`` is delegated to the Copilot SDK's native
|
||||
``on_pre_tool_use`` hook:
|
||||
|
||||
- If the caller supplies their own ``on_pre_tool_use`` (via per-run ``options``
|
||||
or ``default_options``), it takes precedence and is returned unchanged. A
|
||||
warning is logged naming any approval-required tool that will therefore not
|
||||
be automatically gated, since the caller's hook is responsible for enforcing
|
||||
approval.
|
||||
- Otherwise, when any approval-required tool is present, a default hook is
|
||||
installed that returns ``"ask"`` for those tools (routing the decision to
|
||||
``on_permission_request``) and defers (``None``) for all other tools.
|
||||
- The default hook is **not** installed when the deprecated
|
||||
``on_function_approval`` callback is configured: in that case approval is
|
||||
enforced inside the tool handler (see :meth:`_tool_to_copilot_tool`) to
|
||||
preserve backward-compatible behavior.
|
||||
- When there are no approval-required tools and no caller hook, ``None`` is
|
||||
returned so no hooks are registered.
|
||||
|
||||
Args:
|
||||
all_tools: The full set of tools resolved for the session.
|
||||
opts: Runtime options that take precedence over ``default_options``.
|
||||
|
||||
Returns:
|
||||
The hooks to register for the session, or ``None`` if none are needed.
|
||||
"""
|
||||
user_hook: PreToolUseHandler | None = opts.get("on_pre_tool_use") or self._on_pre_tool_use
|
||||
|
||||
approval_required_names = {
|
||||
tool.name for tool in all_tools if isinstance(tool, FunctionTool) and tool.approval_mode == "always_require"
|
||||
}
|
||||
|
||||
if user_hook is not None:
|
||||
if approval_required_names:
|
||||
logger.warning(
|
||||
"A custom 'on_pre_tool_use' hook is configured, so %d approval-required tool(s) (%s) "
|
||||
"will not be automatically gated by GitHubCopilotAgent. The custom hook is responsible "
|
||||
"for enforcing approval (for example, by returning a 'deny' or 'ask' decision).",
|
||||
len(approval_required_names),
|
||||
", ".join(sorted(approval_required_names)),
|
||||
)
|
||||
return {"on_pre_tool_use": user_hook}
|
||||
|
||||
if not approval_required_names:
|
||||
return None
|
||||
|
||||
# The deprecated on_function_approval callback enforces approval in the tool
|
||||
# handler; don't also install the default ask-hook (which would double-gate).
|
||||
if self._function_approval_handler is not None:
|
||||
return None
|
||||
|
||||
def default_pre_tool_use(
|
||||
hook_input: Mapping[str, Any],
|
||||
_context: Mapping[str, str],
|
||||
) -> PreToolUseHookOutput | None:
|
||||
tool_name = hook_input.get("toolName")
|
||||
if tool_name in approval_required_names:
|
||||
return {
|
||||
"permissionDecision": "ask",
|
||||
"permissionDecisionReason": (
|
||||
f"Tool '{tool_name}' is marked as requiring approval (approval_mode='always_require')."
|
||||
),
|
||||
}
|
||||
return None
|
||||
|
||||
return {"on_pre_tool_use": default_pre_tool_use}
|
||||
|
||||
async def _get_or_create_session(
|
||||
self,
|
||||
agent_session: AgentSession,
|
||||
@@ -907,6 +1040,7 @@ class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
|
||||
instruction_directories = opts.get("instruction_directories", self._instruction_directories)
|
||||
all_tools = list(self._tools or []) + list(opts.get("tools") or [])
|
||||
tools = self._prepare_tools(all_tools) if all_tools else None
|
||||
hooks = self._build_session_hooks(all_tools, opts)
|
||||
|
||||
return await self._client.create_session(
|
||||
on_permission_request=permission_handler,
|
||||
@@ -917,6 +1051,7 @@ class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
|
||||
mcp_servers=mcp_servers or None,
|
||||
provider=provider or None,
|
||||
instruction_directories=instruction_directories,
|
||||
hooks=hooks,
|
||||
)
|
||||
|
||||
async def _resume_session(
|
||||
@@ -946,6 +1081,7 @@ class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
|
||||
instruction_directories = opts.get("instruction_directories", self._instruction_directories)
|
||||
all_tools = list(self._tools or []) + list(opts.get("tools") or [])
|
||||
tools = self._prepare_tools(all_tools) if all_tools else None
|
||||
hooks = self._build_session_hooks(all_tools, opts)
|
||||
|
||||
return await self._client.resume_session(
|
||||
session_id,
|
||||
@@ -957,6 +1093,7 @@ class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
|
||||
mcp_servers=mcp_servers or None,
|
||||
provider=provider or None,
|
||||
instruction_directories=instruction_directories,
|
||||
hooks=hooks,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
|
||||
import os
|
||||
import unittest.mock
|
||||
from collections.abc import Awaitable, Callable, Sequence
|
||||
from collections.abc import Sequence
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, cast
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
@@ -25,7 +25,7 @@ from agent_framework import (
|
||||
tool,
|
||||
)
|
||||
from agent_framework.exceptions import AgentException
|
||||
from copilot.session import PermissionHandler
|
||||
from copilot.session import PermissionHandler, PreToolUseHookInput
|
||||
from copilot.session_events import (
|
||||
Data,
|
||||
SessionEvent,
|
||||
@@ -43,6 +43,17 @@ def copilot_options(options: GitHubCopilotOptions) -> GitHubCopilotOptions:
|
||||
return options
|
||||
|
||||
|
||||
def pre_tool_use_input(tool_name: str) -> PreToolUseHookInput:
|
||||
"""Build a complete PreToolUseHookInput for exercising on_pre_tool_use hooks in tests."""
|
||||
return {
|
||||
"sessionId": "test-session",
|
||||
"timestamp": datetime.now(timezone.utc),
|
||||
"workingDirectory": ".",
|
||||
"toolName": tool_name,
|
||||
"toolArgs": {},
|
||||
}
|
||||
|
||||
|
||||
def create_session_event(
|
||||
event_type: SessionEventType,
|
||||
content: str | None = None,
|
||||
@@ -955,6 +966,7 @@ class TestGitHubCopilotAgentSessionManagement:
|
||||
mcp_servers=unittest.mock.ANY,
|
||||
provider=unittest.mock.ANY,
|
||||
instruction_directories=unittest.mock.ANY,
|
||||
hooks=unittest.mock.ANY,
|
||||
)
|
||||
|
||||
async def test_session_config_includes_model(
|
||||
@@ -1645,168 +1657,89 @@ class TestGitHubCopilotAgentToolConversion:
|
||||
|
||||
|
||||
class TestGitHubCopilotAgentFunctionApproval:
|
||||
"""Tests that ``approval_mode='always_require'`` is enforced at the agent boundary."""
|
||||
"""Tests that ``approval_mode='always_require'`` is gated via the SDK ``on_pre_tool_use`` hook."""
|
||||
|
||||
async def test_handler_denies_when_no_callback_configured(
|
||||
def test_default_hook_asks_for_approval_required_tool(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""Approval-required tool must be denied without executing when no callback is set."""
|
||||
from agent_framework import tool
|
||||
|
||||
invocations: list[Any] = []
|
||||
"""The default hook returns 'ask' for always_require tools and defers others."""
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def dangerous(path: str) -> str:
|
||||
"""A tool that requires human approval."""
|
||||
invocations.append(path)
|
||||
return f"deleted {path}"
|
||||
|
||||
@tool
|
||||
def safe(x: int) -> str:
|
||||
"""A tool that does not require approval."""
|
||||
return f"safe={x}"
|
||||
|
||||
agent = GitHubCopilotAgent(client=mock_client)
|
||||
copilot_tool = agent._tool_to_copilot_tool(dangerous) # type: ignore[reportPrivateUsage]
|
||||
hooks = agent._build_session_hooks([dangerous, safe], {}) # type: ignore[reportPrivateUsage]
|
||||
|
||||
handler = cast("Callable[[ToolInvocation], Awaitable[ToolResult]]", copilot_tool.handler)
|
||||
result = await handler(ToolInvocation(arguments={"path": "/critical"}))
|
||||
assert hooks is not None
|
||||
hook = hooks["on_pre_tool_use"]
|
||||
|
||||
assert invocations == []
|
||||
assert result.result_type == "failure"
|
||||
assert result.error == "approval_denied"
|
||||
assert "no on_function_approval callback is configured" in result.text_result_for_llm
|
||||
approval_decision = hook(pre_tool_use_input("dangerous"), {"session_id": "s"})
|
||||
assert approval_decision == {
|
||||
"permissionDecision": "ask",
|
||||
"permissionDecisionReason": (
|
||||
"Tool 'dangerous' is marked as requiring approval (approval_mode='always_require')."
|
||||
),
|
||||
}
|
||||
|
||||
async def test_handler_denies_when_callback_returns_false(
|
||||
assert hook(pre_tool_use_input("safe"), {"session_id": "s"}) is None
|
||||
|
||||
def test_no_hook_when_no_approval_required_tools(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""Falsy callback return value must deny the call and skip execution."""
|
||||
from agent_framework import Content, tool
|
||||
"""No approval-required tools and no user hook means no hooks are installed."""
|
||||
|
||||
invocations: list[Any] = []
|
||||
seen: list[Content] = []
|
||||
@tool
|
||||
def safe(x: int) -> str:
|
||||
"""A tool that does not require approval."""
|
||||
return f"safe={x}"
|
||||
|
||||
def deny(call: Content) -> bool:
|
||||
seen.append(call)
|
||||
return False
|
||||
agent = GitHubCopilotAgent(client=mock_client)
|
||||
assert agent._build_session_hooks([safe], {}) is None # type: ignore[reportPrivateUsage]
|
||||
|
||||
def test_user_hook_takes_precedence_and_warns(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""A caller-supplied on_pre_tool_use takes precedence and triggers a warning."""
|
||||
|
||||
def user_hook(_input: Any, _context: Any) -> Any:
|
||||
return {"permissionDecision": "allow"}
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def dangerous(path: str) -> str:
|
||||
"""A tool that requires human approval."""
|
||||
invocations.append(path)
|
||||
return f"deleted {path}"
|
||||
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": deny}),
|
||||
default_options=copilot_options({"on_pre_tool_use": user_hook}),
|
||||
)
|
||||
copilot_tool = agent._tool_to_copilot_tool(dangerous) # type: ignore[reportPrivateUsage]
|
||||
|
||||
handler = cast("Callable[[ToolInvocation], Awaitable[ToolResult]]", copilot_tool.handler)
|
||||
result = await handler(ToolInvocation(arguments={"path": "/critical"}))
|
||||
with caplog.at_level("WARNING", logger="agent_framework.github_copilot"):
|
||||
hooks = agent._build_session_hooks([dangerous], {}) # type: ignore[reportPrivateUsage]
|
||||
|
||||
assert invocations == []
|
||||
assert len(seen) == 1
|
||||
assert seen[0].type == "function_call"
|
||||
assert seen[0].name == "dangerous" # type: ignore[attr-defined]
|
||||
assert seen[0].arguments == {"path": "/critical"} # type: ignore[attr-defined]
|
||||
assert result.result_type == "failure"
|
||||
assert result.error == "approval_denied"
|
||||
assert hooks == {"on_pre_tool_use": user_hook}
|
||||
assert any("dangerous" in record.message and record.levelname == "WARNING" for record in caplog.records)
|
||||
|
||||
async def test_handler_executes_when_callback_returns_true(
|
||||
def test_user_hook_no_warning_without_approval_tools(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Truthy callback return value must allow the tool to execute normally."""
|
||||
from agent_framework import Content, tool
|
||||
"""A caller hook with no approval-required tools is preserved without a warning."""
|
||||
|
||||
def approve(call: Content) -> bool:
|
||||
return True
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def guarded(x: int) -> str:
|
||||
"""A tool that requires human approval."""
|
||||
return f"result={x}"
|
||||
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": approve}),
|
||||
)
|
||||
copilot_tool = agent._tool_to_copilot_tool(guarded) # type: ignore[reportPrivateUsage]
|
||||
|
||||
handler = cast("Callable[[ToolInvocation], Awaitable[ToolResult]]", copilot_tool.handler)
|
||||
result = await handler(ToolInvocation(arguments={"x": 42}))
|
||||
|
||||
assert result.result_type == "success"
|
||||
assert result.text_result_for_llm == "result=42"
|
||||
|
||||
async def test_handler_supports_async_callback(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""Async callback must be awaited and respected."""
|
||||
from agent_framework import Content, tool
|
||||
|
||||
async def approve(call: Content) -> bool:
|
||||
return True
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def guarded(x: int) -> str:
|
||||
"""A tool that requires human approval."""
|
||||
return f"async={x}"
|
||||
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": approve}),
|
||||
)
|
||||
copilot_tool = agent._tool_to_copilot_tool(guarded) # type: ignore[reportPrivateUsage]
|
||||
|
||||
handler = cast("Callable[[ToolInvocation], Awaitable[ToolResult]]", copilot_tool.handler)
|
||||
result = await handler(ToolInvocation(arguments={"x": 7}))
|
||||
|
||||
assert result.result_type == "success"
|
||||
assert result.text_result_for_llm == "async=7"
|
||||
|
||||
async def test_callback_failure_denies_safely(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""A callback that raises must result in denial, not in tool execution."""
|
||||
from agent_framework import Content, tool
|
||||
|
||||
invocations: list[Any] = []
|
||||
|
||||
def boom(call: Content) -> bool:
|
||||
raise RuntimeError("nope")
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def dangerous(x: int) -> str:
|
||||
"""A tool that requires human approval."""
|
||||
invocations.append(x)
|
||||
return f"x={x}"
|
||||
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": boom}),
|
||||
)
|
||||
copilot_tool = agent._tool_to_copilot_tool(dangerous) # type: ignore[reportPrivateUsage]
|
||||
|
||||
handler = cast("Callable[[ToolInvocation], Awaitable[ToolResult]]", copilot_tool.handler)
|
||||
result = await handler(ToolInvocation(arguments={"x": 1}))
|
||||
|
||||
assert invocations == []
|
||||
assert result.result_type == "failure"
|
||||
assert result.error == "approval_denied"
|
||||
|
||||
async def test_handler_does_not_invoke_callback_for_never_require(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""Tools without approval_mode='always_require' must not trigger the callback."""
|
||||
from agent_framework import Content, tool
|
||||
|
||||
callback_calls: list[Any] = []
|
||||
|
||||
def approve(call: Content) -> bool:
|
||||
callback_calls.append(call)
|
||||
return True
|
||||
def user_hook(_input: Any, _context: Any) -> Any:
|
||||
return None
|
||||
|
||||
@tool
|
||||
def safe(x: int) -> str:
|
||||
@@ -1815,16 +1748,236 @@ class TestGitHubCopilotAgentFunctionApproval:
|
||||
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": approve}),
|
||||
default_options=copilot_options({"on_pre_tool_use": user_hook}),
|
||||
)
|
||||
copilot_tool = agent._tool_to_copilot_tool(safe) # type: ignore[reportPrivateUsage]
|
||||
|
||||
handler = cast("Callable[[ToolInvocation], Awaitable[ToolResult]]", copilot_tool.handler)
|
||||
result = await handler(ToolInvocation(arguments={"x": 5}))
|
||||
with caplog.at_level("WARNING", logger="agent_framework.github_copilot"):
|
||||
hooks = agent._build_session_hooks([safe], {}) # type: ignore[reportPrivateUsage]
|
||||
|
||||
assert hooks == {"on_pre_tool_use": user_hook}
|
||||
assert not any(record.levelname == "WARNING" for record in caplog.records)
|
||||
|
||||
def test_runtime_on_pre_tool_use_overrides_default_options(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""A per-run on_pre_tool_use option takes precedence over default_options."""
|
||||
|
||||
def default_hook(_input: Any, _context: Any) -> Any:
|
||||
return None
|
||||
|
||||
def runtime_hook(_input: Any, _context: Any) -> Any:
|
||||
return {"permissionDecision": "deny"}
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def dangerous(path: str) -> str:
|
||||
"""A tool that requires human approval."""
|
||||
return f"deleted {path}"
|
||||
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_pre_tool_use": default_hook}),
|
||||
)
|
||||
|
||||
hooks = agent._build_session_hooks([dangerous], {"on_pre_tool_use": runtime_hook}) # type: ignore[reportPrivateUsage]
|
||||
assert hooks == {"on_pre_tool_use": runtime_hook}
|
||||
|
||||
async def test_default_hook_forwarded_to_create_session(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
mock_session: MagicMock,
|
||||
assistant_message_event: SessionEvent,
|
||||
) -> None:
|
||||
"""An always_require tool causes the default hook to be forwarded to the SDK session."""
|
||||
mock_session.send_and_wait.return_value = assistant_message_event
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def dangerous(path: str) -> str:
|
||||
"""A tool that requires human approval."""
|
||||
return f"deleted {path}"
|
||||
|
||||
agent = GitHubCopilotAgent(client=mock_client, tools=[dangerous])
|
||||
await agent.run("hello")
|
||||
|
||||
hooks = mock_client.create_session.call_args.kwargs["hooks"]
|
||||
assert hooks is not None
|
||||
assert "on_pre_tool_use" in hooks
|
||||
|
||||
|
||||
class TestGitHubCopilotAgentDeprecatedFunctionApproval:
|
||||
"""Tests for the deprecated ``on_function_approval`` callback (still enforced)."""
|
||||
|
||||
def test_setting_callback_emits_deprecation_warning(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""Configuring on_function_approval emits a DeprecationWarning."""
|
||||
|
||||
def approve(_call: Content) -> bool:
|
||||
return True
|
||||
|
||||
with pytest.warns(DeprecationWarning, match="on_function_approval is deprecated"):
|
||||
GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": approve}),
|
||||
)
|
||||
|
||||
async def test_handler_denies_when_callback_returns_false(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""A falsy callback return value denies the call and skips execution."""
|
||||
invocations: list[str] = []
|
||||
|
||||
def deny(_call: Content) -> bool:
|
||||
return False
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def dangerous(path: str) -> str:
|
||||
"""A tool that requires human approval."""
|
||||
invocations.append(path)
|
||||
return f"deleted {path}"
|
||||
|
||||
with pytest.warns(DeprecationWarning):
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": deny}),
|
||||
)
|
||||
copilot_tool = agent._tool_to_copilot_tool(dangerous) # type: ignore[reportPrivateUsage]
|
||||
|
||||
handler = cast("Any", copilot_tool.handler)
|
||||
result = await handler(ToolInvocation(arguments={"path": "/critical"}))
|
||||
|
||||
assert invocations == []
|
||||
assert result.result_type == "failure"
|
||||
assert result.error == "approval_denied"
|
||||
|
||||
async def test_handler_executes_when_callback_returns_true(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""A truthy callback return value allows the tool to execute."""
|
||||
|
||||
def approve(_call: Content) -> bool:
|
||||
return True
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def guarded(x: int) -> str:
|
||||
"""A tool that requires human approval."""
|
||||
return f"result={x}"
|
||||
|
||||
with pytest.warns(DeprecationWarning):
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": approve}),
|
||||
)
|
||||
copilot_tool = agent._tool_to_copilot_tool(guarded) # type: ignore[reportPrivateUsage]
|
||||
|
||||
handler = cast("Any", copilot_tool.handler)
|
||||
result = await handler(ToolInvocation(arguments={"x": 42}))
|
||||
|
||||
assert callback_calls == []
|
||||
assert result.result_type == "success"
|
||||
assert result.text_result_for_llm == "safe=5"
|
||||
assert result.text_result_for_llm == "result=42"
|
||||
|
||||
def test_default_hook_not_installed_when_callback_set(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""When on_function_approval is set, the default ask-hook is not installed."""
|
||||
|
||||
def approve(_call: Content) -> bool:
|
||||
return True
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def dangerous(path: str) -> str:
|
||||
"""A tool that requires human approval."""
|
||||
return f"deleted {path}"
|
||||
|
||||
with pytest.warns(DeprecationWarning):
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": approve}),
|
||||
)
|
||||
|
||||
assert agent._build_session_hooks([dangerous], {}) is None # type: ignore[reportPrivateUsage]
|
||||
|
||||
def test_both_options_in_default_options_raises(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""Setting both on_function_approval and on_pre_tool_use at construction raises."""
|
||||
|
||||
def deny(_call: Content) -> bool:
|
||||
return False
|
||||
|
||||
def hook(_input: Any, _context: Any) -> Any:
|
||||
return None
|
||||
|
||||
with pytest.raises(ValueError, match="cannot both be set"):
|
||||
GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": deny, "on_pre_tool_use": hook}),
|
||||
)
|
||||
|
||||
async def test_runtime_on_pre_tool_use_with_deprecated_callback_raises(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""A per-run on_pre_tool_use combined with a construction-time on_function_approval raises."""
|
||||
|
||||
def deny(_call: Content) -> bool:
|
||||
return False
|
||||
|
||||
def allow_hook(_input: Any, _context: Any) -> Any:
|
||||
return {"permissionDecision": "allow"}
|
||||
|
||||
with pytest.warns(DeprecationWarning):
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": deny}),
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="cannot be combined with the deprecated on_function_approval"):
|
||||
await agent.run("hello", options=cast(Any, {"on_pre_tool_use": allow_hook}))
|
||||
|
||||
async def test_runtime_on_pre_tool_use_with_deprecated_callback_raises_streaming(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
) -> None:
|
||||
"""The mutual-exclusivity check also applies on the streaming path."""
|
||||
|
||||
def deny(_call: Content) -> bool:
|
||||
return False
|
||||
|
||||
def allow_hook(_input: Any, _context: Any) -> Any:
|
||||
return {"permissionDecision": "allow"}
|
||||
|
||||
with pytest.warns(DeprecationWarning):
|
||||
agent = GitHubCopilotAgent(
|
||||
client=mock_client,
|
||||
default_options=copilot_options({"on_function_approval": deny}),
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="cannot be combined with the deprecated on_function_approval"):
|
||||
async for _ in agent.run("hello", stream=True, options=cast(Any, {"on_pre_tool_use": allow_hook})):
|
||||
pass
|
||||
|
||||
async def test_runtime_on_function_approval_rejected(self, mock_client: MagicMock) -> None:
|
||||
"""Passing on_function_approval at runtime raises rather than being silently ignored."""
|
||||
agent = GitHubCopilotAgent(client=mock_client)
|
||||
with pytest.raises(ValueError, match="on_function_approval"):
|
||||
await agent.run("hello", options=cast(Any, {"on_function_approval": lambda _c: True}))
|
||||
|
||||
async def test_runtime_on_function_approval_rejected_streaming(self, mock_client: MagicMock) -> None:
|
||||
"""Passing on_function_approval at runtime raises on the streaming path too."""
|
||||
agent = GitHubCopilotAgent(client=mock_client)
|
||||
with pytest.raises(ValueError, match="on_function_approval"):
|
||||
async for _ in agent.run(
|
||||
"hello",
|
||||
stream=True,
|
||||
options=cast(Any, {"on_function_approval": lambda _c: True}),
|
||||
):
|
||||
pass
|
||||
|
||||
|
||||
class TestGitHubCopilotAgentErrorHandling:
|
||||
@@ -2491,22 +2644,54 @@ class TestGitHubCopilotAgentContextProviders:
|
||||
|
||||
assert observed_options.get("timeout") == 120
|
||||
|
||||
async def test_runtime_on_function_approval_rejected(self, mock_client: MagicMock) -> None:
|
||||
"""Passing on_function_approval at runtime must raise rather than be silently ignored."""
|
||||
agent = GitHubCopilotAgent(client=mock_client)
|
||||
with pytest.raises(ValueError, match="on_function_approval"):
|
||||
await agent.run("hello", options=cast(Any, {"on_function_approval": lambda _c: True}))
|
||||
async def test_runtime_on_pre_tool_use_forwarded(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
mock_session: MagicMock,
|
||||
assistant_message_event: SessionEvent,
|
||||
) -> None:
|
||||
"""Passing on_pre_tool_use at runtime is accepted and forwarded to the session."""
|
||||
mock_session.send_and_wait.return_value = assistant_message_event
|
||||
|
||||
def runtime_hook(_input: Any, _context: Any) -> Any:
|
||||
return {"permissionDecision": "deny"}
|
||||
|
||||
async def test_runtime_on_function_approval_rejected_streaming(self, mock_client: MagicMock) -> None:
|
||||
"""Passing on_function_approval at runtime must raise on the streaming path too."""
|
||||
agent = GitHubCopilotAgent(client=mock_client)
|
||||
with pytest.raises(ValueError, match="on_function_approval"):
|
||||
async for _ in agent.run(
|
||||
"hello",
|
||||
stream=True,
|
||||
options=cast(Any, {"on_function_approval": lambda _c: True}),
|
||||
):
|
||||
pass
|
||||
await agent.run("hello", options=cast(Any, {"on_pre_tool_use": runtime_hook}))
|
||||
|
||||
hooks = mock_client.create_session.call_args.kwargs["hooks"]
|
||||
assert hooks == {"on_pre_tool_use": runtime_hook}
|
||||
|
||||
async def test_runtime_on_pre_tool_use_forwarded_streaming(
|
||||
self,
|
||||
mock_client: MagicMock,
|
||||
mock_session: MagicMock,
|
||||
assistant_delta_event: SessionEvent,
|
||||
session_idle_event: SessionEvent,
|
||||
) -> None:
|
||||
"""Passing on_pre_tool_use at runtime is accepted on the streaming path too."""
|
||||
events = [assistant_delta_event, session_idle_event]
|
||||
|
||||
def mock_on(handler: Any) -> Any:
|
||||
for event in events:
|
||||
handler(event)
|
||||
return lambda: None
|
||||
|
||||
mock_session.on = mock_on
|
||||
|
||||
def runtime_hook(_input: Any, _context: Any) -> Any:
|
||||
return {"permissionDecision": "deny"}
|
||||
|
||||
agent = GitHubCopilotAgent(client=mock_client)
|
||||
async for _ in agent.run(
|
||||
"hello",
|
||||
stream=True,
|
||||
options=cast(Any, {"on_pre_tool_use": runtime_hook}),
|
||||
):
|
||||
pass
|
||||
|
||||
hooks = mock_client.create_session.call_args.kwargs["hooks"]
|
||||
assert hooks == {"on_pre_tool_use": runtime_hook}
|
||||
|
||||
async def test_provider_tools_forwarded_to_session(
|
||||
self,
|
||||
|
||||
Reference in New Issue
Block a user