ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-23 05:54:19 +02:00
Code Issues Projects Releases Packages Wiki Activity

Consolidate get_* and _list_* methods into single API (#2719)

Browse source
This commit is contained in:
Jeremiah Lowin 2025-12-24 22:26:52 -05:00 • committed by GitHub
commit 125f79ff39
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
22 changed files with 582 additions and 587 deletions
Download patch file Download diff file Expand all files Collapse all files

19
docs/development/upgrade-guide.mdx
View file

@ -72,6 +72,25 @@ tool = await mcp.get_tool(name="my_tool")
```
</CodeGroup>
### Component Listing Methods Return Lists
The `get_tools()`, `get_resources()`, `get_prompts()`, and `get_resource_templates()` methods now return lists instead of dicts:
<CodeGroup>
```python Before
tools = await server.get_tools()
if "my_tool" in tools:
tool = tools["my_tool"]
```
```python After
tools = await server.get_tools()
tool = next((t for t in tools if t.name == "my_tool"), None)
```
</CodeGroup>
The dict key was redundant since components already have `.name` or `.uri` attributes. Use list comprehensions or `next()` for lookups.
## v2.14.0
### OpenAPI Parser Promotion

8
src/fastmcp/server/providers/fastmcp_provider.py
View file

@ -481,7 +481,7 @@ class FastMCPProvider(Provider):
each tool as a FastMCPProviderTool that delegates execution to the
nested server's middleware.
"""
raw_tools = await self.server._list_tools_middleware()
raw_tools = await self.server.get_tools(apply_middleware=True)
return [FastMCPProviderTool.wrap(self.server, t) for t in raw_tools]
async def get_tool(self, name: str) -> Tool | None:
@ -500,7 +500,7 @@ class FastMCPProvider(Provider):
each resource as a FastMCPProviderResource that delegates reading to the
nested server's middleware.
"""
raw_resources = await self.server._list_resources_middleware()
raw_resources = await self.server.get_resources(apply_middleware=True)
return [FastMCPProviderResource.wrap(self.server, r) for r in raw_resources]
async def get_resource(self, uri: str) -> Resource | None:
@ -518,7 +518,7 @@ class FastMCPProvider(Provider):
Returns FastMCPProviderResourceTemplate instances that create
FastMCPProviderResources when materialized.
"""
raw_templates = await self.server._list_resource_templates_middleware()
raw_templates = await self.server.get_resource_templates(apply_middleware=True)
return [
FastMCPProviderResourceTemplate.wrap(self.server, t) for t in raw_templates
]
@ -541,7 +541,7 @@ class FastMCPProvider(Provider):
Returns FastMCPProviderPrompt instances that delegate rendering to the
wrapped server's middleware.
"""
raw_prompts = await self.server._list_prompts_middleware()
raw_prompts = await self.server.get_prompts(apply_middleware=True)
return [FastMCPProviderPrompt.wrap(self.server, p) for p in raw_prompts]
async def get_prompt(self, name: str) -> Prompt | None:

380
src/fastmcp/server/server.py
View file

@ -790,12 +790,34 @@ class FastMCP(Generic[LifespanResultT]):
"""Check if a component is enabled (not in blocklist, passes allowlist)."""
return self._visibility.is_enabled(component)
async def get_tools(self) -> dict[str, Tool]:
"""Get all enabled tools from providers, indexed by name.
async def get_tools(self, *, apply_middleware: bool = False) -> list[Tool]:
"""Get all enabled tools from providers.
Queries all providers in parallel and collects tools.
First provider wins for duplicate names. Filters by server blocklist.
First provider wins for duplicate keys. Filters by server blocklist.
Args:
apply_middleware: If True, apply the middleware chain before
returning results. Used by MCP handlers and mounted servers.
"""
if apply_middleware:
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
mw_context = MiddlewareContext(
message=mcp.types.ListToolsRequest(method="tools/list"),
source="client",
type="request",
method="tools/list",
fastmcp_context=fastmcp_ctx,
)
return list(
await self._apply_middleware(
context=mw_context,
call_next=lambda context: self.get_tools(
apply_middleware=False
),
)
)
results = await gather(
*[p.list_tools() for p in self._providers],
return_exceptions=True,
@ -805,14 +827,14 @@ class FastMCP(Generic[LifespanResultT]):
for i, result in enumerate(results):
if isinstance(result, BaseException):
provider = self._providers[i]
logger.warning(f"Failed to get tools from {provider}: {result}")
logger.exception(f"Error listing tools from provider {provider}")
if fastmcp.settings.mounted_components_raise_on_load_error:
raise result
continue
for tool in result:
if tool.name not in all_tools and self._is_component_enabled(tool):
all_tools[tool.name] = tool
return all_tools
if self._is_component_enabled(tool) and tool.key not in all_tools:
all_tools[tool.key] = tool
return list(all_tools.values())
async def get_tool(self, name: str) -> Tool:
"""Get an enabled tool by name.
@ -867,12 +889,34 @@ class FastMCP(Generic[LifespanResultT]):
return None
async def get_resources(self) -> dict[str, Resource]:
"""Get all enabled resources from providers, indexed by URI.
async def get_resources(self, *, apply_middleware: bool = False) -> list[Resource]:
"""Get all enabled resources from providers.
Queries all providers in parallel and collects resources.
First provider wins for duplicate URIs. Filters by server blocklist.
First provider wins for duplicate keys. Filters by server blocklist.
Args:
apply_middleware: If True, apply the middleware chain before
returning results. Used by MCP handlers and mounted servers.
"""
if apply_middleware:
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
mw_context = MiddlewareContext(
message={}, # List resources doesn't have parameters
source="client",
type="request",
method="resources/list",
fastmcp_context=fastmcp_ctx,
)
return list(
await self._apply_middleware(
context=mw_context,
call_next=lambda context: self.get_resources(
apply_middleware=False
),
)
)
results = await gather(
*[p.list_resources() for p in self._providers],
return_exceptions=True,
@ -882,15 +926,17 @@ class FastMCP(Generic[LifespanResultT]):
for i, result in enumerate(results):
if isinstance(result, BaseException):
provider = self._providers[i]
logger.warning(f"Failed to get resources from {provider}: {result}")
logger.exception(f"Error listing resources from provider {provider}")
if fastmcp.settings.mounted_components_raise_on_load_error:
raise result
continue
for resource in result:
uri = str(resource.uri)
if uri not in all_resources and self._is_component_enabled(resource):
all_resources[uri] = resource
return all_resources
if (
self._is_component_enabled(resource)
and resource.key not in all_resources
):
all_resources[resource.key] = resource
return list(all_resources.values())
async def get_resource(self, uri: str) -> Resource:
"""Get an enabled resource by URI.
@ -915,12 +961,36 @@ class FastMCP(Generic[LifespanResultT]):
raise NotFoundError(f"Unknown resource: {uri}")
async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
"""Get all enabled resource templates from providers, indexed by uri_template.
async def get_resource_templates(
self, *, apply_middleware: bool = False
) -> list[ResourceTemplate]:
"""Get all enabled resource templates from providers.
Queries all providers in parallel and collects templates.
First provider wins for duplicate uri_templates. Filters by server blocklist.
First provider wins for duplicate keys. Filters by server blocklist.
Args:
apply_middleware: If True, apply the middleware chain before
returning results. Used by MCP handlers and mounted servers.
"""
if apply_middleware:
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
mw_context = MiddlewareContext(
message={}, # List resource templates doesn't have parameters
source="client",
type="request",
method="resources/templates/list",
fastmcp_context=fastmcp_ctx,
)
return list(
await self._apply_middleware(
context=mw_context,
call_next=lambda context: self.get_resource_templates(
apply_middleware=False
),
)
)
results = await gather(
*[p.list_resource_templates() for p in self._providers],
return_exceptions=True,
@ -930,19 +1000,19 @@ class FastMCP(Generic[LifespanResultT]):
for i, result in enumerate(results):
if isinstance(result, BaseException):
provider = self._providers[i]
logger.warning(
f"Failed to get resource templates from {provider}: {result}"
logger.exception(
f"Error listing resource templates from provider {provider}"
)
if fastmcp.settings.mounted_components_raise_on_load_error:
raise result
continue
for template in result:
if (
template.uri_template not in all_templates
and self._is_component_enabled(template)
self._is_component_enabled(template)
and template.key not in all_templates
):
all_templates[template.uri_template] = template
return all_templates
all_templates[template.key] = template
return list(all_templates.values())
async def get_resource_template(self, uri: str) -> ResourceTemplate:
"""Get an enabled resource template that matches the given URI.
@ -969,12 +1039,34 @@ class FastMCP(Generic[LifespanResultT]):
raise NotFoundError(f"Unknown resource template: {uri}")
async def get_prompts(self) -> dict[str, Prompt]:
"""Get all enabled prompts from providers, indexed by name.
async def get_prompts(self, *, apply_middleware: bool = False) -> list[Prompt]:
"""Get all enabled prompts from providers.
Queries all providers in parallel and collects prompts.
First provider wins for duplicate names. Filters by server blocklist.
First provider wins for duplicate keys. Filters by server blocklist.
Args:
apply_middleware: If True, apply the middleware chain before
returning results. Used by MCP handlers and mounted servers.
"""
if apply_middleware:
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
mw_context = MiddlewareContext(
message=mcp.types.ListPromptsRequest(method="prompts/list"),
source="client",
type="request",
method="prompts/list",
fastmcp_context=fastmcp_ctx,
)
return list(
await self._apply_middleware(
context=mw_context,
call_next=lambda context: self.get_prompts(
apply_middleware=False
),
)
)
results = await gather(
*[p.list_prompts() for p in self._providers],
return_exceptions=True,
@ -984,16 +1076,14 @@ class FastMCP(Generic[LifespanResultT]):
for i, result in enumerate(results):
if isinstance(result, BaseException):
provider = self._providers[i]
logger.warning(f"Failed to get prompts from {provider}: {result}")
logger.exception(f"Error listing prompts from provider {provider}")
if fastmcp.settings.mounted_components_raise_on_load_error:
raise result
continue
for prompt in result:
if prompt.name not in all_prompts and self._is_component_enabled(
prompt
):
all_prompts[prompt.name] = prompt
return all_prompts
if self._is_component_enabled(prompt) and prompt.key not in all_prompts:
all_prompts[prompt.key] = prompt
return list(all_prompts.values())
async def get_prompt(self, name: str) -> Prompt:
"""Get an enabled prompt by name.
@ -1121,7 +1211,7 @@ class FastMCP(Generic[LifespanResultT]):
logger.debug(f"[{self.name}] Handler called: list_tools")
async with fastmcp.server.context.Context(fastmcp=self):
tools = await self._list_tools_middleware()
tools = await self.get_tools(apply_middleware=True)
return [
tool.to_mcp_tool(
name=tool.name,
@ -1130,55 +1220,6 @@ class FastMCP(Generic[LifespanResultT]):
for tool in tools
]
async def _list_tools_middleware(self) -> list[Tool]:
"""
List all available tools, applying MCP middleware.
"""
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
# Create the middleware context.
mw_context = MiddlewareContext(
message=mcp.types.ListToolsRequest(method="tools/list"),
source="client",
type="request",
method="tools/list",
fastmcp_context=fastmcp_ctx,
)
# Apply the middleware chain.
return list(
await self._apply_middleware(
context=mw_context, call_next=self._list_tools
)
)
async def _list_tools(
self,
context: MiddlewareContext[mcp.types.ListToolsRequest],
) -> list[Tool]:
"""
List all available tools.
Queries all providers in parallel and collects tools.
First provider wins for duplicate keys.
"""
results = await gather(
*[p.list_tools() for p in self._providers],
return_exceptions=True,
)
all_tools: dict[str, Tool] = {}
for result in results:
if isinstance(result, BaseException):
logger.exception("Error listing tools from provider", exc_info=result)
if fastmcp.settings.mounted_components_raise_on_load_error:
raise result
continue
for tool in result:
if self._is_component_enabled(tool) and tool.key not in all_tools:
all_tools[tool.key] = tool
return list(all_tools.values())
async def _list_resources_mcp(self) -> list[SDKResource]:
"""
List all available resources, in the format expected by the low-level MCP
@ -1187,7 +1228,7 @@ class FastMCP(Generic[LifespanResultT]):
logger.debug(f"[{self.name}] Handler called: list_resources")
async with fastmcp.server.context.Context(fastmcp=self):
resources = await self._list_resources_middleware()
resources = await self.get_resources(apply_middleware=True)
return [
resource.to_mcp_resource(
uri=str(resource.uri),
@ -1196,60 +1237,6 @@ class FastMCP(Generic[LifespanResultT]):
for resource in resources
]
async def _list_resources_middleware(self) -> list[Resource]:
"""
List all available resources, applying MCP middleware.
"""
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
# Create the middleware context.
mw_context = MiddlewareContext(
message={}, # List resources doesn't have parameters
source="client",
type="request",
method="resources/list",
fastmcp_context=fastmcp_ctx,
)
# Apply the middleware chain.
return list(
await self._apply_middleware(
context=mw_context, call_next=self._list_resources
)
)
async def _list_resources(
self,
context: MiddlewareContext[dict[str, Any]],
) -> list[Resource]:
"""
List all available resources.
Queries all providers in parallel and collects resources.
First provider wins for duplicate keys.
"""
results = await gather(
*[p.list_resources() for p in self._providers],
return_exceptions=True,
)
all_resources: dict[str, Resource] = {}
for result in results:
if isinstance(result, BaseException):
logger.exception(
"Error listing resources from provider", exc_info=result
)
if fastmcp.settings.mounted_components_raise_on_load_error:
raise result
continue
for resource in result:
if (
self._is_component_enabled(resource)
and resource.key not in all_resources
):
all_resources[resource.key] = resource
return list(all_resources.values())
async def _list_resource_templates_mcp(self) -> list[SDKResourceTemplate]:
"""
List all available resource templates, in the format expected by the low-level MCP
@ -1258,7 +1245,7 @@ class FastMCP(Generic[LifespanResultT]):
logger.debug(f"[{self.name}] Handler called: list_resource_templates")
async with fastmcp.server.context.Context(fastmcp=self):
templates = await self._list_resource_templates_middleware()
templates = await self.get_resource_templates(apply_middleware=True)
return [
template.to_mcp_template(
uriTemplate=template.uri_template,
@ -1267,61 +1254,6 @@ class FastMCP(Generic[LifespanResultT]):
for template in templates
]
async def _list_resource_templates_middleware(self) -> list[ResourceTemplate]:
"""
List all available resource templates, applying MCP middleware.
"""
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
# Create the middleware context.
mw_context = MiddlewareContext(
message={}, # List resource templates doesn't have parameters
source="client",
type="request",
method="resources/templates/list",
fastmcp_context=fastmcp_ctx,
)
# Apply the middleware chain.
return list(
await self._apply_middleware(
context=mw_context, call_next=self._list_resource_templates
)
)
async def _list_resource_templates(
self,
context: MiddlewareContext[dict[str, Any]],
) -> list[ResourceTemplate]:
"""
List all available resource templates.
Queries all providers in parallel and collects templates.
First provider wins for duplicate keys.
"""
results = await gather(
*[p.list_resource_templates() for p in self._providers],
return_exceptions=True,
)
all_templates: dict[str, ResourceTemplate] = {}
for result in results:
if isinstance(result, BaseException):
logger.exception(
"Error listing resource templates from provider", exc_info=result
)
if fastmcp.settings.mounted_components_raise_on_load_error:
raise result
continue
for template in result:
if (
self._is_component_enabled(template)
and template.key not in all_templates
):
all_templates[template.key] = template
return list(all_templates.values())
async def _list_prompts_mcp(self) -> list[SDKPrompt]:
"""
List all available prompts, in the format expected by the low-level MCP
@ -1330,7 +1262,7 @@ class FastMCP(Generic[LifespanResultT]):
logger.debug(f"[{self.name}] Handler called: list_prompts")
async with fastmcp.server.context.Context(fastmcp=self):
prompts = await self._list_prompts_middleware()
prompts = await self.get_prompts(apply_middleware=True)
return [
prompt.to_mcp_prompt(
name=prompt.name,
@ -1339,56 +1271,6 @@ class FastMCP(Generic[LifespanResultT]):
for prompt in prompts
]
async def _list_prompts_middleware(self) -> list[Prompt]:
"""
List all available prompts, applying MCP middleware.
"""
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
# Create the middleware context.
mw_context = MiddlewareContext(
message=mcp.types.ListPromptsRequest(method="prompts/list"),
source="client",
type="request",
method="prompts/list",
fastmcp_context=fastmcp_ctx,
)
# Apply the middleware chain.
return list(
await self._apply_middleware(
context=mw_context, call_next=self._list_prompts
)
)
async def _list_prompts(
self,
context: MiddlewareContext[mcp.types.ListPromptsRequest],
) -> list[Prompt]:
"""
List all available prompts.
Queries all providers in parallel and collects prompts.
First provider wins for duplicate keys.
"""
results = await gather(
*[p.list_prompts() for p in self._providers],
return_exceptions=True,
)
all_prompts: dict[str, Prompt] = {}
for result in results:
if isinstance(result, BaseException):
logger.exception("Error listing prompts from provider", exc_info=result)
if fastmcp.settings.mounted_components_raise_on_load_error:
raise result
continue
for prompt in result:
if self._is_component_enabled(prompt) and prompt.key not in all_prompts:
all_prompts[prompt.key] = prompt
return list(all_prompts.values())
async def _call_tool_mcp(
self, key: str, arguments: dict[str, Any]
) -> (
@ -2574,19 +2456,19 @@ class FastMCP(Generic[LifespanResultT]):
return uri
# Import tools from the server
for tool in (await server.get_tools()).values():
for tool in await server.get_tools():
if prefix:
tool = tool.model_copy(update={"name": f"{prefix}_{tool.name}"})
self.add_tool(tool)
# Import resources and templates from the server
for resource in (await server.get_resources()).values():
for resource in await server.get_resources():
if prefix:
new_uri = add_resource_prefix(str(resource.uri), prefix)
resource = resource.model_copy(update={"uri": new_uri})
self.add_resource(resource)
for template in (await server.get_resource_templates()).values():
for template in await server.get_resource_templates():
if prefix:
new_uri_template = add_resource_prefix(template.uri_template, prefix)
template = template.model_copy(
@ -2595,7 +2477,7 @@ class FastMCP(Generic[LifespanResultT]):
self.add_template(template)
# Import prompts from the server
for prompt in (await server.get_prompts()).values():
for prompt in await server.get_prompts():
if prefix:
prompt = prompt.model_copy(update={"name": f"{prefix}_{prompt.name}"})
self.add_prompt(prompt)

8
src/fastmcp/utilities/inspect.py
View file

@ -107,10 +107,10 @@ async def inspect_fastmcp_v2(mcp: FastMCP[Any]) -> FastMCPInfo:
FastMCPInfo dataclass containing the extracted information
"""
# Get all components via middleware to respect filtering and preserve metadata
tools_list = await mcp._list_tools_middleware()
prompts_list = await mcp._list_prompts_middleware()
resources_list = await mcp._list_resources_middleware()
templates_list = await mcp._list_resource_templates_middleware()
tools_list = await mcp.get_tools(apply_middleware=True)
prompts_list = await mcp.get_prompts(apply_middleware=True)
resources_list = await mcp.get_resources(apply_middleware=True)
templates_list = await mcp.get_resource_templates(apply_middleware=True)
# Extract detailed tool information
tool_infos = []

6
tests/cli/test_run.py
View file

@ -164,7 +164,7 @@ def greet(name: str) -> str:
server = await source.load_server()
assert server.name == "TestServer"
tools = await server.get_tools()
assert "greet" in tools
assert any(t.name == "greet" for t in tools)
async def test_import_server_with_main_block(self, tmp_path):
"""Test importing server with if __name__ == '__main__' block."""
@ -186,7 +186,7 @@ if __name__ == "__main__":
server = await source.load_server()
assert server.name == "MainServer"
tools = await server.get_tools()
assert "calculate" in tools
assert any(t.name == "calculate" for t in tools)
async def test_import_server_standard_names(self, tmp_path):
"""Test automatic detection of standard names (mcp, server, app)."""
@ -240,7 +240,7 @@ def custom_tool() -> str:
server = await source.load_server()
assert server.name == "CustomServer"
tools = await server.get_tools()
assert "custom_tool" in tools
assert any(t.name == "custom_tool" for t in tools)
async def test_import_server_no_standard_names_fails(self, tmp_path):
"""Test importing server when no standard names exist fails."""

6
tests/cli/test_server_args.py
View file

@ -50,7 +50,7 @@ def get_config() -> dict:
# Test the tool works and can access the parsed args
tools = await server.get_tools()
assert "get_config" in tools
assert any(t.name == "get_config" for t in tools)
async def test_server_with_no_args(self, tmp_path):
"""Test a server that uses argparse with no arguments (defaults)."""
@ -131,5 +131,5 @@ mcp = FastMCP(name)
# Verify tools are available
tools = await server.get_tools()
assert "get_status" in tools
assert "echo_message" in tools
assert any(t.name == "get_status" for t in tools)
assert any(t.name == "echo_message" for t in tools)

168
tests/contrib/test_component_manager.py
View file

@ -81,7 +81,7 @@ class TestComponentManagementRoutes:
# First disable the tool
mcp.disable(keys=["tool:test_tool"])
tools = await mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
# Enable the tool via the HTTP route
response = client.post("/tools/test_tool/enable")
@ -91,13 +91,13 @@ class TestComponentManagementRoutes:
# Verify the tool is enabled
tools = await mcp.get_tools()
assert "test_tool" in tools
assert any(t.name == "test_tool" for t in tools)
async def test_disable_tool_route(self, client, mcp):
"""Test disabling a tool via the HTTP route."""
# First ensure the tool is enabled
tools = await mcp.get_tools()
assert "test_tool" in tools
assert any(t.name == "test_tool" for t in tools)
# Disable the tool via the HTTP route
response = client.post("/tools/test_tool/disable")
@ -107,14 +107,14 @@ class TestComponentManagementRoutes:
# Verify the tool is disabled
tools = await mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
async def test_enable_resource_route(self, client, mcp):
"""Test enabling a resource via the HTTP route."""
# First disable the resource
mcp.disable(keys=["resource:data://test_resource"])
resources = await mcp.get_resources()
assert "data://test_resource" not in resources
assert not any(str(r.uri) == "data://test_resource" for r in resources)
# Enable the resource via the HTTP route
response = client.post("/resources/data://test_resource/enable")
@ -124,13 +124,13 @@ class TestComponentManagementRoutes:
# Verify the resource is enabled
resources = await mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
async def test_disable_resource_route(self, client, mcp):
"""Test disabling a resource via the HTTP route."""
# First ensure the resource is enabled
resources = await mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
# Disable the resource via the HTTP route
response = client.post("/resources/data://test_resource/disable")
@ -140,41 +140,41 @@ class TestComponentManagementRoutes:
# Verify the resource is disabled
resources = await mcp.get_resources()
assert "data://test_resource" not in resources
assert not any(str(r.uri) == "data://test_resource" for r in resources)
async def test_enable_template_route(self, client, mcp):
"""Test enabling a resource on a mounted server via the parent server's HTTP route."""
key = "data://test_resource/{id}"
mcp.disable(keys=["template:data://test_resource/{id}"])
templates = await mcp.get_resource_templates()
assert key not in templates
assert not any(t.uri_template == key for t in templates)
response = client.post("/resources/data://test_resource/{id}/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Enabled resource: data://test_resource/{id}"
}
templates = await mcp.get_resource_templates()
assert key in templates
assert any(t.uri_template == key for t in templates)
async def test_disable_template_route(self, client, mcp):
"""Test disabling a resource on a mounted server via the parent server's HTTP route."""
key = "data://test_resource/{id}"
templates = await mcp.get_resource_templates()
assert key in templates
assert any(t.uri_template == key for t in templates)
response = client.post("/resources/data://test_resource/{id}/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Disabled resource: data://test_resource/{id}"
}
templates = await mcp.get_resource_templates()
assert key not in templates
assert not any(t.uri_template == key for t in templates)
async def test_enable_prompt_route(self, client, mcp):
"""Test enabling a prompt via the HTTP route."""
# First disable the prompt
mcp.disable(keys=["prompt:test_prompt"])
prompts = await mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
# Enable the prompt via the HTTP route
response = client.post("/prompts/test_prompt/enable")
@ -184,13 +184,13 @@ class TestComponentManagementRoutes:
# Verify the prompt is enabled
prompts = await mcp.get_prompts()
assert "test_prompt" in prompts
assert any(p.name == "test_prompt" for p in prompts)
async def test_disable_prompt_route(self, client, mcp):
"""Test disabling a prompt via the HTTP route."""
# First ensure the prompt is enabled
prompts = await mcp.get_prompts()
assert "test_prompt" in prompts
assert any(p.name == "test_prompt" for p in prompts)
# Disable the prompt via the HTTP route
response = client.post("/prompts/test_prompt/disable")
@ -200,107 +200,107 @@ class TestComponentManagementRoutes:
# Verify the prompt is disabled
prompts = await mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
async def test_enable_tool_route_on_mounted_server(self, client, mounted_mcp):
"""Test enabling a tool on a mounted server via the parent server's HTTP route."""
# Disable the tool on the sub-server
mounted_mcp.disable(keys=["tool:mounted_tool"])
tools = await mounted_mcp.get_tools()
assert "mounted_tool" not in tools
assert not any(t.name == "mounted_tool" for t in tools)
# Enable via parent
response = client.post("/tools/sub_mounted_tool/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled tool: sub_mounted_tool"}
# Confirm enabled on sub-server
tools = await mounted_mcp.get_tools()
assert "mounted_tool" in tools
assert any(t.name == "mounted_tool" for t in tools)
async def test_disable_tool_route_on_mounted_server(self, client, mounted_mcp):
"""Test disabling a tool on a mounted server via the parent server's HTTP route."""
# Ensure the tool is enabled on the sub-server
tools = await mounted_mcp.get_tools()
assert "mounted_tool" in tools
assert any(t.name == "mounted_tool" for t in tools)
# Disable via parent
response = client.post("/tools/sub_mounted_tool/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Disabled tool: sub_mounted_tool"}
# Confirm disabled on sub-server
tools = await mounted_mcp.get_tools()
assert "mounted_tool" not in tools
assert not any(t.name == "mounted_tool" for t in tools)
async def test_enable_resource_route_on_mounted_server(self, client, mounted_mcp):
"""Test enabling a resource on a mounted server via the parent server's HTTP route."""
mounted_mcp.disable(keys=["resource:data://mounted_resource"])
resources = await mounted_mcp.get_resources()
assert "data://mounted_resource" not in resources
assert not any(str(r.uri) == "data://mounted_resource" for r in resources)
response = client.post("/resources/data://sub/mounted_resource/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Enabled resource: data://sub/mounted_resource"
}
resources = await mounted_mcp.get_resources()
assert "data://mounted_resource" in resources
assert any(str(r.uri) == "data://mounted_resource" for r in resources)
async def test_disable_resource_route_on_mounted_server(self, client, mounted_mcp):
"""Test disabling a resource on a mounted server via the parent server's HTTP route."""
resources = await mounted_mcp.get_resources()
assert "data://mounted_resource" in resources
assert any(str(r.uri) == "data://mounted_resource" for r in resources)
response = client.post("/resources/data://sub/mounted_resource/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Disabled resource: data://sub/mounted_resource"
}
resources = await mounted_mcp.get_resources()
assert "data://mounted_resource" not in resources
assert not any(str(r.uri) == "data://mounted_resource" for r in resources)
async def test_enable_template_route_on_mounted_server(self, client, mounted_mcp):
"""Test enabling a resource on a mounted server via the parent server's HTTP route."""
key = "data://mounted_resource/{id}"
mounted_mcp.disable(keys=["template:data://mounted_resource/{id}"])
templates = await mounted_mcp.get_resource_templates()
assert key not in templates
assert not any(t.uri_template == key for t in templates)
response = client.post("/resources/data://sub/mounted_resource/{id}/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Enabled resource: data://sub/mounted_resource/{id}"
}
templates = await mounted_mcp.get_resource_templates()
assert key in templates
assert any(t.uri_template == key for t in templates)
async def test_disable_template_route_on_mounted_server(self, client, mounted_mcp):
"""Test disabling a resource on a mounted server via the parent server's HTTP route."""
key = "data://mounted_resource/{id}"
templates = await mounted_mcp.get_resource_templates()
assert key in templates
assert any(t.uri_template == key for t in templates)
response = client.post("/resources/data://sub/mounted_resource/{id}/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Disabled resource: data://sub/mounted_resource/{id}"
}
templates = await mounted_mcp.get_resource_templates()
assert key not in templates
assert not any(t.uri_template == key for t in templates)
async def test_enable_prompt_route_on_mounted_server(self, client, mounted_mcp):
"""Test enabling a prompt on a mounted server via the parent server's HTTP route."""
mounted_mcp.disable(keys=["prompt:mounted_prompt"])
prompts = await mounted_mcp.get_prompts()
assert "mounted_prompt" not in prompts
assert not any(p.name == "mounted_prompt" for p in prompts)
response = client.post("/prompts/sub_mounted_prompt/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled prompt: sub_mounted_prompt"}
prompts = await mounted_mcp.get_prompts()
assert "mounted_prompt" in prompts
assert any(p.name == "mounted_prompt" for p in prompts)
async def test_disable_prompt_route_on_mounted_server(self, client, mounted_mcp):
"""Test disabling a prompt on a mounted server via the parent server's HTTP route."""
prompts = await mounted_mcp.get_prompts()
assert "mounted_prompt" in prompts
assert any(p.name == "mounted_prompt" for p in prompts)
response = client.post("/prompts/sub_mounted_prompt/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Disabled prompt: sub_mounted_prompt"}
prompts = await mounted_mcp.get_prompts()
assert "mounted_prompt" not in prompts
assert not any(p.name == "mounted_prompt" for p in prompts)
def test_enable_nonexistent_tool(self, client):
"""Test enabling a non-existent tool returns 404."""
@ -391,18 +391,18 @@ class TestAuthComponentManagementRoutes:
"""Test that unauthenticated requests to enable a tool are rejected."""
self.mcp.disable(keys=["tool:test_tool"])
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
response = self.client.post("/tools/test_tool/enable")
assert response.status_code == 401
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
async def test_authorized_enable_tool(self):
"""Test that authenticated requests to enable a tool are allowed."""
self.mcp.disable(keys=["tool:test_tool"])
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
response = self.client.post(
"/tools/test_tool/enable", headers={"Authorization": "Bearer " + self.token}
@ -410,22 +410,22 @@ class TestAuthComponentManagementRoutes:
assert response.status_code == 200
assert response.json() == {"message": "Enabled tool: test_tool"}
tools = await self.mcp.get_tools()
assert "test_tool" in tools
assert any(t.name == "test_tool" for t in tools)
async def test_unauthorized_disable_tool(self):
"""Test that unauthenticated requests to disable a tool are rejected."""
tools = await self.mcp.get_tools()
assert "test_tool" in tools
assert any(t.name == "test_tool" for t in tools)
response = self.client.post("/tools/test_tool/disable")
assert response.status_code == 401
tools = await self.mcp.get_tools()
assert "test_tool" in tools
assert any(t.name == "test_tool" for t in tools)
async def test_authorized_disable_tool(self):
"""Test that authenticated requests to disable a tool are allowed."""
tools = await self.mcp.get_tools()
assert "test_tool" in tools
assert any(t.name == "test_tool" for t in tools)
response = self.client.post(
"/tools/test_tool/disable",
@ -434,13 +434,13 @@ class TestAuthComponentManagementRoutes:
assert response.status_code == 200
assert response.json() == {"message": "Disabled tool: test_tool"}
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
async def test_forbidden_enable_tool(self):
"""Test that requests with insufficient scopes are rejected."""
self.mcp.disable(keys=["tool:test_tool"])
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
response = self.client.post(
"/tools/test_tool/enable",
@ -448,13 +448,13 @@ class TestAuthComponentManagementRoutes:
)
assert response.status_code == 403
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
async def test_authorized_enable_resource(self):
"""Test that authenticated requests to enable a resource are allowed."""
self.mcp.disable(keys=["resource:data://test_resource"])
resources = await self.mcp.get_resources()
assert "data://test_resource" not in resources
assert not any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post(
"/resources/data://test_resource/enable",
@ -463,23 +463,23 @@ class TestAuthComponentManagementRoutes:
assert response.status_code == 200
assert response.json() == {"message": "Enabled resource: data://test_resource"}
resources = await self.mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
async def test_unauthorized_disable_resource(self):
"""Test that unauthenticated requests to disable a resource are rejected."""
resources = await self.mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post("/resources/data://test_resource/disable")
assert response.status_code == 401
resources = await self.mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
async def test_forbidden_enable_resource(self):
"""Test that requests with insufficient scopes are rejected."""
self.mcp.disable(keys=["resource:data://test_resource"])
resources = await self.mcp.get_resources()
assert "data://test_resource" not in resources
assert not any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post(
"/resources/data://test_resource/disable",
@ -487,12 +487,12 @@ class TestAuthComponentManagementRoutes:
)
assert response.status_code == 403
resources = await self.mcp.get_resources()
assert "data://test_resource" not in resources
assert not any(str(r.uri) == "data://test_resource" for r in resources)
async def test_authorized_disable_resource(self):
"""Test that authenticated requests to disable a resource are allowed."""
resources = await self.mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post(
"/resources/data://test_resource/disable",
@ -501,24 +501,24 @@ class TestAuthComponentManagementRoutes:
assert response.status_code == 200
assert response.json() == {"message": "Disabled resource: data://test_resource"}
resources = await self.mcp.get_resources()
assert "data://test_resource" not in resources
assert not any(str(r.uri) == "data://test_resource" for r in resources)
async def test_unauthorized_enable_prompt(self):
"""Test that unauthenticated requests to enable a prompt are rejected."""
self.mcp.disable(keys=["prompt:test_prompt"])
prompts = await self.mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
response = self.client.post("/prompts/test_prompt/enable")
assert response.status_code == 401
prompts = await self.mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
async def test_authorized_enable_prompt(self):
"""Test that authenticated requests to enable a prompt are allowed."""
self.mcp.disable(keys=["prompt:test_prompt"])
prompts = await self.mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
response = self.client.post(
"/prompts/test_prompt/enable",
@ -527,22 +527,22 @@ class TestAuthComponentManagementRoutes:
assert response.status_code == 200
assert response.json() == {"message": "Enabled prompt: test_prompt"}
prompts = await self.mcp.get_prompts()
assert "test_prompt" in prompts
assert any(p.name == "test_prompt" for p in prompts)
async def test_unauthorized_disable_prompt(self):
"""Test that unauthenticated requests to disable a prompt are rejected."""
prompts = await self.mcp.get_prompts()
assert "test_prompt" in prompts
assert any(p.name == "test_prompt" for p in prompts)
response = self.client.post("/prompts/test_prompt/disable")
assert response.status_code == 401
prompts = await self.mcp.get_prompts()
assert "test_prompt" in prompts
assert any(p.name == "test_prompt" for p in prompts)
async def test_forbidden_disable_prompt(self):
"""Test that requests with insufficient scopes are rejected."""
prompts = await self.mcp.get_prompts()
assert "test_prompt" in prompts
assert any(p.name == "test_prompt" for p in prompts)
response = self.client.post(
"/prompts/test_prompt/disable",
@ -550,12 +550,12 @@ class TestAuthComponentManagementRoutes:
)
assert response.status_code == 403
prompts = await self.mcp.get_prompts()
assert "test_prompt" in prompts
assert any(p.name == "test_prompt" for p in prompts)
async def test_authorized_disable_prompt(self):
"""Test that authenticated requests to disable a prompt are allowed."""
prompts = await self.mcp.get_prompts()
assert "test_prompt" in prompts
assert any(p.name == "test_prompt" for p in prompts)
response = self.client.post(
"/prompts/test_prompt/disable",
@ -564,7 +564,7 @@ class TestAuthComponentManagementRoutes:
assert response.status_code == 200
assert response.json() == {"message": "Disabled prompt: test_prompt"}
prompts = await self.mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
class TestComponentManagerWithPath:
@ -596,33 +596,33 @@ class TestComponentManagerWithPath:
async def test_enable_tool_route_with_path(self, client_with_path, mcp_with_path):
mcp_with_path.disable(keys=["tool:test_tool"])
tools = await mcp_with_path.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
response = client_with_path.post("/test/tools/test_tool/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled tool: test_tool"}
tools = await mcp_with_path.get_tools()
assert "test_tool" in tools
assert any(t.name == "test_tool" for t in tools)
async def test_disable_resource_route_with_path(
self, client_with_path, mcp_with_path
):
resources = await mcp_with_path.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
response = client_with_path.post("/test/resources/data://test_resource/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Disabled resource: data://test_resource"}
resources = await mcp_with_path.get_resources()
assert "data://test_resource" not in resources
assert not any(str(r.uri) == "data://test_resource" for r in resources)
async def test_enable_prompt_route_with_path(self, client_with_path, mcp_with_path):
mcp_with_path.disable(keys=["prompt:test_prompt"])
prompts = await mcp_with_path.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
response = client_with_path.post("/test/prompts/test_prompt/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled prompt: test_prompt"}
prompts = await mcp_with_path.get_prompts()
assert "test_prompt" in prompts
assert any(p.name == "test_prompt" for p in prompts)
class TestComponentManagerWithPathAuth:
@ -670,28 +670,28 @@ class TestComponentManagerWithPathAuth:
async def test_unauthorized_enable_tool(self):
self.mcp.disable(keys=["tool:test_tool"])
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
response = self.client.post("/test/tools/test_tool/enable")
assert response.status_code == 401
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
async def test_forbidden_enable_tool(self):
self.mcp.disable(keys=["tool:test_tool"])
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
response = self.client.post(
"/test/tools/test_tool/enable",
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
async def test_authorized_enable_tool(self):
self.mcp.disable(keys=["tool:test_tool"])
tools = await self.mcp.get_tools()
assert "test_tool" not in tools
assert not any(t.name == "test_tool" for t in tools)
response = self.client.post(
"/test/tools/test_tool/enable",
headers={"Authorization": "Bearer " + self.token},
@ -699,30 +699,30 @@ class TestComponentManagerWithPathAuth:
assert response.status_code == 200
assert response.json() == {"message": "Enabled tool: test_tool"}
tools = await self.mcp.get_tools()
assert "test_tool" in tools
assert any(t.name == "test_tool" for t in tools)
async def test_unauthorized_disable_resource(self):
resources = await self.mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post("/test/resources/data://test_resource/disable")
assert response.status_code == 401
resources = await self.mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
async def test_forbidden_disable_resource(self):
resources = await self.mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post(
"/test/resources/data://test_resource/disable",
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
resources = await self.mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
async def test_authorized_disable_resource(self):
resources = await self.mcp.get_resources()
assert "data://test_resource" in resources
assert any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post(
"/test/resources/data://test_resource/disable",
headers={"Authorization": "Bearer " + self.token},
@ -730,33 +730,33 @@ class TestComponentManagerWithPathAuth:
assert response.status_code == 200
assert response.json() == {"message": "Disabled resource: data://test_resource"}
resources = await self.mcp.get_resources()
assert "data://test_resource" not in resources
assert not any(str(r.uri) == "data://test_resource" for r in resources)
async def test_unauthorized_enable_prompt(self):
self.mcp.disable(keys=["prompt:test_prompt"])
prompts = await self.mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
response = self.client.post("/test/prompts/test_prompt/enable")
assert response.status_code == 401
prompts = await self.mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
async def test_forbidden_enable_prompt(self):
self.mcp.disable(keys=["prompt:test_prompt"])
prompts = await self.mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
response = self.client.post(
"/test/prompts/test_prompt/enable",
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
prompts = await self.mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
async def test_authorized_enable_prompt(self):
self.mcp.disable(keys=["prompt:test_prompt"])
prompts = await self.mcp.get_prompts()
assert "test_prompt" not in prompts
assert not any(p.name == "test_prompt" for p in prompts)
response = self.client.post(
"/test/prompts/test_prompt/enable",
headers={"Authorization": "Bearer " + self.token},
@ -764,4 +764,4 @@ class TestComponentManagerWithPathAuth:
assert response.status_code == 200
assert response.json() == {"message": "Enabled prompt: test_prompt"}
prompts = await self.mcp.get_prompts()
assert "test_prompt" in prompts
assert any(p.name == "test_prompt" for p in prompts)

61
tests/contrib/test_mcp_mixin.py
View file

@ -68,8 +68,8 @@ class TestMCPMixin:
instance.register_tools(mcp, prefix=prefix, separator=separator)
registered_tools = await mcp.get_tools()
assert expected_key in registered_tools
assert unexpected_key not in registered_tools
assert any(t.name == expected_key for t in registered_tools)
assert not any(t.name == unexpected_key for t in registered_tools)
@pytest.mark.parametrize(
"prefix, separator, expected_uri_key, expected_name, unexpected_uri_key",
@ -113,9 +113,12 @@ class TestMCPMixin:
instance.register_resources(mcp, prefix=prefix, separator=separator)
registered_resources = await mcp.get_resources()
assert expected_uri_key in registered_resources
assert registered_resources[expected_uri_key].name == expected_name
assert unexpected_uri_key not in registered_resources
assert any(str(r.uri) == expected_uri_key for r in registered_resources)
resource = next(
r for r in registered_resources if str(r.uri) == expected_uri_key
)
assert resource.name == expected_name
assert not any(str(r.uri) == unexpected_uri_key for r in registered_resources)
@pytest.mark.parametrize(
"prefix, separator, expected_name, unexpected_name",
@ -156,8 +159,8 @@ class TestMCPMixin:
instance.register_prompts(mcp, prefix=prefix, separator=separator)
prompts = await mcp.get_prompts()
assert expected_name in prompts
assert unexpected_name not in prompts
assert any(p.name == expected_name for p in prompts)
assert not any(p.name == unexpected_name for p in prompts)
async def test_register_all_no_prefix(self):
"""Test register_all method registers all types without a prefix."""
@ -183,9 +186,9 @@ class TestMCPMixin:
resources = await mcp.get_resources()
prompts = await mcp.get_prompts()
assert "tool_all" in tools
assert "res://all" in resources
assert "prompt_all" in prompts
assert any(t.name == "tool_all" for t in tools)
assert any(str(r.uri) == "res://all" for r in resources)
assert any(p.name == "prompt_all" for p in prompts)
async def test_register_all_with_prefix_default_separators(self):
"""Test register_all method registers all types with a prefix and default separators."""
@ -211,9 +214,14 @@ class TestMCPMixin:
resources = await mcp.get_resources()
prompts = await mcp.get_prompts()
assert f"all{_DEFAULT_SEPARATOR_TOOL}tool_all_p" in tools
assert f"all{_DEFAULT_SEPARATOR_RESOURCE}res://all_p" in resources
assert f"all{_DEFAULT_SEPARATOR_PROMPT}prompt_all_p" in prompts
assert any(t.name == f"all{_DEFAULT_SEPARATOR_TOOL}tool_all_p" for t in tools)
assert any(
str(r.uri) == f"all{_DEFAULT_SEPARATOR_RESOURCE}res://all_p"
for r in resources
)
assert any(
p.name == f"all{_DEFAULT_SEPARATOR_PROMPT}prompt_all_p" for p in prompts
)
async def test_register_all_with_prefix_custom_separators(self):
"""Test register_all method registers all types with a prefix and custom separators."""
@ -245,14 +253,21 @@ class TestMCPMixin:
resources = await mcp.get_resources()
prompts = await mcp.get_prompts()
assert "cust-tool_cust" in tools
assert "cust::res://cust" in resources
assert "cust.prompt_cust" in prompts
assert any(t.name == "cust-tool_cust" for t in tools)
assert any(str(r.uri) == "cust::res://cust" for r in resources)
assert any(p.name == "cust.prompt_cust" for p in prompts)
# Check default separators weren't used
assert f"cust{_DEFAULT_SEPARATOR_TOOL}tool_cust" not in tools
assert f"cust{_DEFAULT_SEPARATOR_RESOURCE}res://cust" not in resources
assert f"cust{_DEFAULT_SEPARATOR_PROMPT}prompt_cust" not in prompts
assert not any(
t.name == f"cust{_DEFAULT_SEPARATOR_TOOL}tool_cust" for t in tools
)
assert not any(
str(r.uri) == f"cust{_DEFAULT_SEPARATOR_RESOURCE}res://cust"
for r in resources
)
assert not any(
p.name == f"cust{_DEFAULT_SEPARATOR_PROMPT}prompt_cust" for p in prompts
)
async def test_tool_with_title_and_meta(self):
"""Test that title (via annotations) and meta arguments are properly passed through."""
@ -272,7 +287,7 @@ class TestMCPMixin:
instance.register_tools(mcp)
registered_tools = await mcp.get_tools()
tool = registered_tools["sample_tool"]
tool = next(t for t in registered_tools if t.name == "sample_tool")
assert tool.annotations is not None
assert tool.annotations.title == "My Tool Title"
@ -295,7 +310,9 @@ class TestMCPMixin:
instance.register_resources(mcp)
registered_resources = await mcp.get_resources()
resource = registered_resources["test://resource"]
resource = next(
r for r in registered_resources if str(r.uri) == "test://resource"
)
assert resource.meta == {"category": "data", "internal": True}
assert resource.title == "My Resource Title"
@ -316,7 +333,7 @@ class TestMCPMixin:
instance.register_prompts(mcp)
prompts = await mcp.get_prompts()
prompt = prompts["sample_prompt"]
prompt = next(p for p in prompts if p.name == "sample_prompt")
assert prompt.title == "My Prompt Title"
assert prompt.meta == {"priority": "high", "category": "analysis"}

6
tests/deprecated/test_exclude_args.py
View file

@ -19,8 +19,7 @@ async def test_tool_exclude_args():
pass
return message
tools_dict = await mcp.get_tools()
tools = list(tools_dict.values())
tools = await mcp.get_tools()
assert len(tools) == 1
assert "state" not in tools[0].parameters["properties"]
@ -61,8 +60,7 @@ async def test_add_tool_method_exclude_args():
mcp.add_tool(tool)
# Check tool via public API
tools_dict = await mcp.get_tools()
tools = list(tools_dict.values())
tools = await mcp.get_tools()
assert len(tools) == 1
assert "state" not in tools[0].parameters["properties"]

52
tests/deprecated/test_import_server.py
View file

@ -26,8 +26,8 @@ async def test_import_basic_functionality():
# Verify the tool was imported with the prefix
main_tools = await main_app.get_tools()
sub_tools = await sub_app.get_tools()
assert "sub_sub_tool" in main_tools
assert "sub_tool" in sub_tools
assert any(t.name == "sub_sub_tool" for t in main_tools)
assert any(t.name == "sub_tool" for t in sub_tools)
# Verify the original tool still exists in the sub-app
tool = await main_app.get_tool("sub_sub_tool")
@ -60,8 +60,8 @@ async def test_import_multiple_apps():
# Verify tools were imported with the correct prefixes
tools = await main_app.get_tools()
assert "weather_get_forecast" in tools
assert "news_get_headlines" in tools
assert any(t.name == "weather_get_forecast" for t in tools)
assert any(t.name == "news_get_headlines" for t in tools)
async def test_import_combines_tools():
@ -83,17 +83,17 @@ async def test_import_combines_tools():
# Import first app
await main_app.import_server(first_app, "api")
tools = await main_app.get_tools()
assert "api_first_tool" in tools
assert any(t.name == "api_first_tool" for t in tools)
# Import second app to same prefix
await main_app.import_server(second_app, "api")
# Verify second tool is there
tools = await main_app.get_tools()
assert "api_second_tool" in tools
assert any(t.name == "api_second_tool" for t in tools)
# Tools from both imports are combined
assert "api_first_tool" in tools
assert any(t.name == "api_first_tool" for t in tools)
async def test_import_with_resources():
@ -112,7 +112,7 @@ async def test_import_with_resources():
# Verify the resource was imported with the prefix
resources = await main_app.get_resources()
assert "data://data/users" in resources
assert any(str(r.uri) == "data://data/users" for r in resources)
async def test_import_with_resource_templates():
@ -131,7 +131,7 @@ async def test_import_with_resource_templates():
# Verify the template was imported with the prefix
templates = await main_app.get_resource_templates()
assert "users://api/{user_id}/profile" in templates
assert any(t.uri_template == "users://api/{user_id}/profile" for t in templates)
async def test_import_with_prompts():
@ -150,7 +150,7 @@ async def test_import_with_prompts():
# Verify the prompt was imported with the prefix
prompts = await main_app.get_prompts()
assert "assistant_greeting" in prompts
assert any(p.name == "assistant_greeting" for p in prompts)
async def test_import_multiple_resource_templates():
@ -175,8 +175,8 @@ async def test_import_multiple_resource_templates():
# Verify templates were imported with correct prefixes
templates = await main_app.get_resource_templates()
assert "weather://data/{city}" in templates
assert "news://content/{category}" in templates
assert any(t.uri_template == "weather://data/{city}" for t in templates)
assert any(t.uri_template == "news://content/{category}" for t in templates)
async def test_import_multiple_prompts():
@ -201,8 +201,8 @@ async def test_import_multiple_prompts():
# Verify prompts were imported with correct prefixes
prompts = await main_app.get_prompts()
assert "python_review_python" in prompts
assert "sql_explain_sql" in prompts
assert any(p.name == "python_review_python" for p in prompts)
assert any(p.name == "sql_explain_sql" for p in prompts)
async def test_tool_custom_name_preserved_when_imported():
@ -436,10 +436,10 @@ async def test_import_with_no_prefix():
resources = await main_app.get_resources()
templates = await main_app.get_resource_templates()
prompts = await main_app.get_prompts()
assert "sub_tool" in tools
assert "data://config" in resources
assert "users://{user_id}/info" in templates
assert "sub_prompt" in prompts
assert any(t.name == "sub_tool" for t in tools)
assert any(str(r.uri) == "data://config" for r in resources)
assert any(t.uri_template == "users://{user_id}/info" for t in templates)
assert any(p.name == "sub_prompt" for p in prompts)
# Test actual functionality through Client
async with Client(main_app) as client:
@ -630,7 +630,9 @@ async def test_import_server_resource_uri_prefixing():
# Get resources and verify URI prefixing (name should NOT be prefixed)
resources = await main_server.get_resources()
resource = resources["resource://imported/test_resource"]
resource = next(
r for r in resources if str(r.uri) == "resource://imported/test_resource"
)
assert resource.name == "test_resource"
@ -649,7 +651,9 @@ async def test_import_server_resource_template_uri_prefixing():
# Get resource templates and verify URI prefixing (name should NOT be prefixed)
templates = await main_server.get_resource_templates()
template = templates["resource://imported/data/{item_id}"]
template = next(
t for t in templates if t.uri_template == "resource://imported/data/{item_id}"
)
assert template.name == "data_template"
@ -678,9 +682,11 @@ async def test_import_server_with_new_prefix_format():
resources = await target_server.get_resources()
templates = await target_server.get_resource_templates()
assert "resource://imported/test-resource" in resources
assert "resource://imported//absolute/path" in resources
assert "resource://imported/{param}/template" in templates
assert any(str(r.uri) == "resource://imported/test-resource" for r in resources)
assert any(str(r.uri) == "resource://imported//absolute/path" for r in resources)
assert any(
t.uri_template == "resource://imported/{param}/template" for t in templates
)
# Verify we can access the resources
async with Client(target_server) as client:

2
tests/server/providers/openapi/test_performance_comparison.py
View file

@ -247,7 +247,7 @@ class TestPerformance:
"delete_user",
"search_users",
}
assert set(tools.keys()) == expected_operations
assert {t.name for t in tools} == expected_operations
def test_memory_efficiency(self, comprehensive_spec):
"""Test that implementation doesn't significantly increase memory usage."""

8
tests/server/providers/test_local_provider.py
View file

@ -589,7 +589,7 @@ class TestLocalProviderStandaloneUsage:
server = FastMCP("Test", providers=[provider])
tools = await server.get_tools()
assert "provider_tool" in tools
assert any(t.name == "provider_tool" for t in tools)
async def test_server_decorator_and_provider_tools_coexist(self):
"""Test that server decorators and provider tools coexist."""
@ -606,8 +606,8 @@ class TestLocalProviderStandaloneUsage:
return "from server"
tools = await server.get_tools()
assert "provider_tool" in tools
assert "server_tool" in tools
assert any(t.name == "provider_tool" for t in tools)
assert any(t.name == "server_tool" for t in tools)
async def test_local_provider_first_wins_duplicates(self):
"""Test that LocalProvider tools take precedence over added providers."""
@ -625,7 +625,7 @@ class TestLocalProviderStandaloneUsage:
# Server's LocalProvider is first, so its tool wins
tools = await server.get_tools()
assert "duplicate_tool" in tools
assert any(t.name == "duplicate_tool" for t in tools)
async with Client(server) as client:
result = await client.call_tool("duplicate_tool", {})

43
tests/server/providers/test_local_provider_prompts.py
View file

@ -53,9 +53,9 @@ class TestPromptDecorator:
def fn() -> str:
return "Hello, world!"
prompts_dict = await mcp.get_prompts()
assert len(prompts_dict) == 1
prompt = prompts_dict["fn"]
prompts = await mcp.get_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "fn")
assert prompt.name == "fn"
content = await prompt.render()
if not isinstance(content, PromptResult):
@ -71,7 +71,7 @@ class TestPromptDecorator:
return "Hello, world!"
prompts = await mcp.get_prompts()
assert "fn" in prompts
assert any(p.name == "fn" for p in prompts)
async with Client(mcp) as client:
result = await client.get_prompt("fn")
@ -86,9 +86,9 @@ class TestPromptDecorator:
def fn() -> str:
return "Hello, world!"
prompts_dict = await mcp.get_prompts()
assert len(prompts_dict) == 1
prompt = prompts_dict["custom_name"]
prompts_list = await mcp.get_prompts()
assert len(prompts_list) == 1
prompt = next(p for p in prompts_list if p.name == "custom_name")
assert prompt.name == "custom_name"
content = await prompt.render()
if not isinstance(content, PromptResult):
@ -103,9 +103,9 @@ class TestPromptDecorator:
def fn() -> str:
return "Hello, world!"
prompts_dict = await mcp.get_prompts()
assert len(prompts_dict) == 1
prompt = prompts_dict["fn"]
prompts_list = await mcp.get_prompts()
assert len(prompts_list) == 1
prompt = next(p for p in prompts_list if p.name == "fn")
assert prompt.description == "A custom description"
content = await prompt.render()
if not isinstance(content, PromptResult):
@ -120,9 +120,9 @@ class TestPromptDecorator:
def test_prompt(name: str, greeting: str = "Hello") -> str:
return f"{greeting}, {name}!"
prompts_dict = await mcp.get_prompts()
assert len(prompts_dict) == 1
prompt = prompts_dict["test_prompt"]
prompts = await mcp.get_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "test_prompt")
assert prompt.arguments is not None
assert len(prompt.arguments) == 2
assert prompt.arguments[0].name == "name"
@ -233,9 +233,9 @@ class TestPromptDecorator:
def sample_prompt() -> str:
return "Hello, world!"
prompts_dict = await mcp.get_prompts()
assert len(prompts_dict) == 1
prompt = prompts_dict["sample_prompt"]
prompts = await mcp.get_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "sample_prompt")
assert prompt.tags == {"example", "test-tag"}
async def test_prompt_decorator_with_string_name(self):
@ -248,8 +248,8 @@ class TestPromptDecorator:
return "Hello from string named prompt!"
prompts = await mcp.get_prompts()
assert "string_named_prompt" in prompts
assert "my_function" not in prompts
assert any(p.name == "string_named_prompt" for p in prompts)
assert not any(p.name == "my_function" for p in prompts)
async with Client(mcp) as client:
result = await client.get_prompt("string_named_prompt")
@ -270,7 +270,8 @@ class TestPromptDecorator:
assert isinstance(result_fn, FunctionPrompt)
prompts = await mcp.get_prompts()
assert prompts["direct_call_prompt"] is result_fn
prompt = next(p for p in prompts if p.name == "direct_call_prompt")
assert prompt is result_fn
async with Client(mcp) as client:
result = await client.get_prompt("direct_call_prompt")
@ -318,7 +319,7 @@ class TestPromptDecorator:
def test_prompt(message: str) -> str:
return f"Response: {message}"
prompts_dict = await mcp.get_prompts()
prompt = prompts_dict["test_prompt"]
prompts = await mcp.get_prompts()
prompt = next(p for p in prompts if p.name == "test_prompt")
assert prompt.meta == meta_data

46
tests/server/providers/test_local_provider_resources.py
View file

@ -143,8 +143,7 @@ class TestResourceTemplates:
def add(x: int, y: int = 10) -> int:
return x + y
templates_dict = await mcp.get_resource_templates()
templates = list(templates_dict.values())
templates = await mcp.get_resource_templates()
assert len(templates) == 1
assert templates[0].uri_template == "math://add/{x}"
@ -165,8 +164,7 @@ class TestResourceTemplates:
def get_data(name: str) -> str:
return f"Data for {name}"
templates_dict = await mcp.get_resource_templates()
templates = list(templates_dict.values())
templates = await mcp.get_resource_templates()
assert len(templates) == 1
assert templates[0].uri_template == "resource://{name}/data"
@ -182,8 +180,8 @@ class TestResourceTemplates:
def template_resource(param: str) -> str:
return f"Template resource: {param}"
templates_dict = await mcp.get_resource_templates()
template = templates_dict["resource://{param}"]
templates = await mcp.get_resource_templates()
template = next(t for t in templates if t.uri_template == "resource://{param}")
assert template.tags == {"template", "test-tag"}
async def test_template_decorator_wildcard_param(self):
@ -358,8 +356,7 @@ class TestResourceDecorator:
def get_data() -> str:
return "Hello, world!"
resources_dict = await mcp.get_resources()
resources = list(resources_dict.values())
resources = await mcp.get_resources()
assert len(resources) == 1
assert resources[0].name == "custom-data"
@ -375,8 +372,7 @@ class TestResourceDecorator:
def get_data() -> str:
return "Hello, world!"
resources_dict = await mcp.get_resources()
resources = list(resources_dict.values())
resources = await mcp.get_resources()
assert len(resources) == 1
assert resources[0].description == "Data resource"
@ -388,8 +384,7 @@ class TestResourceDecorator:
def get_data() -> str:
return "Hello, world!"
resources_dict = await mcp.get_resources()
resources = list(resources_dict.values())
resources = await mcp.get_resources()
assert len(resources) == 1
assert resources[0].tags == {"example", "test-tag"}
@ -499,8 +494,8 @@ class TestResourceDecorator:
def get_data() -> str:
return "Hello, world!"
resources_dict = await mcp.get_resources()
resource = resources_dict["resource://data"]
resources = await mcp.get_resources()
resource = next(r for r in resources if str(r.uri) == "resource://data")
assert resource.meta == meta_data
@ -568,8 +563,7 @@ class TestTemplateDecorator:
def get_data(name: str) -> str:
return f"Data for {name}"
templates_dict = await mcp.get_resource_templates()
templates = list(templates_dict.values())
templates = await mcp.get_resource_templates()
assert len(templates) == 1
assert templates[0].name == "get_data"
assert templates[0].uri_template == "resource://{name}/data"
@ -597,8 +591,7 @@ class TestTemplateDecorator:
def get_data(name: str) -> str:
return f"Data for {name}"
templates_dict = await mcp.get_resource_templates()
templates = list(templates_dict.values())
templates = await mcp.get_resource_templates()
assert len(templates) == 1
assert templates[0].name == "custom-template"
@ -614,8 +607,7 @@ class TestTemplateDecorator:
def get_data(name: str) -> str:
return f"Data for {name}"
templates_dict = await mcp.get_resource_templates()
templates = list(templates_dict.values())
templates = await mcp.get_resource_templates()
assert len(templates) == 1
assert templates[0].description == "Template description"
@ -698,8 +690,8 @@ class TestTemplateDecorator:
def template_resource(param: str) -> str:
return f"Template resource: {param}"
templates_dict = await mcp.get_resource_templates()
template = templates_dict["resource://{param}"]
templates = await mcp.get_resource_templates()
template = next(t for t in templates if t.uri_template == "resource://{param}")
assert template.tags == {"template", "test-tag"}
async def test_template_decorator_wildcard_param(self):
@ -709,8 +701,8 @@ class TestTemplateDecorator:
def template_resource(param: str) -> str:
return f"Template resource: {param}"
templates_dict = await mcp.get_resource_templates()
template = templates_dict["resource://{param*}"]
templates = await mcp.get_resource_templates()
template = next(t for t in templates if t.uri_template == "resource://{param*}")
assert template.uri_template == "resource://{param*}"
assert template.name == "template_resource"
@ -724,7 +716,9 @@ class TestTemplateDecorator:
def get_template_data(param: str) -> str:
return f"Data for {param}"
templates_dict = await mcp.get_resource_templates()
template = templates_dict["resource://{param}/data"]
templates = await mcp.get_resource_templates()
template = next(
t for t in templates if t.uri_template == "resource://{param}/data"
)
assert template.meta == meta_data

30
tests/server/providers/test_local_provider_tools.py
View file

@ -1188,7 +1188,7 @@ class TestToolDecorator:
return x + y
tools = await mcp.get_tools()
assert "add" in tools
assert any(t.name == "add" for t in tools)
async with Client(mcp) as client:
result = await client.call_tool("add", {"x": 1, "y": 2})
@ -1333,10 +1333,9 @@ class TestToolDecorator:
def sample_tool(x: int) -> int:
return x * 2
tools_dict = await mcp.get_tools()
assert len(tools_dict) == 1
only_tool = next(iter(tools_dict.values()))
assert only_tool.tags == {"example", "test-tag"}
tools = await mcp.get_tools()
assert len(tools) == 1
assert tools[0].tags == {"example", "test-tag"}
async def test_add_tool_with_custom_name(self):
"""Test adding a tool with a custom name using server.add_tool()."""
@ -1349,13 +1348,13 @@ class TestToolDecorator:
mcp.add_tool(Tool.from_function(multiply, name="custom_multiply"))
tools = await mcp.get_tools()
assert "custom_multiply" in tools
assert any(t.name == "custom_multiply" for t in tools)
async with Client(mcp) as client:
result = await client.call_tool("custom_multiply", {"a": 5, "b": 3})
assert result.data == 15
assert "multiply" not in tools
assert not any(t.name == "multiply" for t in tools)
async def test_tool_with_annotated_arguments(self):
"""Test that tools with annotated arguments work correctly."""
@ -1368,7 +1367,8 @@ class TestToolDecorator:
) -> None:
pass
tool = (await mcp.get_tools())["add"]
tools = await mcp.get_tools()
tool = next(t for t in tools if t.name == "add")
assert tool.parameters["properties"]["x"]["description"] == "x is an int"
assert tool.parameters["properties"]["y"]["description"] == "y is not an int"
@ -1383,7 +1383,8 @@ class TestToolDecorator:
) -> None:
pass
tool = (await mcp.get_tools())["add"]
tools = await mcp.get_tools()
tool = next(t for t in tools if t.name == "add")
assert tool.parameters["properties"]["x"]["description"] == "x is an int"
assert tool.parameters["properties"]["y"]["description"] == "y is not an int"
@ -1402,7 +1403,8 @@ class TestToolDecorator:
assert isinstance(result_fn, FunctionTool)
tools = await mcp.get_tools()
assert tools["direct_call_tool"] is result_fn
tool = next(t for t in tools if t.name == "direct_call_tool")
assert tool is result_fn
async with Client(mcp) as client:
result = await client.call_tool("direct_call_tool", {"x": 5, "y": 3})
@ -1418,8 +1420,8 @@ class TestToolDecorator:
return f"Result: {x}"
tools = await mcp.get_tools()
assert "string_named_tool" in tools
assert "my_function" not in tools
assert any(t.name == "string_named_tool" for t in tools)
assert not any(t.name == "my_function" for t in tools)
async with Client(mcp) as client:
result = await client.call_tool("string_named_tool", {"x": 42})
@ -1460,7 +1462,7 @@ class TestToolDecorator:
"""Multiply two numbers."""
return a * b
tools_dict = await mcp.get_tools()
tool = tools_dict["multiply"]
tools = await mcp.get_tools()
tool = next(t for t in tools if t.name == "multiply")
assert tool.meta == meta_data

55
tests/server/proxy/test_proxy_server.py
View file

@ -153,14 +153,14 @@ async def test_proxy_with_async_client_factory():
class TestTools:
async def test_get_tools(self, proxy_server):
tools = await proxy_server.get_tools()
assert "greet" in tools
assert "add" in tools
assert "error_tool" in tools
assert "tool_without_description" in tools
assert any(t.name == "greet" for t in tools)
assert any(t.name == "add" for t in tools)
assert any(t.name == "error_tool" for t in tools)
assert any(t.name == "tool_without_description" for t in tools)
async def test_get_tools_meta(self, proxy_server):
tools = await proxy_server.get_tools()
greet_tool = tools["greet"]
greet_tool = next(t for t in tools if t.name == "greet")
assert greet_tool.title == "Greet"
assert greet_tool.meta == {"_fastmcp": {"tags": ["greet"]}}
assert greet_tool.icons == [Icon(src="https://example.com/greet-icon.png")]
@ -174,8 +174,8 @@ class TestTools:
"add", ToolTransformConfig(name="add_transformed")
)
tools = await proxy_server.get_tools()
assert "add_transformed" in tools
assert "add" not in tools
assert any(t.name == "add_transformed" for t in tools)
assert not any(t.name == "add" for t in tools)
async def test_call_transformed_tools(
self, fastmcp_server: FastMCP, proxy_server: FastMCPProxy
@ -191,7 +191,8 @@ class TestTools:
async def test_tool_without_description(self, proxy_server):
tools = await proxy_server.get_tools()
assert tools["tool_without_description"].description is None
tool = next(t for t in tools if t.name == "tool_without_description")
assert tool.description is None
async def test_list_tools_same_as_original(self, fastmcp_server, proxy_server):
assert (
@ -266,15 +267,15 @@ class TestTools:
class TestResources:
async def test_get_resources(self, proxy_server):
resources = await proxy_server.get_resources()
assert [r.uri for r in resources.values()] == Contains(
assert [r.uri for r in resources] == Contains(
AnyUrl("data://users"),
AnyUrl("resource://wave"),
)
assert [r.name for r in resources.values()] == Contains("get_users", "wave")
assert [r.name for r in resources] == Contains("get_users", "wave")
async def test_get_resources_meta(self, proxy_server):
resources = await proxy_server.get_resources()
wave_resource = resources["resource://wave"]
wave_resource = next(r for r in resources if str(r.uri) == "resource://wave")
assert wave_resource.title == "Wave"
assert wave_resource.meta == {"_fastmcp": {"tags": ["wave"]}}
assert wave_resource.icons == [Icon(src="https://example.com/wave-icon.png")]
@ -345,11 +346,13 @@ class TestResources:
class TestResourceTemplates:
async def test_get_resource_templates(self, proxy_server):
templates = await proxy_server.get_resource_templates()
assert [t.name for t in templates.values()] == Contains("get_user")
assert [t.name for t in templates] == Contains("get_user")
async def test_get_resource_templates_meta(self, proxy_server):
templates = await proxy_server.get_resource_templates()
get_user_template = templates["data://user/{user_id}"]
get_user_template = next(
t for t in templates if t.uri_template == "data://user/{user_id}"
)
assert get_user_template.title == "User Template"
assert get_user_template.meta == {"_fastmcp": {"tags": ["users"]}}
assert get_user_template.icons == [
@ -420,11 +423,11 @@ class TestResourceTemplates:
class TestPrompts:
async def test_get_prompts_server_method(self, proxy_server: FastMCPProxy):
prompts = await proxy_server.get_prompts()
assert [p.name for p in prompts.values()] == Contains("welcome")
assert [p.name for p in prompts] == Contains("welcome")
async def test_get_prompts_meta(self, proxy_server):
prompts = await proxy_server.get_prompts()
welcome_prompt = prompts["welcome"]
welcome_prompt = next(p for p in prompts if p.name == "welcome")
assert welcome_prompt.title == "Welcome"
assert welcome_prompt.meta == {"_fastmcp": {"tags": ["welcome"]}}
assert welcome_prompt.icons == [
@ -505,15 +508,13 @@ async def test_proxy_handles_multiple_concurrent_tasks_correctly(
tg.start_soon(get_and_store, "tools", proxy_server.get_tools)
assert list(results) == Contains("resources", "prompts", "tools")
assert list(results["prompts"]) == Contains("welcome")
assert [r.uri for r in results["resources"].values()] == Contains(
assert [p.name for p in results["prompts"]] == Contains("welcome")
assert [r.uri for r in results["resources"]] == Contains(
AnyUrl("data://users"),
AnyUrl("resource://wave"),
)
assert [r.name for r in results["resources"].values()] == Contains(
"get_users", "wave"
)
assert list(results["tools"]) == Contains(
assert [r.name for r in results["resources"]] == Contains("get_users", "wave")
assert [t.name for t in results["tools"]] == Contains(
"greet", "add", "error_tool", "tool_without_description"
)
@ -524,7 +525,7 @@ class TestProxyComponentEnableDisable:
async def test_proxy_tool_enable_raises_not_implemented(self, proxy_server):
"""Test that enable() on proxy tools raises NotImplementedError."""
tools = await proxy_server.get_tools()
tool = tools["greet"]
tool = next(t for t in tools if t.name == "greet")
with pytest.raises(NotImplementedError, match="server.enable"):
tool.enable()
@ -532,7 +533,7 @@ class TestProxyComponentEnableDisable:
async def test_proxy_tool_disable_raises_not_implemented(self, proxy_server):
"""Test that disable() on proxy tools raises NotImplementedError."""
tools = await proxy_server.get_tools()
tool = tools["greet"]
tool = next(t for t in tools if t.name == "greet")
with pytest.raises(NotImplementedError, match="server.disable"):
tool.disable()
@ -540,7 +541,7 @@ class TestProxyComponentEnableDisable:
async def test_proxy_resource_enable_raises_not_implemented(self, proxy_server):
"""Test that enable() on proxy resources raises NotImplementedError."""
resources = await proxy_server.get_resources()
resource = resources["resource://wave"]
resource = next(r for r in resources if str(r.uri) == "resource://wave")
with pytest.raises(NotImplementedError, match="server.enable"):
resource.enable()
@ -548,7 +549,7 @@ class TestProxyComponentEnableDisable:
async def test_proxy_resource_disable_raises_not_implemented(self, proxy_server):
"""Test that disable() on proxy resources raises NotImplementedError."""
resources = await proxy_server.get_resources()
resource = resources["resource://wave"]
resource = next(r for r in resources if str(r.uri) == "resource://wave")
with pytest.raises(NotImplementedError, match="server.disable"):
resource.disable()
@ -556,7 +557,7 @@ class TestProxyComponentEnableDisable:
async def test_proxy_prompt_enable_raises_not_implemented(self, proxy_server):
"""Test that enable() on proxy prompts raises NotImplementedError."""
prompts = await proxy_server.get_prompts()
prompt = prompts["welcome"]
prompt = next(p for p in prompts if p.name == "welcome")
with pytest.raises(NotImplementedError, match="server.enable"):
prompt.enable()
@ -564,7 +565,7 @@ class TestProxyComponentEnableDisable:
async def test_proxy_prompt_disable_raises_not_implemented(self, proxy_server):
"""Test that disable() on proxy prompts raises NotImplementedError."""
prompts = await proxy_server.get_prompts()
prompt = prompts["welcome"]
prompt = next(p for p in prompts if p.name == "welcome")
with pytest.raises(NotImplementedError, match="server.disable"):
prompt.disable()

84
tests/server/test_mount.py
View file

@ -40,8 +40,8 @@ class TestBasicMount:
# Get tools from main app, should include sub_app's tools
tools = await main_app.get_tools()
assert "sub_tool" in tools
assert "sub_transformed_tool" in tools
assert any(t.name == "sub_tool" for t in tools)
assert any(t.name == "sub_transformed_tool" for t in tools)
async with Client(main_app) as client:
result = await client.call_tool("sub_tool", {})
@ -61,7 +61,7 @@ class TestBasicMount:
# Tool should be accessible with the default separator
tools = await main_app.get_tools()
assert "sub_greet" in tools
assert any(t.name == "sub_greet" for t in tools)
# Call the tool
async with Client(main_app) as client:
@ -82,7 +82,7 @@ class TestBasicMount:
tools = await main_app.get_tools()
# With empty prefix, the tool should keep its original name
assert "sub_tool" in tools
assert any(t.name == "sub_tool" for t in tools)
async def test_mount_with_no_prefix_provided(self):
"""Test mounting without providing a prefix at all."""
@ -98,7 +98,7 @@ class TestBasicMount:
tools = await main_app.get_tools()
# Without prefix, the tool should keep its original name
assert "sub_tool" in tools
assert any(t.name == "sub_tool" for t in tools)
# Call the tool to verify it works
async with Client(main_app) as client:
@ -119,7 +119,7 @@ class TestBasicMount:
# Verify tool is accessible with original name
tools = await main_app.get_tools()
assert "sub_tool" in tools
assert any(t.name == "sub_tool" for t in tools)
# Test actual functionality
async with Client(main_app) as client:
@ -140,7 +140,7 @@ class TestBasicMount:
# Verify resource is accessible with original URI
resources = await main_app.get_resources()
assert "data://config" in resources
assert any(str(r.uri) == "data://config" for r in resources)
# Test actual functionality
async with Client(main_app) as client:
@ -162,7 +162,7 @@ class TestBasicMount:
# Verify template is accessible with original URI template
templates = await main_app.get_resource_templates()
assert "users://{user_id}/info" in templates
assert any(t.uri_template == "users://{user_id}/info" for t in templates)
# Test actual functionality
async with Client(main_app) as client:
@ -184,7 +184,7 @@ class TestBasicMount:
# Verify prompt is accessible with original name
prompts = await main_app.get_prompts()
assert "sub_prompt" in prompts
assert any(p.name == "sub_prompt" for p in prompts)
# Test actual functionality
async with Client(main_app) as client:
@ -215,8 +215,8 @@ class TestMultipleServerMount:
# Check both are accessible
tools = await main_app.get_tools()
assert "weather_get_forecast" in tools
assert "news_get_headlines" in tools
assert any(t.name == "weather_get_forecast" for t in tools)
assert any(t.name == "news_get_headlines" for t in tools)
# Call tools from both mounted servers
async with Client(main_app) as client:
@ -242,15 +242,15 @@ class TestMultipleServerMount:
# Mount first app
main_app.mount(first_app, "api")
tools = await main_app.get_tools()
assert "api_first_tool" in tools
assert any(t.name == "api_first_tool" for t in tools)
# Mount second app with same prefix
main_app.mount(second_app, "api")
tools = await main_app.get_tools()
# Both apps' tools should be accessible (new behavior)
assert "api_first_tool" in tools
assert "api_second_tool" in tools
assert any(t.name == "api_first_tool" for t in tools)
assert any(t.name == "api_second_tool" for t in tools)
@pytest.mark.skipif(
sys.platform == "win32", reason="Windows asyncio networking timeouts."
@ -593,7 +593,7 @@ class TestDynamicChanges:
# Initially, there should be no tools from sub_app
tools = await main_app.get_tools()
assert not any(key.startswith("sub_") for key in tools)
assert not any(t.name.startswith("sub_") for t in tools)
# Add a tool to the sub-app after mounting
@sub_app.tool
@ -602,7 +602,7 @@ class TestDynamicChanges:
# The tool should be accessible through the main app
tools = await main_app.get_tools()
assert "sub_dynamic_tool" in tools
assert any(t.name == "sub_dynamic_tool" for t in tools)
# Call the dynamically added tool
async with Client(main_app) as client:
@ -623,14 +623,14 @@ class TestDynamicChanges:
# Initially, the tool should be accessible
tools = await main_app.get_tools()
assert "sub_temp_tool" in tools
assert any(t.name == "sub_temp_tool" for t in tools)
# Remove the tool from sub_app using public API
sub_app.remove_tool("temp_tool")
# The tool should no longer be accessible
tools = await main_app.get_tools()
assert "sub_temp_tool" not in tools
assert not any(t.name == "sub_temp_tool" for t in tools)
class TestResourcesAndTemplates:
@ -650,7 +650,7 @@ class TestResourcesAndTemplates:
# Resource should be accessible through main app
resources = await main_app.get_resources()
assert "data://data/users" in resources
assert any(str(r.uri) == "data://data/users" for r in resources)
# Check that resource can be accessed
async with Client(main_app) as client:
@ -672,7 +672,7 @@ class TestResourcesAndTemplates:
# Template should be accessible through main app
templates = await main_app.get_resource_templates()
assert "users://api/{user_id}/profile" in templates
assert any(t.uri_template == "users://api/{user_id}/profile" for t in templates)
# Check template instantiation
async with Client(main_app) as client:
@ -697,7 +697,7 @@ class TestResourcesAndTemplates:
# Resource should be accessible through main app
resources = await main_app.get_resources()
assert "data://data/config" in resources
assert any(str(r.uri) == "data://data/config" for r in resources)
# Check access to the resource
async with Client(main_app) as client:
@ -724,7 +724,7 @@ class TestPrompts:
# Prompt should be accessible through main app
prompts = await main_app.get_prompts()
assert "assistant_greeting" in prompts
assert any(p.name == "assistant_greeting" for p in prompts)
# Render the prompt
async with Client(main_app) as client:
@ -747,7 +747,7 @@ class TestPrompts:
# Prompt should be accessible through main app
prompts = await main_app.get_prompts()
assert "assistant_farewell" in prompts
assert any(p.name == "assistant_farewell" for p in prompts)
# Render the prompt
async with Client(main_app) as client:
@ -777,7 +777,7 @@ class TestProxyServer:
# Tool should be accessible through main app
tools = await main_app.get_tools()
assert "proxy_get_data" in tools
assert any(t.name == "proxy_get_data" for t in tools)
# Call the tool
async with Client(main_app) as client:
@ -803,7 +803,7 @@ class TestProxyServer:
# Tool should be accessible through main app via proxy
tools = await main_app.get_tools()
assert "proxy_dynamic_data" in tools
assert any(t.name == "proxy_dynamic_data" for t in tools)
# Call the tool
async with Client(main_app) as client:
@ -1021,10 +1021,12 @@ class TestResourceUriPrefixing:
resources = await main_app.get_resources()
# Should have prefixed key (using path format: resource://prefix/resource_name)
assert "resource://prefix/my_resource" in resources
assert any(str(r.uri) == "resource://prefix/my_resource" for r in resources)
# The resource name should NOT be prefixed (only URI is prefixed)
resource = resources["resource://prefix/my_resource"]
resource = next(
r for r in resources if str(r.uri) == "resource://prefix/my_resource"
)
assert resource.name == "my_resource"
async def test_resource_template_uri_prefixing(self):
@ -1045,10 +1047,14 @@ class TestResourceUriPrefixing:
templates = await main_app.get_resource_templates()
# Should have prefixed key (using path format: resource://prefix/template_uri)
assert "resource://prefix/user/{user_id}" in templates
assert any(
t.uri_template == "resource://prefix/user/{user_id}" for t in templates
)
# The template name should NOT be prefixed (only URI template is prefixed)
template = templates["resource://prefix/user/{user_id}"]
template = next(
t for t in templates if t.uri_template == "resource://prefix/user/{user_id}"
)
assert template.name == "user_template"
@ -1393,9 +1399,9 @@ class TestToolNameOverrides:
# Server introspection shows transformed names
tools = await main.get_tools()
assert "custom_name" in tools
assert "original_tool" not in tools
assert "prefix_original_tool" not in tools
assert any(t.name == "custom_name" for t in tools)
assert not any(t.name == "original_tool" for t in tools)
assert not any(t.name == "prefix_original_tool" for t in tools)
# Client-facing API shows the same transformed names
async with Client(main) as client:
@ -1514,7 +1520,7 @@ class TestComponentServicePrefixLess:
# Initially the tool is enabled
tools = await main_app.get_tools()
assert "my_tool" in tools
assert any(t.name == "my_tool" for t in tools)
# Disable and re-enable via ComponentService
service = ComponentService(main_app)
@ -1522,13 +1528,13 @@ class TestComponentServicePrefixLess:
assert tool is not None
# Verify tool is now disabled
tools = await main_app.get_tools()
assert "my_tool" not in tools
assert not any(t.name == "my_tool" for t in tools)
tool = await service._enable_tool("my_tool")
assert tool is not None
# Verify tool is now enabled
tools = await main_app.get_tools()
assert "my_tool" in tools
assert any(t.name == "my_tool" for t in tools)
async def test_enable_resource_prefixless_mount(self):
"""Test enabling a resource on a prefix-less mounted server."""
@ -1550,13 +1556,13 @@ class TestComponentServicePrefixLess:
assert resource is not None
# Verify resource is now disabled
resources = await main_app.get_resources()
assert "data://test" not in resources
assert not any(str(r.uri) == "data://test" for r in resources)
resource = await service._enable_resource("data://test")
assert resource is not None
# Verify resource is now enabled
resources = await main_app.get_resources()
assert "data://test" in resources
assert any(str(r.uri) == "data://test" for r in resources)
async def test_enable_prompt_prefixless_mount(self):
"""Test enabling a prompt on a prefix-less mounted server."""
@ -1578,10 +1584,10 @@ class TestComponentServicePrefixLess:
assert prompt is not None
# Verify prompt is now disabled
prompts = await main_app.get_prompts()
assert "my_prompt" not in prompts
assert not any(p.name == "my_prompt" for p in prompts)
prompt = await service._enable_prompt("my_prompt")
assert prompt is not None
# Verify prompt is now enabled
prompts = await main_app.get_prompts()
assert "my_prompt" in prompts
assert any(p.name == "my_prompt" for p in prompts)

22
tests/server/test_server.py
View file

@ -85,11 +85,11 @@ class TestTools:
return a + b
mcp_tools = await mcp.get_tools()
assert "adder" in mcp_tools
assert any(t.name == "adder" for t in mcp_tools)
mcp.remove_tool("adder")
mcp_tools = await mcp.get_tools()
assert "adder" not in mcp_tools
assert not any(t.name == "adder" for t in mcp_tools)
with pytest.raises(NotFoundError, match="Unknown tool: adder"):
await mcp._call_tool_mcp("adder", {"a": 1, "b": 2})
@ -108,9 +108,11 @@ class TestTools:
tools = await mcp.get_tools()
assert len(tools) == 2
assert tools["f"].name == "f"
assert tools["g-tool"].name == "g-tool"
assert tools["g-tool"].description == "add two to a number"
f_tool = next(t for t in tools if t.name == "f")
g_tool = next(t for t in tools if t.name == "g-tool")
assert f_tool.name == "f"
assert g_tool.name == "g-tool"
assert g_tool.description == "add two to a number"
class TestServerDelegation:
@ -177,7 +179,7 @@ class TestServerDelegation:
return "local"
tools = await mcp.get_tools()
assert "local_tool" in tools
assert any(t.name == "local_tool" for t in tools)
class TestResourcePrefixMounting:
@ -208,9 +210,11 @@ class TestResourcePrefixMounting:
resources = await main_server.get_resources()
templates = await main_server.get_resource_templates()
assert "resource://prefix/test-resource" in resources
assert "resource://prefix//absolute/path" in resources
assert "resource://prefix/{param}/template" in templates
assert any(str(r.uri) == "resource://prefix/test-resource" for r in resources)
assert any(str(r.uri) == "resource://prefix//absolute/path" for r in resources)
assert any(
t.uri_template == "resource://prefix/{param}/template" for t in templates
)
# Test that prefixed resources can be accessed
async with Client(main_server) as client:

67
tests/server/test_server_interactions.py
View file

@ -277,19 +277,19 @@ class TestToolEnabled:
# Tool is enabled by default
tools = await mcp.get_tools()
assert "sample_tool" in tools
assert any(t.name == "sample_tool" for t in tools)
# Disable via server
mcp.disable(keys=["tool:sample_tool"])
# Tool should not be in list when disabled
tools = await mcp.get_tools()
assert "sample_tool" not in tools
assert not any(t.name == "sample_tool" for t in tools)
# Re-enable via server
mcp.enable(keys=["tool:sample_tool"])
tools = await mcp.get_tools()
assert "sample_tool" in tools
assert any(t.name == "sample_tool" for t in tools)
async def test_tool_disabled_via_server(self):
mcp = FastMCP()
@ -550,17 +550,17 @@ class TestResourceEnabled:
return "Hello, world!"
resources = await mcp.get_resources()
assert "resource://data" in resources
assert any(str(r.uri) == "resource://data" for r in resources)
mcp.disable(keys=["resource:resource://data"])
resources = await mcp.get_resources()
assert "resource://data" not in resources
assert not any(str(r.uri) == "resource://data" for r in resources)
mcp.enable(keys=["resource:resource://data"])
resources = await mcp.get_resources()
assert "resource://data" in resources
assert any(str(r.uri) == "resource://data" for r in resources)
async def test_resource_disabled(self):
mcp = FastMCP()
@ -588,7 +588,7 @@ class TestResourceEnabled:
mcp.disable(keys=["resource:resource://data"])
resources = await mcp.get_resources()
assert "resource://data" not in resources
assert not any(str(r.uri) == "resource://data" for r in resources)
mcp.enable(keys=["resource:resource://data"])
@ -729,17 +729,17 @@ class TestResourceTemplateEnabled:
return f"Template: {param}"
templates = await mcp.get_resource_templates()
assert "resource://{param}" in templates
assert any(t.uri_template == "resource://{param}" for t in templates)
mcp.disable(keys=["template:resource://{param}"])
templates = await mcp.get_resource_templates()
assert "resource://{param}" not in templates
assert not any(t.uri_template == "resource://{param}" for t in templates)
mcp.enable(keys=["template:resource://{param}"])
templates = await mcp.get_resource_templates()
assert "resource://{param}" in templates
assert any(t.uri_template == "resource://{param}" for t in templates)
async def test_template_disabled(self):
mcp = FastMCP()
@ -767,7 +767,7 @@ class TestResourceTemplateEnabled:
mcp.disable(keys=["template:resource://{param}"])
templates = await mcp.get_resource_templates()
assert "resource://{param}" not in templates
assert not any(t.uri_template == "resource://{param}" for t in templates)
mcp.enable(keys=["template:resource://{param}"])
@ -835,9 +835,9 @@ class TestPrompts:
def fn() -> str:
return "Hello, world!"
prompts_dict = await mcp.get_prompts()
assert len(prompts_dict) == 1
prompt = prompts_dict["fn"]
prompts = await mcp.get_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "fn")
assert prompt.name == "fn"
# Don't compare functions directly since validate_call wraps them
content = await prompt.render()
@ -854,9 +854,9 @@ class TestPrompts:
def fn() -> str:
return "Hello, world!"
prompts_dict = await mcp.get_prompts()
assert len(prompts_dict) == 1
prompt = prompts_dict["custom_name"]
prompts = await mcp.get_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "custom_name")
assert prompt.name == "custom_name"
content = await prompt.render()
if not isinstance(content, PromptResult):
@ -872,9 +872,9 @@ class TestPrompts:
def fn() -> str:
return "Hello, world!"
prompts_dict = await mcp.get_prompts()
assert len(prompts_dict) == 1
prompt = prompts_dict["fn"]
prompts = await mcp.get_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "fn")
assert prompt.description == "A custom description"
content = await prompt.render()
if not isinstance(content, PromptResult):
@ -889,9 +889,9 @@ class TestPrompts:
def fn() -> str:
return "Hello, world!"
prompts_dict = await mcp.get_prompts()
assert len(prompts_dict) == 1
prompt = prompts_dict["fn"]
prompts = await mcp.get_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "fn")
assert prompt.name == "fn"
async def test_list_prompts(self):
@ -1044,8 +1044,7 @@ class TestPrompts:
def get_data() -> str:
return "Hello, world!"
resources_dict = await mcp.get_resources()
resources = list(resources_dict.values())
resources = await mcp.get_resources()
assert len(resources) == 1
assert resources[0].tags == {"example", "test-tag"}
@ -1057,8 +1056,8 @@ class TestPrompts:
def template_resource(param: str) -> str:
return f"Template resource: {param}"
templates_dict = await mcp.get_resource_templates()
template = templates_dict["resource://{param}"]
templates = await mcp.get_resource_templates()
template = next(t for t in templates if t.uri_template == "resource://{param}")
assert template.tags == {"template", "test-tag"}
async def test_prompt_decorator_with_tags(self):
@ -1069,9 +1068,9 @@ class TestPrompts:
def sample_prompt() -> str:
return "Hello, world!"
prompts_dict = await mcp.get_prompts()
assert len(prompts_dict) == 1
prompt = prompts_dict["sample_prompt"]
prompts = await mcp.get_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "sample_prompt")
assert prompt.tags == {"example", "test-tag"}
@ -1084,17 +1083,17 @@ class TestPromptEnabled:
return "Hello, world!"
prompts = await mcp.get_prompts()
assert "sample_prompt" in prompts
assert any(p.name == "sample_prompt" for p in prompts)
mcp.disable(keys=["prompt:sample_prompt"])
prompts = await mcp.get_prompts()
assert "sample_prompt" not in prompts
assert not any(p.name == "sample_prompt" for p in prompts)
mcp.enable(keys=["prompt:sample_prompt"])
prompts = await mcp.get_prompts()
assert "sample_prompt" in prompts
assert any(p.name == "sample_prompt" for p in prompts)
async def test_prompt_disabled(self):
mcp = FastMCP()
@ -1122,7 +1121,7 @@ class TestPromptEnabled:
mcp.disable(keys=["prompt:sample_prompt"])
prompts = await mcp.get_prompts()
assert "sample_prompt" not in prompts
assert not any(p.name == "sample_prompt" for p in prompts)
mcp.enable(keys=["prompt:sample_prompt"])

9
tests/server/test_tool_annotations.py
View file

@ -23,8 +23,7 @@ async def test_tool_annotations_in_tool_manager():
return message
# Check internal tool objects directly
tools_dict = await mcp.get_tools()
tools = list(tools_dict.values())
tools = await mcp.get_tools()
assert len(tools) == 1
assert tools[0].annotations is not None
assert tools[0].annotations.title == "Echo Tool"
@ -126,8 +125,7 @@ async def test_direct_tool_annotations_in_tool_manager():
return {"modified": True, **data}
# Check internal tool objects directly
tools_dict = await mcp.get_tools()
tools = list(tools_dict.values())
tools = await mcp.get_tools()
assert len(tools) == 1
assert tools[0].annotations is not None
assert tools[0].annotations.title == "Direct Tool"
@ -186,8 +184,7 @@ async def test_add_tool_method_annotations():
mcp.add_tool(tool)
# Check internal tool objects directly
tools_dict = await mcp.get_tools()
tools = list(tools_dict.values())
tools = await mcp.get_tools()
assert len(tools) == 1
assert tools[0].annotations is not None
assert tools[0].annotations.title == "Create Item"

12
tests/server/test_tool_transformation.py
View file

@ -13,11 +13,11 @@ async def test_tool_transformation_in_tool_manager():
mcp.add_tool_transformation("echo", ToolTransformConfig(name="echo_transformed"))
tools_dict = await mcp.get_tools()
tools = list(tools_dict.values())
tools = await mcp.get_tools()
assert len(tools) == 1
assert "echo_transformed" in tools_dict
assert tools_dict["echo_transformed"].name == "echo_transformed"
assert any(t.name == "echo_transformed" for t in tools)
tool = next(t for t in tools if t.name == "echo_transformed")
assert tool.name == "echo_transformed"
async def test_transformed_tool_filtering():
@ -29,14 +29,14 @@ async def test_transformed_tool_filtering():
"""Echo back the message provided."""
return message
tools = list(await mcp._list_tools_middleware())
tools = await mcp.get_tools(apply_middleware=True)
assert len(tools) == 0
mcp.add_tool_transformation(
"echo", ToolTransformConfig(name="echo_transformed", tags={"enabled_tools"})
)
tools = list(await mcp._list_tools_middleware())
tools = await mcp.get_tools(apply_middleware=True)
assert len(tools) == 1

69
v3-notes/get-methods-consolidation.md Normal file
View file

@ -0,0 +1,69 @@
# Consolidating get_* and _list_* Methods
This document captures the design decision to consolidate component listing methods in FastMCP 3.0.
## Problem
The server had parallel implementations for listing components:
- `get_tools()` / `_list_tools()`
- `get_resources()` / `_list_resources()`
- `get_prompts()` / `_list_prompts()`
- `get_resource_templates()` / `_list_resource_templates()`
These were nearly identical but with subtle differences in dedup keys, logging, and return types. The `_list_*` methods were internal and used by the MCP protocol handlers, while `get_*` methods were the public API.
## Solution
`get_*` is now the canonical method. The `_list_*` methods were deleted entirely.
```python
async def get_tools(self, *, apply_middleware: bool = False) -> list[Tool]:
"""Canonical method for listing tools."""
if apply_middleware:
# Apply middleware chain (for MCP protocol handlers)
mw_context = MiddlewareContext(...)
return await self._apply_middleware(
context=mw_context,
call_next=lambda context: self.get_tools(apply_middleware=False)
)
# Core implementation: query providers, dedupe, filter visibility
...
```
## Key Changes
### Return Type: dict → list
The dict return type was removed because the key was redundant—components already have `.name` or `.uri` attributes.
```python
# Before
tools = await server.get_tools()
tool = tools["my_tool"]
# After
tools = await server.get_tools()
tool = next(t for t in tools if t.name == "my_tool")
```
### Middleware via Parameter
The `apply_middleware=True` parameter applies the middleware chain. This replaces the separate `_list_*_middleware()` methods.
Callers:
- MCP protocol handlers: `get_tools(apply_middleware=True)`
- Direct access: `get_tools()` (default False)
## Benefits
1. **Single source of truth** - One method, not two
2. **Consistent behavior** - Same dedup key, same visibility filtering
3. **Clearer API** - Public method with explicit middleware opt-in
4. **Less code** - Deleted ~200 lines of duplicate implementation
## Implementation Files
- `src/fastmcp/server/server.py` - Canonical `get_*` methods
- `src/fastmcp/server/providers/fastmcp_provider.py` - Uses `apply_middleware=True`
- `src/fastmcp/utilities/inspect.py` - Uses `apply_middleware=True`