Coverage for mcpgateway / plugins / policy.py: 100%

1# -*- coding: utf-8 -*- 

2"""Location: ./mcpgateway/plugins/policy.py 

3Copyright 2026 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Fred Araujo 

6 

7Concrete hook payload policies for the gateway. 

8 

9This module defines which payload fields plugins are allowed to modify 

10for each hook type. The policies are injected into the PluginManager 

11at initialization time. 

12 

13Examples: 

14 >>> from mcpgateway.plugins.policy import HOOK_PAYLOAD_POLICIES 

15 >>> "name" in HOOK_PAYLOAD_POLICIES["tool_pre_invoke"].writable_fields 

16 True 

17 >>> "result" in HOOK_PAYLOAD_POLICIES["tool_post_invoke"].writable_fields 

18 True 

19""" 

20 

21# First-Party 

22from mcpgateway.plugins.framework.hooks.policies import HookPayloadPolicy 

23 

24HOOK_PAYLOAD_POLICIES: dict[str, HookPayloadPolicy] = { 

25 # Tools 

26 "tool_pre_invoke": HookPayloadPolicy(writable_fields=frozenset({"name", "args", "headers"})), 

27 "tool_post_invoke": HookPayloadPolicy(writable_fields=frozenset({"result"})), 

28 # Prompts 

29 "prompt_pre_fetch": HookPayloadPolicy(writable_fields=frozenset({"args"})), 

30 "prompt_post_fetch": HookPayloadPolicy(writable_fields=frozenset({"result"})), 

31 # Resources 

32 "resource_pre_fetch": HookPayloadPolicy(writable_fields=frozenset({"uri", "metadata"})), 

33 "resource_post_fetch": HookPayloadPolicy(writable_fields=frozenset({"content"})), 

34 # Agents 

35 "agent_pre_invoke": HookPayloadPolicy(writable_fields=frozenset({"agent_id", "messages", "tools", "model", "system_prompt", "parameters", "headers"})), 

36 "agent_post_invoke": HookPayloadPolicy(writable_fields=frozenset({"messages", "tool_calls"})), 

37 # HTTP hooks (cross-type results — input and output payload types differ, 

38 # so field-level filtering is not applicable; policy presence authorises 

39 # the hook so it is never subject to default_hook_policy=deny). 

40 "http_pre_request": HookPayloadPolicy(writable_fields=frozenset({"headers"})), 

41 "http_post_request": HookPayloadPolicy(writable_fields=frozenset({"headers"})), 

42 "http_auth_resolve_user": HookPayloadPolicy(writable_fields=frozenset()), 

43 "http_auth_check_permission": HookPayloadPolicy(writable_fields=frozenset({"reason"})), 

44}