Coverage for mcpgateway / plugins / framework / hooks / resources.py: 96%

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

2"""Location: ./mcpgateway/plugins/framework/hooks/resources.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Teryl Taylor 

6 

7Pydantic models for resource hooks. 

8""" 

9 

10# Standard 

11from enum import Enum 

12from typing import Any, Optional 

13 

14# Third-Party 

15from pydantic import Field 

16 

17# First-Party 

18from mcpgateway.plugins.framework.models import PluginPayload, PluginResult 

19 

20 

21class ResourceHookType(str, Enum): 

22 """MCP Forge Gateway resource hook points. 

23 

24 Attributes: 

25 resource_pre_fetch: The resource pre fetch hook. 

26 resource_post_fetch: The resource post fetch hook. 

27 

28 Examples: 

29 >>> ResourceHookType.RESOURCE_PRE_FETCH 

30 <ResourceHookType.RESOURCE_PRE_FETCH: 'resource_pre_fetch'> 

31 >>> ResourceHookType.RESOURCE_PRE_FETCH.value 

32 'resource_pre_fetch' 

33 >>> ResourceHookType('resource_post_fetch') 

34 <ResourceHookType.RESOURCE_POST_FETCH: 'resource_post_fetch'> 

35 >>> list(ResourceHookType) 

36 [<ResourceHookType.RESOURCE_PRE_FETCH: 'resource_pre_fetch'>, <ResourceHookType.RESOURCE_POST_FETCH: 'resource_post_fetch'>] 

37 """ 

38 

39 RESOURCE_PRE_FETCH = "resource_pre_fetch" 

40 RESOURCE_POST_FETCH = "resource_post_fetch" 

41 

42 

43class ResourcePreFetchPayload(PluginPayload): 

44 """A resource payload for a resource pre-fetch hook. 

45 

46 Attributes: 

47 uri: The resource URI. 

48 metadata: Optional metadata for the resource request. 

49 

50 Examples: 

51 >>> payload = ResourcePreFetchPayload(uri="file:///data.txt") 

52 >>> payload.uri 

53 'file:///data.txt' 

54 >>> payload2 = ResourcePreFetchPayload(uri="http://api/data", metadata={"Accept": "application/json"}) 

55 >>> payload2.metadata 

56 {'Accept': 'application/json'} 

57 >>> p = ResourcePreFetchPayload(uri="file:///docs/readme.md", metadata={"version": "1.0"}) 

58 >>> p.uri 

59 'file:///docs/readme.md' 

60 >>> p.metadata["version"] 

61 '1.0' 

62 """ 

63 

64 uri: str 

65 metadata: Optional[dict[str, Any]] = Field(default_factory=dict) 

66 

67 

68class ResourcePostFetchPayload(PluginPayload): 

69 """A resource payload for a resource post-fetch hook. 

70 

71 Attributes: 

72 uri: The resource URI. 

73 content: The fetched resource content. 

74 

75 Examples: 

76 >>> from mcpgateway.common.models import ResourceContent 

77 >>> content = ResourceContent(type="resource", id="res-1", uri="file:///data.txt", 

78 ... text="Hello World") 

79 >>> payload = ResourcePostFetchPayload(uri="file:///data.txt", content=content) 

80 >>> payload.uri 

81 'file:///data.txt' 

82 >>> payload.content.text 

83 'Hello World' 

84 >>> from mcpgateway.common.models import ResourceContent 

85 >>> resource_content = ResourceContent(type="resource", id="res-2", uri="test://resource", text="Test data") 

86 >>> p = ResourcePostFetchPayload(uri="test://resource", content=resource_content) 

87 >>> p.uri 

88 'test://resource' 

89 """ 

90 

91 uri: str 

92 content: Any 

93 

94 

95ResourcePreFetchResult = PluginResult[ResourcePreFetchPayload] 

96ResourcePostFetchResult = PluginResult[ResourcePostFetchPayload] 

97 

98 

99def _register_resource_hooks() -> None: 

100 """Register resource hooks in the global registry. 

101 

102 This is called lazily to avoid circular import issues. 

103 """ 

104 # Import here to avoid circular dependency at module load time 

105 # First-Party 

106 from mcpgateway.plugins.framework.hooks.registry import get_hook_registry # pylint: disable=import-outside-toplevel 

107 

108 registry = get_hook_registry() 

109 

110 # Only register if not already registered (idempotent) 

111 if not registry.is_registered(ResourceHookType.RESOURCE_PRE_FETCH): 111 ↛ exitline 111 didn't return from function '_register_resource_hooks' because the condition on line 111 was always true

112 registry.register_hook(ResourceHookType.RESOURCE_PRE_FETCH, ResourcePreFetchPayload, ResourcePreFetchResult) 

113 registry.register_hook(ResourceHookType.RESOURCE_POST_FETCH, ResourcePostFetchPayload, ResourcePostFetchResult) 

114 

115 

116_register_resource_hooks()