Jeremiah Lowin commited on
Commit
a32223b
·
1 Parent(s): d37c5e4

Clean up mode kwarg

Browse files
src/fastmcp/prompts/prompt_manager.py CHANGED
@@ -2,7 +2,7 @@ from __future__ import annotations as _annotations
2
 
3
  import warnings
4
  from collections.abc import Awaitable, Callable
5
- from typing import TYPE_CHECKING, Any, Literal
6
 
7
  from mcp import GetPromptResult
8
 
@@ -46,25 +46,23 @@ class PromptManager:
46
  """Adds a mounted server as a source for prompts."""
47
  self._mounted_sources.append(server)
48
 
49
- async def _load_prompts(
50
- self, *, mode: Literal["inventory", "protocol"]
51
- ) -> dict[str, Prompt]:
52
  """
53
- The single, consolidated recursive method for fetching prompts. The 'mode'
54
  parameter determines the communication path.
55
 
56
- - mode="inventory": Manager-to-manager path for complete, unfiltered inventory
57
- - mode="protocol": Server-to-server path for filtered MCP requests
58
  """
59
  all_prompts: dict[str, Prompt] = {}
60
 
61
  for mounted in self._mounted_sources:
62
  try:
63
- if mode == "protocol":
64
- # PATH 2: Use the server-to-server filtered path
65
  child_results = await mounted.server._list_prompts()
66
- else: # mode == "inventory"
67
- # PATH 1: Use the manager-to-manager unfiltered path
68
  child_results = await mounted.server._prompt_manager._list_prompts()
69
 
70
  # The combination logic is the same for both paths
@@ -104,13 +102,13 @@ class PromptManager:
104
  """
105
  Gets the complete, unfiltered inventory of all prompts.
106
  """
107
- return await self._load_prompts(mode="inventory")
108
 
109
  async def _list_prompts(self) -> list[Prompt]:
110
  """
111
  Lists all prompts, applying protocol filtering.
112
  """
113
- prompts_dict = await self._load_prompts(mode="protocol")
114
  return list(prompts_dict.values())
115
 
116
  def add_prompt_from_fn(
 
2
 
3
  import warnings
4
  from collections.abc import Awaitable, Callable
5
+ from typing import TYPE_CHECKING, Any
6
 
7
  from mcp import GetPromptResult
8
 
 
46
  """Adds a mounted server as a source for prompts."""
47
  self._mounted_sources.append(server)
48
 
49
+ async def _load_prompts(self, *, via_server: bool = False) -> dict[str, Prompt]:
 
 
50
  """
51
+ The single, consolidated recursive method for fetching prompts. The 'via_server'
52
  parameter determines the communication path.
53
 
54
+ - via_server=False: Manager-to-manager path for complete, unfiltered inventory
55
+ - via_server=True: Server-to-server path for filtered MCP requests
56
  """
57
  all_prompts: dict[str, Prompt] = {}
58
 
59
  for mounted in self._mounted_sources:
60
  try:
61
+ if via_server:
62
+ # Use the server-to-server filtered path
63
  child_results = await mounted.server._list_prompts()
64
+ else:
65
+ # Use the manager-to-manager unfiltered path
66
  child_results = await mounted.server._prompt_manager._list_prompts()
67
 
68
  # The combination logic is the same for both paths
 
102
  """
103
  Gets the complete, unfiltered inventory of all prompts.
104
  """
105
+ return await self._load_prompts(via_server=False)
106
 
107
  async def _list_prompts(self) -> list[Prompt]:
108
  """
109
  Lists all prompts, applying protocol filtering.
110
  """
111
+ prompts_dict = await self._load_prompts(via_server=True)
112
  return list(prompts_dict.values())
113
 
114
  def add_prompt_from_fn(
src/fastmcp/resources/resource_manager.py CHANGED
@@ -5,7 +5,7 @@ from __future__ import annotations
5
  import inspect
6
  import warnings
7
  from collections.abc import Callable
8
- from typing import TYPE_CHECKING, Any, Literal
9
 
10
  from pydantic import AnyUrl
11
 
@@ -63,34 +63,32 @@ class ResourceManager:
63
 
64
  async def get_resources(self) -> dict[str, Resource]:
65
  """Get all registered resources, keyed by URI."""
66
- return await self._load_resources(mode="inventory")
67
 
68
  async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
69
  """Get all registered templates, keyed by URI template."""
70
- return await self._load_resource_templates(mode="inventory")
71
 
72
- async def _load_resources(
73
- self, *, mode: Literal["inventory", "protocol"]
74
- ) -> dict[str, Resource]:
75
  """
76
- The single, consolidated recursive method for fetching resources. The 'mode'
77
  parameter determines the communication path.
78
 
79
- - mode="inventory": Manager-to-manager path for complete, unfiltered inventory
80
- - mode="protocol": Server-to-server path for filtered MCP requests
81
  """
82
  all_resources: dict[str, Resource] = {}
83
 
84
  for mounted in self._mounted_sources:
85
  try:
86
- if mode == "protocol":
87
- # PATH 2: Use the server-to-server filtered path
88
  child_resources_list = await mounted.server._list_resources()
89
  child_resources = {
90
  resource.key: resource for resource in child_resources_list
91
  }
92
- else: # mode == "inventory"
93
- # PATH 1: Use the manager-to-manager unfiltered path
94
  child_resources = (
95
  await mounted.server._resource_manager.get_resources()
96
  )
@@ -120,24 +118,24 @@ class ResourceManager:
120
  return all_resources
121
 
122
  async def _load_resource_templates(
123
- self, *, mode: Literal["inventory", "protocol"]
124
  ) -> dict[str, ResourceTemplate]:
125
  """
126
- The single, consolidated recursive method for fetching templates. The 'mode'
127
  parameter determines the communication path.
128
 
129
- - mode="inventory": Manager-to-manager path for complete, unfiltered inventory
130
- - mode="protocol": Server-to-server path for filtered MCP requests
131
  """
132
  all_templates: dict[str, ResourceTemplate] = {}
133
 
134
  for mounted in self._mounted_sources:
135
  try:
136
- if mode == "protocol":
137
- # PATH 2: Use the server-to-server filtered path
138
  child_templates = await mounted.server._list_resource_templates()
139
- else: # mode == "inventory"
140
- # PATH 1: Use the manager-to-manager unfiltered path
141
  child_templates = await mounted.server._resource_manager._list_resource_templates()
142
  child_dict = {template.key: template for template in child_templates}
143
 
@@ -169,14 +167,14 @@ class ResourceManager:
169
  """
170
  Lists all resources, applying protocol filtering.
171
  """
172
- resources_dict = await self._load_resources(mode="protocol")
173
  return list(resources_dict.values())
174
 
175
  async def _list_resource_templates(self) -> list[ResourceTemplate]:
176
  """
177
  Lists all templates, applying protocol filtering.
178
  """
179
- templates_dict = await self._load_resource_templates(mode="protocol")
180
  return list(templates_dict.values())
181
 
182
  def add_resource_or_template_from_fn(
 
5
  import inspect
6
  import warnings
7
  from collections.abc import Callable
8
+ from typing import TYPE_CHECKING, Any
9
 
10
  from pydantic import AnyUrl
11
 
 
63
 
64
  async def get_resources(self) -> dict[str, Resource]:
65
  """Get all registered resources, keyed by URI."""
66
+ return await self._load_resources(via_server=False)
67
 
68
  async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
69
  """Get all registered templates, keyed by URI template."""
70
+ return await self._load_resource_templates(via_server=False)
71
 
72
+ async def _load_resources(self, *, via_server: bool = False) -> dict[str, Resource]:
 
 
73
  """
74
+ The single, consolidated recursive method for fetching resources. The 'via_server'
75
  parameter determines the communication path.
76
 
77
+ - via_server=False: Manager-to-manager path for complete, unfiltered inventory
78
+ - via_server=True: Server-to-server path for filtered MCP requests
79
  """
80
  all_resources: dict[str, Resource] = {}
81
 
82
  for mounted in self._mounted_sources:
83
  try:
84
+ if via_server:
85
+ # Use the server-to-server filtered path
86
  child_resources_list = await mounted.server._list_resources()
87
  child_resources = {
88
  resource.key: resource for resource in child_resources_list
89
  }
90
+ else:
91
+ # Use the manager-to-manager unfiltered path
92
  child_resources = (
93
  await mounted.server._resource_manager.get_resources()
94
  )
 
118
  return all_resources
119
 
120
  async def _load_resource_templates(
121
+ self, *, via_server: bool = False
122
  ) -> dict[str, ResourceTemplate]:
123
  """
124
+ The single, consolidated recursive method for fetching templates. The 'via_server'
125
  parameter determines the communication path.
126
 
127
+ - via_server=False: Manager-to-manager path for complete, unfiltered inventory
128
+ - via_server=True: Server-to-server path for filtered MCP requests
129
  """
130
  all_templates: dict[str, ResourceTemplate] = {}
131
 
132
  for mounted in self._mounted_sources:
133
  try:
134
+ if via_server:
135
+ # Use the server-to-server filtered path
136
  child_templates = await mounted.server._list_resource_templates()
137
+ else:
138
+ # Use the manager-to-manager unfiltered path
139
  child_templates = await mounted.server._resource_manager._list_resource_templates()
140
  child_dict = {template.key: template for template in child_templates}
141
 
 
167
  """
168
  Lists all resources, applying protocol filtering.
169
  """
170
+ resources_dict = await self._load_resources(via_server=True)
171
  return list(resources_dict.values())
172
 
173
  async def _list_resource_templates(self) -> list[ResourceTemplate]:
174
  """
175
  Lists all templates, applying protocol filtering.
176
  """
177
+ templates_dict = await self._load_resource_templates(via_server=True)
178
  return list(templates_dict.values())
179
 
180
  def add_resource_or_template_from_fn(
src/fastmcp/tools/tool_manager.py CHANGED
@@ -2,7 +2,7 @@ from __future__ import annotations
2
 
3
  import warnings
4
  from collections.abc import Callable
5
- from typing import TYPE_CHECKING, Any, Literal
6
 
7
  from mcp.types import ToolAnnotations
8
 
@@ -47,25 +47,23 @@ class ToolManager:
47
  """Adds a mounted server as a source for tools."""
48
  self._mounted_sources.append(server)
49
 
50
- async def _load_tools(
51
- self, *, mode: Literal["inventory", "protocol"]
52
- ) -> dict[str, Tool]:
53
  """
54
- The single, consolidated recursive method for fetching tools. The 'mode'
55
  parameter determines the communication path.
56
 
57
- - mode="inventory": Manager-to-manager path for complete, unfiltered inventory
58
- - mode="protocol": Server-to-server path for filtered MCP requests
59
  """
60
  all_tools: dict[str, Tool] = {}
61
 
62
  for mounted in self._mounted_sources:
63
  try:
64
- if mode == "protocol":
65
- # PATH 2: Use the server-to-server filtered path
66
  child_results = await mounted.server._list_tools()
67
- else: # mode == "inventory"
68
- # PATH 1: Use the manager-to-manager unfiltered path
69
  child_results = await mounted.server._tool_manager._list_tools()
70
 
71
  # The combination logic is the same for both paths
@@ -103,13 +101,13 @@ class ToolManager:
103
  """
104
  Gets the complete, unfiltered inventory of all tools.
105
  """
106
- return await self._load_tools(mode="inventory")
107
 
108
  async def _list_tools(self) -> list[Tool]:
109
  """
110
  Lists all tools, applying protocol filtering.
111
  """
112
- tools_dict = await self._load_tools(mode="protocol")
113
  return list(tools_dict.values())
114
 
115
  def add_tool_from_fn(
 
2
 
3
  import warnings
4
  from collections.abc import Callable
5
+ from typing import TYPE_CHECKING, Any
6
 
7
  from mcp.types import ToolAnnotations
8
 
 
47
  """Adds a mounted server as a source for tools."""
48
  self._mounted_sources.append(server)
49
 
50
+ async def _load_tools(self, *, via_server: bool = False) -> dict[str, Tool]:
 
 
51
  """
52
+ The single, consolidated recursive method for fetching tools. The 'via_server'
53
  parameter determines the communication path.
54
 
55
+ - via_server=False: Manager-to-manager path for complete, unfiltered inventory
56
+ - via_server=True: Server-to-server path for filtered MCP requests
57
  """
58
  all_tools: dict[str, Tool] = {}
59
 
60
  for mounted in self._mounted_sources:
61
  try:
62
+ if via_server:
63
+ # Use the server-to-server filtered path
64
  child_results = await mounted.server._list_tools()
65
+ else:
66
+ # Use the manager-to-manager unfiltered path
67
  child_results = await mounted.server._tool_manager._list_tools()
68
 
69
  # The combination logic is the same for both paths
 
101
  """
102
  Gets the complete, unfiltered inventory of all tools.
103
  """
104
+ return await self._load_tools(via_server=False)
105
 
106
  async def _list_tools(self) -> list[Tool]:
107
  """
108
  Lists all tools, applying protocol filtering.
109
  """
110
+ tools_dict = await self._load_tools(via_server=True)
111
  return list(tools_dict.values())
112
 
113
  def add_tool_from_fn(