Spaces:
Running
Running
File size: 8,759 Bytes
1a49d90 8612b4f 1a49d90 8612b4f 1a49d90 8612b4f 1a49d90 8612b4f 1a49d90 8612b4f 1a49d90 8612b4f 1a49d90 e9d44f5 8612b4f 1a49d90 8612b4f 1a49d90 e9d44f5 8612b4f 1a49d90 8612b4f 1a49d90 8612b4f 1a49d90 8612b4f 1a49d90 8612b4f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 | """
ComponentService: Provides async management of tools, resources, and prompts for FastMCP servers.
Handles enabling/disabling components both locally and across mounted servers.
"""
from fastmcp.exceptions import NotFoundError
from fastmcp.prompts.prompt import Prompt
from fastmcp.resources.resource import Resource
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.server import FastMCP, has_resource_prefix, remove_resource_prefix
from fastmcp.tools.tool import Tool
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
class ComponentService:
"""Service for managing components like tools, resources, and prompts."""
def __init__(self, server: FastMCP):
self._server = server
self._tool_manager = server._tool_manager
self._resource_manager = server._resource_manager
self._prompt_manager = server._prompt_manager
async def _enable_tool(self, key: str) -> Tool:
"""Handle 'enableTool' requests.
Args:
key: The key of the tool to enable
Returns:
The tool that was enabled
"""
logger.debug("Enabling tool: %s", key)
# 1. Check local tools first. The server will have already applied its filter.
if key in self._server._tool_manager._tools:
tool: Tool = await self._server.get_tool(key)
tool.enable()
return tool
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._tool_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
tool_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
tool = await mounted_service._enable_tool(tool_key)
return tool
else:
continue
raise NotFoundError(f"Unknown tool: {key}")
async def _disable_tool(self, key: str) -> Tool:
"""Handle 'disableTool' requests.
Args:
key: The key of the tool to disable
Returns:
The tool that was disabled
"""
logger.debug("Disable tool: %s", key)
# 1. Check local tools first. The server will have already applied its filter.
if key in self._server._tool_manager._tools:
tool: Tool = await self._server.get_tool(key)
tool.disable()
return tool
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._tool_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
tool_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
tool = await mounted_service._disable_tool(tool_key)
return tool
else:
continue
raise NotFoundError(f"Unknown tool: {key}")
async def _enable_resource(self, key: str) -> Resource | ResourceTemplate:
"""Handle 'enableResource' requests.
Args:
key: The key of the resource to enable
Returns:
The resource that was enabled
"""
logger.debug("Enabling resource: %s", key)
# 1. Check local resources first. The server will have already applied its filter.
if key in self._resource_manager._resources:
resource: Resource = await self._server.get_resource(key)
resource.enable()
return resource
if key in self._resource_manager._templates:
template: ResourceTemplate = await self._server.get_resource_template(key)
template.enable()
return template
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._resource_manager._mounted_servers):
if mounted.prefix:
if has_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
):
key = remove_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
)
mounted_service = ComponentService(mounted.server)
mounted_resource: (
Resource | ResourceTemplate
) = await mounted_service._enable_resource(key)
return mounted_resource
else:
continue
raise NotFoundError(f"Unknown resource: {key}")
async def _disable_resource(self, key: str) -> Resource | ResourceTemplate:
"""Handle 'disableResource' requests.
Args:
key: The key of the resource to disable
Returns:
The resource that was disabled
"""
logger.debug("Disable resource: %s", key)
# 1. Check local resources first. The server will have already applied its filter.
if key in self._resource_manager._resources:
resource: Resource = await self._server.get_resource(key)
resource.disable()
return resource
if key in self._resource_manager._templates:
template: ResourceTemplate = await self._server.get_resource_template(key)
template.disable()
return template
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._resource_manager._mounted_servers):
if mounted.prefix:
if has_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
):
key = remove_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
)
mounted_service = ComponentService(mounted.server)
mounted_resource: (
Resource | ResourceTemplate
) = await mounted_service._disable_resource(key)
return mounted_resource
else:
continue
raise NotFoundError(f"Unknown resource: {key}")
async def _enable_prompt(self, key: str) -> Prompt:
"""Handle 'enablePrompt' requests.
Args:
key: The key of the prompt to enable
Returns:
The prompt that was enable
"""
logger.debug("Enabling prompt: %s", key)
# 1. Check local prompts first. The server will have already applied its filter.
if key in self._server._prompt_manager._prompts:
prompt: Prompt = await self._server.get_prompt(key)
prompt.enable()
return prompt
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._prompt_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
prompt_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
prompt = await mounted_service._enable_prompt(prompt_key)
return prompt
else:
continue
raise NotFoundError(f"Unknown prompt: {key}")
async def _disable_prompt(self, key: str) -> Prompt:
"""Handle 'disablePrompt' requests.
Args:
key: The key of the prompt to disable
Returns:
The prompt that was disabled
"""
# 1. Check local prompts first. The server will have already applied its filter.
if key in self._server._prompt_manager._prompts:
prompt: Prompt = await self._server.get_prompt(key)
prompt.disable()
return prompt
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._prompt_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
prompt_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
prompt = await mounted_service._disable_prompt(prompt_key)
return prompt
else:
continue
raise NotFoundError(f"Unknown prompt: {key}")
|