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

Unify discovery API: deduplicate at protocol layer only (#2919)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-01-18 21:01:53 -05:00 • committed by GitHub
commit c3111a8978
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
27 changed files with 734 additions and 589 deletions
Download patch file Download diff file Expand all files Collapse all files

8
src/fastmcp/contrib/mcp_mixin/example.py
View file

@ -40,11 +40,11 @@ first_sample.register_all(mcp_server=mcp, prefix="first")
second_sample.register_all(mcp_server=mcp, prefix="second")
async def list_components():
async def list_components() -> None:
print("MCP Server running with registered components...")
print("Tools:", list(await mcp.get_tools()))
print("Resources:", list(await mcp.get_resources()))
print("Prompts:", list(await mcp.get_prompts()))
print("Tools:", list(await mcp.list_tools()))
print("Resources:", list(await mcp.list_resources()))
print("Prompts:", list(await mcp.list_prompts()))
if __name__ == "__main__":

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

@ -114,11 +114,14 @@ class FastMCPProviderTool(Tool):
backgrounding appropriately. fn_key is already set by the parent
server before calling this method.
"""
# Pass exact version so child executes the correct version
version = VersionSpec(eq=self.version) if self.version else None
with delegate_span(
self._original_name or "", "FastMCPProvider", self._original_name or ""
):
return await self._server.call_tool(
self._original_name, arguments, task_meta=task_meta
self._original_name, arguments, version=version, task_meta=task_meta
)
async def run(self, arguments: dict[str, Any]) -> ToolResult:
@ -127,7 +130,12 @@ class FastMCPProviderTool(Tool):
This is called when the tool is used within a TransformedTool
forwarding function or other contexts where task_meta is not available.
"""
result = await self._server.call_tool(self._original_name, arguments)
# Pass exact version so child executes the correct version
version = VersionSpec(eq=self.version) if self.version else None
result = await self._server.call_tool(
self._original_name, arguments, version=version
)
# Result from call_tool should always be ToolResult when no task_meta
if isinstance(result, mcp.types.CreateTaskResult):
raise RuntimeError(
@ -193,11 +201,14 @@ class FastMCPProviderResource(Resource):
backgrounding appropriately. fn_key is already set by the parent
server before calling this method.
"""
# Pass exact version so child reads the correct version
version = VersionSpec(eq=self.version) if self.version else None
with delegate_span(
self._original_uri or "", "FastMCPProvider", self._original_uri or ""
):
return await self._server.read_resource(
self._original_uri, task_meta=task_meta
self._original_uri, version=version, task_meta=task_meta
)
def get_span_attributes(self) -> dict[str, Any]:
@ -266,11 +277,14 @@ class FastMCPProviderPrompt(Prompt):
backgrounding appropriately. fn_key is already set by the parent
server before calling this method.
"""
# Pass exact version so child renders the correct version
version = VersionSpec(eq=self.version) if self.version else None
with delegate_span(
self._original_name or "", "FastMCPProvider", self._original_name or ""
):
return await self._server.render_prompt(
self._original_name, arguments, task_meta=task_meta
self._original_name, arguments, version=version, task_meta=task_meta
)
async def render(self, arguments: dict[str, Any] | None = None) -> PromptResult:
@ -279,7 +293,12 @@ class FastMCPProviderPrompt(Prompt):
This is called when the prompt is used within a transformed context
or other contexts where task_meta is not available.
"""
result = await self._server.render_prompt(self._original_name, arguments)
# Pass exact version so child renders the correct version
version = VersionSpec(eq=self.version) if self.version else None
result = await self._server.render_prompt(
self._original_name, arguments, version=version
)
# Result from render_prompt should always be PromptResult when no task_meta
if isinstance(result, mcp.types.CreateTaskResult):
raise RuntimeError(
@ -374,10 +393,15 @@ class FastMCPProviderResourceTemplate(ResourceTemplate):
# Expand the original template with params to get internal URI
original_uri = _expand_uri_template(self._original_uri_template or "", params)
# Pass exact version so child reads the correct version
version = VersionSpec(eq=self.version) if self.version else None
with delegate_span(
original_uri, "FastMCPProvider", self._original_uri_template or ""
):
return await self._server.read_resource(original_uri, task_meta=task_meta)
return await self._server.read_resource(
original_uri, version=version, task_meta=task_meta
)
async def read(self, arguments: dict[str, Any]) -> str | bytes | ResourceResult:
"""Read the resource content for background task execution.
@ -390,8 +414,11 @@ class FastMCPProviderResourceTemplate(ResourceTemplate):
self._original_uri_template or "", arguments
)
# Pass exact version so child reads the correct version
version = VersionSpec(eq=self.version) if self.version else None
# Read from the wrapped server
result = await self._server.read_resource(original_uri)
result = await self._server.read_resource(original_uri, version=version)
if isinstance(result, mcp.types.CreateTaskResult):
raise RuntimeError("Unexpected CreateTaskResult during Docket execution")
@ -489,7 +516,7 @@ class FastMCPProvider(Provider):
Wraps each tool as a FastMCPProviderTool that delegates execution to
the nested server's middleware.
"""
raw_tools = await self.server.get_tools(run_middleware=True)
raw_tools = await self.server.list_tools()
return [FastMCPProviderTool.wrap(self.server, t) for t in raw_tools]
async def _get_tool(
@ -517,7 +544,7 @@ class FastMCPProvider(Provider):
Wraps each resource as a FastMCPProviderResource that delegates reading
to the nested server's middleware.
"""
raw_resources = await self.server.get_resources(run_middleware=True)
raw_resources = await self.server.list_resources()
return [FastMCPProviderResource.wrap(self.server, r) for r in raw_resources]
async def _get_resource(
@ -545,7 +572,7 @@ class FastMCPProvider(Provider):
Returns FastMCPProviderResourceTemplate instances that create
FastMCPProviderResources when materialized.
"""
raw_templates = await self.server.get_resource_templates(run_middleware=True)
raw_templates = await self.server.list_resource_templates()
return [
FastMCPProviderResourceTemplate.wrap(self.server, t) for t in raw_templates
]
@ -575,7 +602,7 @@ class FastMCPProvider(Provider):
Returns FastMCPProviderPrompt instances that delegate rendering to the
wrapped server's middleware.
"""
raw_prompts = await self.server.get_prompts(run_middleware=True)
raw_prompts = await self.server.list_prompts()
return [FastMCPProviderPrompt.wrap(self.server, p) for p in raw_prompts]
async def _get_prompt(

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

@ -1005,15 +1005,12 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
stacklevel=2,
)
async def get_tools(self, *, run_middleware: bool = False) -> list[Tool]:
"""Get all enabled tools from providers.
async def list_tools(self, *, run_middleware: bool = True) -> Sequence[Tool]:
"""List all enabled tools from providers.
Queries all providers via the root provider (which applies provider transforms,
server transforms, and enabled filtering). First provider wins for duplicate keys.
Args:
run_middleware: If True, apply the middleware chain before returning.
Used by MCP handlers and FastMCPProvider for nested servers.
Overrides Provider.list_tools() to add enabled filtering, auth filtering,
and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
@ -1026,17 +1023,11 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.get_tools(run_middleware=False),
call_next=lambda context: self.list_tools(run_middleware=False),
)
# Query through full transform chain (provider transforms + server transforms)
# Then apply enabled filtering at the server level
tools = [t for t in await self.list_tools() if is_enabled(t)]
# Get auth context (skip_auth=True for STDIO which has no auth concept)
tools = [t for t in await super().list_tools() if is_enabled(t)]
skip_auth, token = _get_auth_context()
# Filter by auth
authorized: list[Tool] = []
for tool in tools:
if not skip_auth and tool.auth is not None:
@ -1047,8 +1038,7 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
except AuthorizationError:
continue
authorized.append(tool)
return _dedupe_with_versions(authorized, lambda t: t.name)
return authorized
async def _get_tool(
self, name: str, version: VersionSpec | None = None
@ -1102,15 +1092,14 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
return None
return tool
async def get_resources(self, *, run_middleware: bool = False) -> list[Resource]:
"""Get all enabled resources from providers.
async def list_resources(
self, *, run_middleware: bool = True
) -> Sequence[Resource]:
"""List all enabled resources from providers.
Queries all providers via the root provider (which applies provider transforms,
server transforms, and enabled filtering). First provider wins for duplicate keys.
Args:
run_middleware: If True, apply the middleware chain before returning.
Used by MCP handlers and FastMCPProvider for nested servers.
Overrides Provider.list_resources() to add enabled filtering, auth filtering,
and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
@ -1123,16 +1112,11 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.get_resources(run_middleware=False),
call_next=lambda context: self.list_resources(run_middleware=False),
)
# Query through full transform chain, then apply enabled filtering
resources = [r for r in await self.list_resources() if is_enabled(r)]
# Get auth context (skip_auth=True for STDIO which has no auth concept)
resources = [r for r in await super().list_resources() if is_enabled(r)]
skip_auth, token = _get_auth_context()
# Filter by auth
authorized: list[Resource] = []
for resource in resources:
if not skip_auth and resource.auth is not None:
@ -1143,8 +1127,7 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
except AuthorizationError:
continue
authorized.append(resource)
return _dedupe_with_versions(authorized, lambda r: str(r.uri))
return authorized
async def _get_resource(
self, uri: str, version: VersionSpec | None = None
@ -1197,17 +1180,14 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
return None
return resource
async def get_resource_templates(
self, *, run_middleware: bool = False
) -> list[ResourceTemplate]:
"""Get all enabled resource templates from providers.
async def list_resource_templates(
self, *, run_middleware: bool = True
) -> Sequence[ResourceTemplate]:
"""List all enabled resource templates from providers.
Queries all providers via the root provider (which applies provider transforms,
server transforms, and enabled filtering). First provider wins for duplicate keys.
Args:
run_middleware: If True, apply the middleware chain before returning.
Used by MCP handlers and FastMCPProvider for nested servers.
Overrides Provider.list_resource_templates() to add enabled filtering,
auth filtering, and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
@ -1220,20 +1200,15 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.get_resource_templates(
call_next=lambda context: self.list_resource_templates(
run_middleware=False
),
)
# Query through full transform chain, then apply enabled filtering
templates = [
t for t in await self.list_resource_templates() if is_enabled(t)
t for t in await super().list_resource_templates() if is_enabled(t)
]
# Get auth context (skip_auth=True for STDIO which has no auth concept)
skip_auth, token = _get_auth_context()
# Filter by auth
authorized: list[ResourceTemplate] = []
for template in templates:
if not skip_auth and template.auth is not None:
@ -1244,8 +1219,7 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
except AuthorizationError:
continue
authorized.append(template)
return _dedupe_with_versions(authorized, lambda t: t.uri_template)
return authorized
async def _get_resource_template(
self, uri: str, version: VersionSpec | None = None
@ -1298,15 +1272,12 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
return None
return template
async def get_prompts(self, *, run_middleware: bool = False) -> list[Prompt]:
"""Get all enabled prompts from providers.
async def list_prompts(self, *, run_middleware: bool = True) -> Sequence[Prompt]:
"""List all enabled prompts from providers.
Queries all providers via the root provider (which applies provider transforms,
server transforms, and enabled filtering). First provider wins for duplicate keys.
Args:
run_middleware: If True, apply the middleware chain before returning.
Used by MCP handlers and FastMCPProvider for nested servers.
Overrides Provider.list_prompts() to add enabled filtering, auth filtering,
and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
@ -1319,16 +1290,11 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.get_prompts(run_middleware=False),
call_next=lambda context: self.list_prompts(run_middleware=False),
)
# Query through full transform chain, then apply enabled filtering
prompts = [p for p in await self.list_prompts() if is_enabled(p)]
# Get auth context (skip_auth=True for STDIO which has no auth concept)
prompts = [p for p in await super().list_prompts() if is_enabled(p)]
skip_auth, token = _get_auth_context()
# Filter by auth
authorized: list[Prompt] = []
for prompt in prompts:
if not skip_auth and prompt.auth is not None:
@ -1339,8 +1305,7 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
except AuthorizationError:
continue
authorized.append(prompt)
return _dedupe_with_versions(authorized, lambda p: p.name)
return authorized
async def _get_prompt(
self, name: str, version: VersionSpec | None = None
@ -1797,19 +1762,16 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
"""
logger.debug(f"[{self.name}] Handler called: list_tools")
async with fastmcp.server.context.Context(fastmcp=self):
tools = await self.get_tools(run_middleware=True)
sdk_tools = [tool.to_mcp_tool(name=tool.name) for tool in tools]
# SDK may pass None for internal cache refresh despite type hint
cursor = (
request.params.cursor # type: ignore[union-attr]
if request is not None and request.params
else None
)
page, next_cursor = _apply_pagination(
sdk_tools, cursor, self._list_page_size
)
return mcp.types.ListToolsResult(tools=page, nextCursor=next_cursor)
tools = _dedupe_with_versions(list(await self.list_tools()), lambda t: t.name)
sdk_tools = [tool.to_mcp_tool(name=tool.name) for tool in tools]
# SDK may pass None for internal cache refresh despite type hint
cursor = (
request.params.cursor # type: ignore[union-attr]
if request is not None and request.params
else None
)
page, next_cursor = _apply_pagination(sdk_tools, cursor, self._list_page_size)
return mcp.types.ListToolsResult(tools=page, nextCursor=next_cursor)
async def _list_resources_mcp(
self, request: mcp.types.ListResourcesRequest
@ -1820,17 +1782,17 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
"""
logger.debug(f"[{self.name}] Handler called: list_resources")
async with fastmcp.server.context.Context(fastmcp=self):
resources = await self.get_resources(run_middleware=True)
sdk_resources = [
resource.to_mcp_resource(uri=str(resource.uri))
for resource in resources
]
cursor = request.params.cursor if request.params else None
page, next_cursor = _apply_pagination(
sdk_resources, cursor, self._list_page_size
)
return mcp.types.ListResourcesResult(resources=page, nextCursor=next_cursor)
resources = _dedupe_with_versions(
list(await self.list_resources()), lambda r: str(r.uri)
)
sdk_resources = [
resource.to_mcp_resource(uri=str(resource.uri)) for resource in resources
]
cursor = request.params.cursor if request.params else None
page, next_cursor = _apply_pagination(
sdk_resources, cursor, self._list_page_size
)
return mcp.types.ListResourcesResult(resources=page, nextCursor=next_cursor)
async def _list_resource_templates_mcp(
self, request: mcp.types.ListResourceTemplatesRequest
@ -1841,29 +1803,20 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
"""
logger.debug(f"[{self.name}] Handler called: list_resource_templates")
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
mw_context = MiddlewareContext(
message={},
source="client",
type="request",
method="resources/templates/list",
fastmcp_context=fastmcp_ctx,
)
templates = await self._run_middleware(
context=mw_context,
call_next=lambda context: self.get_resource_templates(),
)
sdk_templates = [
template.to_mcp_template(uriTemplate=template.uri_template)
for template in templates
]
cursor = request.params.cursor if request.params else None
page, next_cursor = _apply_pagination(
sdk_templates, cursor, self._list_page_size
)
return mcp.types.ListResourceTemplatesResult(
resourceTemplates=page, nextCursor=next_cursor
)
templates = _dedupe_with_versions(
list(await self.list_resource_templates()), lambda t: t.uri_template
)
sdk_templates = [
template.to_mcp_template(uriTemplate=template.uri_template)
for template in templates
]
cursor = request.params.cursor if request.params else None
page, next_cursor = _apply_pagination(
sdk_templates, cursor, self._list_page_size
)
return mcp.types.ListResourceTemplatesResult(
resourceTemplates=page, nextCursor=next_cursor
)
async def _list_prompts_mcp(
self, request: mcp.types.ListPromptsRequest
@ -1874,24 +1827,13 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
"""
logger.debug(f"[{self.name}] Handler called: list_prompts")
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
mw_context = MiddlewareContext(
message={},
source="client",
type="request",
method="prompts/list",
fastmcp_context=fastmcp_ctx,
)
prompts = await self._run_middleware(
context=mw_context,
call_next=lambda context: self.get_prompts(),
)
sdk_prompts = [prompt.to_mcp_prompt(name=prompt.name) for prompt in prompts]
cursor = request.params.cursor if request.params else None
page, next_cursor = _apply_pagination(
sdk_prompts, cursor, self._list_page_size
)
return mcp.types.ListPromptsResult(prompts=page, nextCursor=next_cursor)
prompts = _dedupe_with_versions(
list(await self.list_prompts()), lambda p: p.name
)
sdk_prompts = [prompt.to_mcp_prompt(name=prompt.name) for prompt in prompts]
cursor = request.params.cursor if request.params else None
page, next_cursor = _apply_pagination(sdk_prompts, cursor, self._list_page_size)
return mcp.types.ListPromptsResult(prompts=page, nextCursor=next_cursor)
async def _call_tool_mcp(
self, key: str, arguments: dict[str, Any]
@ -2807,19 +2749,19 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
return uri
# Import tools from the server
for tool in await server.get_tools():
for tool in await server.list_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():
for resource in await server.list_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():
for template in await server.list_resource_templates():
if prefix:
new_uri_template = add_resource_prefix(template.uri_template, prefix)
template = template.model_copy(
@ -2828,7 +2770,7 @@ class FastMCP(AggregateProvider, Generic[LifespanResultT]):
self.add_template(template)
# Import prompts from the server
for prompt in await server.get_prompts():
for prompt in await server.list_prompts():
if prefix:
prompt = prompt.model_copy(update={"name": f"{prefix}_{prompt.name}"})
self.add_prompt(prompt)

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

@ -106,11 +106,11 @@ async def inspect_fastmcp_v2(mcp: FastMCP[Any]) -> FastMCPInfo:
Returns:
FastMCPInfo dataclass containing the extracted information
"""
# Get all components
tools_list = await mcp.get_tools()
prompts_list = await mcp.get_prompts()
resources_list = await mcp.get_resources()
templates_list = await mcp.get_resource_templates()
# Get all components (list_* includes middleware, enabled/auth filtering)
tools_list = await mcp.list_tools()
prompts_list = await mcp.list_prompts()
resources_list = await mcp.list_resources()
templates_list = await mcp.list_resource_templates()
# Extract detailed tool information
tool_infos = []

6
tests/cli/test_run.py
View file

@ -163,7 +163,7 @@ def greet(name: str) -> str:
source = FileSystemSource(path=str(test_file))
server = await source.load_server()
assert server.name == "TestServer"
tools = await server.get_tools()
tools = await server.list_tools()
assert any(t.name == "greet" for t in tools)
async def test_import_server_with_main_block(self, tmp_path):
@ -185,7 +185,7 @@ if __name__ == "__main__":
source = FileSystemSource(path=str(test_file))
server = await source.load_server()
assert server.name == "MainServer"
tools = await server.get_tools()
tools = await server.list_tools()
assert any(t.name == "calculate" for t in tools)
async def test_import_server_standard_names(self, tmp_path):
@ -239,7 +239,7 @@ def custom_tool() -> str:
source = FileSystemSource(path=f"{test_file}:my_custom_server")
server = await source.load_server()
assert server.name == "CustomServer"
tools = await server.get_tools()
tools = await server.list_tools()
assert any(t.name == "custom_tool" for t in tools)
async def test_import_server_no_standard_names_fails(self, tmp_path):

4
tests/cli/test_server_args.py
View file

@ -49,7 +49,7 @@ def get_config() -> dict:
assert server.name == "TestServer:9000 (Debug)"
# Test the tool works and can access the parsed args
tools = await server.get_tools()
tools = await server.list_tools()
assert any(t.name == "get_config" for t in tools)
async def test_server_with_no_args(self, tmp_path):
@ -130,6 +130,6 @@ mcp = FastMCP(name)
assert server.name == "TestExample (Debug)"
# Verify tools are available
tools = await server.get_tools()
tools = await server.list_tools()
assert any(t.name == "get_status" for t in tools)
assert any(t.name == "echo_message" for t in tools)

2
tests/client/test_stdio.py
View file

@ -53,7 +53,7 @@ class TestParallelCalls:
count = 10
tasks = [proxy.get_tools() for _ in range(count)]
tasks = [proxy.list_tools() for _ in range(count)]
results = await asyncio.gather(*tasks, return_exceptions=True)

136
tests/contrib/test_component_manager.py
View file

@ -51,7 +51,7 @@ class TestComponentManagementRoutes:
"""Test enabling a tool via the HTTP route."""
# First disable the tool
mcp.disable(names={"test_tool"}, components=["tool"])
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert not any(t.name == "test_tool" for t in tools)
# Enable the tool via the HTTP route
@ -61,13 +61,13 @@ class TestComponentManagementRoutes:
assert response.json() == {"message": "Enabled tool: test_tool"}
# Verify the tool is enabled
tools = await mcp.get_tools()
tools = await mcp.list_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()
tools = await mcp.list_tools()
assert any(t.name == "test_tool" for t in tools)
# Disable the tool via the HTTP route
@ -77,14 +77,14 @@ class TestComponentManagementRoutes:
assert response.json() == {"message": "Disabled tool: test_tool"}
# Verify the tool is disabled
tools = await mcp.get_tools()
tools = await mcp.list_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 (can use URI as name for resources)
mcp.disable(names={"data://test_resource"}, components=["resource"])
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert not any(str(r.uri) == "data://test_resource" for r in resources)
# Enable the resource via the HTTP route
@ -94,13 +94,13 @@ class TestComponentManagementRoutes:
assert response.json() == {"message": "Enabled resource: data://test_resource"}
# Verify the resource is enabled
resources = await mcp.get_resources()
resources = await mcp.list_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()
resources = await mcp.list_resources()
assert any(str(r.uri) == "data://test_resource" for r in resources)
# Disable the resource via the HTTP route
@ -110,41 +110,41 @@ class TestComponentManagementRoutes:
assert response.json() == {"message": "Disabled resource: data://test_resource"}
# Verify the resource is disabled
resources = await mcp.get_resources()
resources = await mcp.list_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 template via the HTTP route."""
key = "data://test_resource/{id}"
mcp.disable(names={"data://test_resource/{id}"}, components=["template"])
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_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()
templates = await mcp.list_resource_templates()
assert any(t.uri_template == key for t in templates)
async def test_disable_template_route(self, client, mcp):
"""Test disabling a resource template via the HTTP route."""
key = "data://test_resource/{id}"
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_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()
templates = await mcp.list_resource_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(names={"test_prompt"}, components=["prompt"])
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert not any(p.name == "test_prompt" for p in prompts)
# Enable the prompt via the HTTP route
@ -154,13 +154,13 @@ class TestComponentManagementRoutes:
assert response.json() == {"message": "Enabled prompt: test_prompt"}
# Verify the prompt is enabled
prompts = await mcp.get_prompts()
prompts = await mcp.list_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()
prompts = await mcp.list_prompts()
assert any(p.name == "test_prompt" for p in prompts)
# Disable the prompt via the HTTP route
@ -170,7 +170,7 @@ class TestComponentManagementRoutes:
assert response.json() == {"message": "Disabled prompt: test_prompt"}
# Verify the prompt is disabled
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert not any(p.name == "test_prompt" for p in prompts)
@ -225,18 +225,18 @@ class TestAuthComponentManagementRoutes:
async def test_unauthorized_enable_tool(self):
"""Test that unauthenticated requests to enable a tool are rejected."""
self.mcp.disable(names={"test_tool"}, components=["tool"])
tools = await self.mcp.get_tools()
tools = await self.mcp.list_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()
tools = await self.mcp.list_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(names={"test_tool"}, components=["tool"])
tools = await self.mcp.get_tools()
tools = await self.mcp.list_tools()
assert not any(t.name == "test_tool" for t in tools)
response = self.client.post(
@ -244,22 +244,22 @@ class TestAuthComponentManagementRoutes:
)
assert response.status_code == 200
assert response.json() == {"message": "Enabled tool: test_tool"}
tools = await self.mcp.get_tools()
tools = await self.mcp.list_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()
tools = await self.mcp.list_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()
tools = await self.mcp.list_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()
tools = await self.mcp.list_tools()
assert any(t.name == "test_tool" for t in tools)
response = self.client.post(
@ -268,13 +268,13 @@ class TestAuthComponentManagementRoutes:
)
assert response.status_code == 200
assert response.json() == {"message": "Disabled tool: test_tool"}
tools = await self.mcp.get_tools()
tools = await self.mcp.list_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(names={"test_tool"}, components=["tool"])
tools = await self.mcp.get_tools()
tools = await self.mcp.list_tools()
assert not any(t.name == "test_tool" for t in tools)
response = self.client.post(
@ -282,13 +282,13 @@ class TestAuthComponentManagementRoutes:
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
tools = await self.mcp.get_tools()
tools = await self.mcp.list_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(names={"data://test_resource"}, components=["resource"])
resources = await self.mcp.get_resources()
resources = await self.mcp.list_resources()
assert not any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post(
@ -297,23 +297,23 @@ class TestAuthComponentManagementRoutes:
)
assert response.status_code == 200
assert response.json() == {"message": "Enabled resource: data://test_resource"}
resources = await self.mcp.get_resources()
resources = await self.mcp.list_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()
resources = await self.mcp.list_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()
resources = await self.mcp.list_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(names={"data://test_resource"}, components=["resource"])
resources = await self.mcp.get_resources()
resources = await self.mcp.list_resources()
assert not any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post(
@ -321,12 +321,12 @@ class TestAuthComponentManagementRoutes:
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
resources = await self.mcp.get_resources()
resources = await self.mcp.list_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()
resources = await self.mcp.list_resources()
assert any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post(
@ -335,24 +335,24 @@ class TestAuthComponentManagementRoutes:
)
assert response.status_code == 200
assert response.json() == {"message": "Disabled resource: data://test_resource"}
resources = await self.mcp.get_resources()
resources = await self.mcp.list_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(names={"test_prompt"}, components=["prompt"])
prompts = await self.mcp.get_prompts()
prompts = await self.mcp.list_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()
prompts = await self.mcp.list_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(names={"test_prompt"}, components=["prompt"])
prompts = await self.mcp.get_prompts()
prompts = await self.mcp.list_prompts()
assert not any(p.name == "test_prompt" for p in prompts)
response = self.client.post(
@ -361,22 +361,22 @@ class TestAuthComponentManagementRoutes:
)
assert response.status_code == 200
assert response.json() == {"message": "Enabled prompt: test_prompt"}
prompts = await self.mcp.get_prompts()
prompts = await self.mcp.list_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()
prompts = await self.mcp.list_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()
prompts = await self.mcp.list_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()
prompts = await self.mcp.list_prompts()
assert any(p.name == "test_prompt" for p in prompts)
response = self.client.post(
@ -384,12 +384,12 @@ class TestAuthComponentManagementRoutes:
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
prompts = await self.mcp.get_prompts()
prompts = await self.mcp.list_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()
prompts = await self.mcp.list_prompts()
assert any(p.name == "test_prompt" for p in prompts)
response = self.client.post(
@ -398,7 +398,7 @@ class TestAuthComponentManagementRoutes:
)
assert response.status_code == 200
assert response.json() == {"message": "Disabled prompt: test_prompt"}
prompts = await self.mcp.get_prompts()
prompts = await self.mcp.list_prompts()
assert not any(p.name == "test_prompt" for p in prompts)
@ -430,33 +430,33 @@ class TestComponentManagerWithPath:
async def test_enable_tool_route_with_path(self, client_with_path, mcp_with_path):
mcp_with_path.disable(names={"test_tool"}, components=["tool"])
tools = await mcp_with_path.get_tools()
tools = await mcp_with_path.list_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()
tools = await mcp_with_path.list_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()
resources = await mcp_with_path.list_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()
resources = await mcp_with_path.list_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(names={"test_prompt"}, components=["prompt"])
prompts = await mcp_with_path.get_prompts()
prompts = await mcp_with_path.list_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()
prompts = await mcp_with_path.list_prompts()
assert any(p.name == "test_prompt" for p in prompts)
@ -504,28 +504,28 @@ class TestComponentManagerWithPathAuth:
async def test_unauthorized_enable_tool(self):
self.mcp.disable(names={"test_tool"}, components=["tool"])
tools = await self.mcp.get_tools()
tools = await self.mcp.list_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()
tools = await self.mcp.list_tools()
assert not any(t.name == "test_tool" for t in tools)
async def test_forbidden_enable_tool(self):
self.mcp.disable(names={"test_tool"}, components=["tool"])
tools = await self.mcp.get_tools()
tools = await self.mcp.list_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()
tools = await self.mcp.list_tools()
assert not any(t.name == "test_tool" for t in tools)
async def test_authorized_enable_tool(self):
self.mcp.disable(names={"test_tool"}, components=["tool"])
tools = await self.mcp.get_tools()
tools = await self.mcp.list_tools()
assert not any(t.name == "test_tool" for t in tools)
response = self.client.post(
"/test/tools/test_tool/enable",
@ -533,30 +533,30 @@ class TestComponentManagerWithPathAuth:
)
assert response.status_code == 200
assert response.json() == {"message": "Enabled tool: test_tool"}
tools = await self.mcp.get_tools()
tools = await self.mcp.list_tools()
assert any(t.name == "test_tool" for t in tools)
async def test_unauthorized_disable_resource(self):
resources = await self.mcp.get_resources()
resources = await self.mcp.list_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()
resources = await self.mcp.list_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()
resources = await self.mcp.list_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()
resources = await self.mcp.list_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()
resources = await self.mcp.list_resources()
assert any(str(r.uri) == "data://test_resource" for r in resources)
response = self.client.post(
"/test/resources/data://test_resource/disable",
@ -564,33 +564,33 @@ class TestComponentManagerWithPathAuth:
)
assert response.status_code == 200
assert response.json() == {"message": "Disabled resource: data://test_resource"}
resources = await self.mcp.get_resources()
resources = await self.mcp.list_resources()
assert not any(str(r.uri) == "data://test_resource" for r in resources)
async def test_unauthorized_enable_prompt(self):
self.mcp.disable(names={"test_prompt"}, components=["prompt"])
prompts = await self.mcp.get_prompts()
prompts = await self.mcp.list_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()
prompts = await self.mcp.list_prompts()
assert not any(p.name == "test_prompt" for p in prompts)
async def test_forbidden_enable_prompt(self):
self.mcp.disable(names={"test_prompt"}, components=["prompt"])
prompts = await self.mcp.get_prompts()
prompts = await self.mcp.list_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()
prompts = await self.mcp.list_prompts()
assert not any(p.name == "test_prompt" for p in prompts)
async def test_authorized_enable_prompt(self):
self.mcp.disable(names={"test_prompt"}, components=["prompt"])
prompts = await self.mcp.get_prompts()
prompts = await self.mcp.list_prompts()
assert not any(p.name == "test_prompt" for p in prompts)
response = self.client.post(
"/test/prompts/test_prompt/enable",
@ -598,5 +598,5 @@ class TestComponentManagerWithPathAuth:
)
assert response.status_code == 200
assert response.json() == {"message": "Enabled prompt: test_prompt"}
prompts = await self.mcp.get_prompts()
prompts = await self.mcp.list_prompts()
assert any(p.name == "test_prompt" for p in prompts)

30
tests/contrib/test_mcp_mixin.py
View file

@ -67,7 +67,7 @@ class TestMCPMixin:
instance = MyToolMixin()
instance.register_tools(mcp, prefix=prefix, separator=separator)
registered_tools = await mcp.get_tools()
registered_tools = await mcp.list_tools()
assert any(t.name == expected_key for t in registered_tools)
assert not any(t.name == unexpected_key for t in registered_tools)
@ -112,7 +112,7 @@ class TestMCPMixin:
instance = MyResourceMixin()
instance.register_resources(mcp, prefix=prefix, separator=separator)
registered_resources = await mcp.get_resources()
registered_resources = await mcp.list_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
@ -158,7 +158,7 @@ class TestMCPMixin:
instance = MyPromptMixin()
instance.register_prompts(mcp, prefix=prefix, separator=separator)
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert any(p.name == expected_name for p in prompts)
assert not any(p.name == unexpected_name for p in prompts)
@ -182,9 +182,9 @@ class TestMCPMixin:
instance = MyFullMixin()
instance.register_all(mcp)
tools = await mcp.get_tools()
resources = await mcp.get_resources()
prompts = await mcp.get_prompts()
tools = await mcp.list_tools()
resources = await mcp.list_resources()
prompts = await mcp.list_prompts()
assert any(t.name == "tool_all" for t in tools)
assert any(str(r.uri) == "res://all" for r in resources)
@ -210,9 +210,9 @@ class TestMCPMixin:
instance = MyFullMixinPrefixed()
instance.register_all(mcp, prefix="all")
tools = await mcp.get_tools()
resources = await mcp.get_resources()
prompts = await mcp.get_prompts()
tools = await mcp.list_tools()
resources = await mcp.list_resources()
prompts = await mcp.list_prompts()
assert any(t.name == f"all{_DEFAULT_SEPARATOR_TOOL}tool_all_p" for t in tools)
assert any(
@ -249,9 +249,9 @@ class TestMCPMixin:
prompt_separator=".",
)
tools = await mcp.get_tools()
resources = await mcp.get_resources()
prompts = await mcp.get_prompts()
tools = await mcp.list_tools()
resources = await mcp.list_resources()
prompts = await mcp.list_prompts()
assert any(t.name == "cust-tool_cust" for t in tools)
assert any(str(r.uri) == "cust::res://cust" for r in resources)
@ -286,7 +286,7 @@ class TestMCPMixin:
instance = MyToolWithMeta()
instance.register_tools(mcp)
registered_tools = await mcp.get_tools()
registered_tools = await mcp.list_tools()
tool = next(t for t in registered_tools if t.name == "sample_tool")
assert tool.annotations is not None
@ -309,7 +309,7 @@ class TestMCPMixin:
instance = MyResourceWithMeta()
instance.register_resources(mcp)
registered_resources = await mcp.get_resources()
registered_resources = await mcp.list_resources()
resource = next(
r for r in registered_resources if str(r.uri) == "test://resource"
)
@ -332,7 +332,7 @@ class TestMCPMixin:
instance = MyPromptWithMeta()
instance.register_prompts(mcp)
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
prompt = next(p for p in prompts if p.name == "sample_prompt")
assert prompt.title == "My Prompt Title"

4
tests/deprecated/test_exclude_args.py
View file

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

36
tests/deprecated/test_import_server.py
View file

@ -25,8 +25,8 @@ async def test_import_basic_functionality():
await main_app.import_server(sub_app, "sub")
# Verify the tool was imported with the prefix
main_tools = await main_app.get_tools()
sub_tools = await sub_app.get_tools()
main_tools = await main_app.list_tools()
sub_tools = await sub_app.list_tools()
assert any(t.name == "sub_sub_tool" for t in main_tools)
assert any(t.name == "sub_tool" for t in sub_tools)
@ -60,7 +60,7 @@ async def test_import_multiple_apps():
await main_app.import_server(news_app, "news")
# Verify tools were imported with the correct prefixes
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert any(t.name == "weather_get_forecast" for t in tools)
assert any(t.name == "news_get_headlines" for t in tools)
@ -83,14 +83,14 @@ async def test_import_combines_tools():
# Import first app
await main_app.import_server(first_app, "api")
tools = await main_app.get_tools()
tools = await main_app.list_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()
tools = await main_app.list_tools()
assert any(t.name == "api_second_tool" for t in tools)
# Tools from both imports are combined
@ -112,7 +112,7 @@ async def test_import_with_resources():
await main_app.import_server(data_app, "data")
# Verify the resource was imported with the prefix
resources = await main_app.get_resources()
resources = await main_app.list_resources()
assert any(str(r.uri) == "data://data/users" for r in resources)
@ -135,7 +135,7 @@ async def test_import_with_resource_templates():
await main_app.import_server(user_app, "api")
# Verify the template was imported with the prefix
templates = await main_app.get_resource_templates()
templates = await main_app.list_resource_templates()
assert any(t.uri_template == "users://api/{user_id}/profile" for t in templates)
@ -154,7 +154,7 @@ async def test_import_with_prompts():
await main_app.import_server(assistant_app, "assistant")
# Verify the prompt was imported with the prefix
prompts = await main_app.get_prompts()
prompts = await main_app.list_prompts()
assert any(p.name == "assistant_greeting" for p in prompts)
@ -179,7 +179,7 @@ async def test_import_multiple_resource_templates():
await main_app.import_server(news_app, "content")
# Verify templates were imported with correct prefixes
templates = await main_app.get_resource_templates()
templates = await main_app.list_resource_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)
@ -205,7 +205,7 @@ async def test_import_multiple_prompts():
await main_app.import_server(sql_app, "sql")
# Verify prompts were imported with correct prefixes
prompts = await main_app.get_prompts()
prompts = await main_app.list_prompts()
assert any(p.name == "python_review_python" for p in prompts)
assert any(p.name == "sql_explain_sql" for p in prompts)
@ -443,10 +443,10 @@ async def test_import_with_no_prefix():
await main_app.import_server(sub_app)
# Verify all component types are accessible with original names
tools = await main_app.get_tools()
resources = await main_app.get_resources()
templates = await main_app.get_resource_templates()
prompts = await main_app.get_prompts()
tools = await main_app.list_tools()
resources = await main_app.list_resources()
templates = await main_app.list_resource_templates()
prompts = await main_app.list_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)
@ -640,7 +640,7 @@ async def test_import_server_resource_uri_prefixing():
await main_server.import_server(sub_server, prefix="imported")
# Get resources and verify URI prefixing (name should NOT be prefixed)
resources = await main_server.get_resources()
resources = await main_server.list_resources()
resource = next(
r for r in resources if str(r.uri) == "resource://imported/test_resource"
)
@ -661,7 +661,7 @@ async def test_import_server_resource_template_uri_prefixing():
await main_server.import_server(sub_server, prefix="imported")
# Get resource templates and verify URI prefixing (name should NOT be prefixed)
templates = await main_server.get_resource_templates()
templates = await main_server.list_resource_templates()
template = next(
t for t in templates if t.uri_template == "resource://imported/data/{item_id}"
)
@ -690,8 +690,8 @@ async def test_import_server_with_new_prefix_format():
await target_server.import_server(source_server, "imported")
# Check that the resources were imported with the correct prefixes
resources = await target_server.get_resources()
templates = await target_server.get_resource_templates()
resources = await target_server.list_resources()
templates = await target_server.list_resource_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)

2
tests/server/auth/providers/test_introspection.py
View file

@ -547,5 +547,5 @@ class TestIntrospectionTokenVerifierIntegration:
# Verify the auth is set correctly
assert mcp.auth is verifier
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(list(tools)) == 1

14
tests/server/auth/test_authorization.py
View file

@ -237,7 +237,7 @@ class TestToolLevelAuth:
def public_tool() -> str:
return "public"
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].name == "public_tool"
@ -249,7 +249,7 @@ class TestToolLevelAuth:
return "protected"
# No token set - tool should be hidden
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 0
async def test_tool_with_auth_visible_with_token(self):
@ -263,7 +263,7 @@ class TestToolLevelAuth:
token = make_token()
tok = set_token(token)
try:
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].name == "protected_tool"
finally:
@ -280,7 +280,7 @@ class TestToolLevelAuth:
token = make_token(scopes=["read"])
tok = set_token(token)
try:
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 0
finally:
auth_context_var.reset(tok)
@ -296,7 +296,7 @@ class TestToolLevelAuth:
token = make_token(scopes=["admin"])
tok = set_token(token)
try:
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].name == "admin_tool"
finally:
@ -517,7 +517,7 @@ class TestTransformedToolAuth:
)
# Without token, transformed tool should not be visible
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 0
async def test_transformed_tool_visible_with_token(self):
@ -539,7 +539,7 @@ class TestTransformedToolAuth:
token = make_token()
tok = set_token(token)
try:
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].name == "renamed_protected"
finally:

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

@ -61,7 +61,7 @@ class TestOpenAPIPerformance:
)
# Verify server and tools were created successfully
tools = await mcp_server.get_tools()
tools = await mcp_server.list_tools()
assert len(tools) > 500
def test_medium_schema_performance(self):

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

@ -233,7 +233,7 @@ class TestPerformance:
)
# Get tools from the server via public API
tools = await server.get_tools()
tools = await server.list_tools()
# Should have 6 operations in the spec
assert len(tools) == 6

38
tests/server/providers/proxy/test_proxy_server.py
View file

@ -225,14 +225,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()
tools = await proxy_server.list_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()
tools = await proxy_server.list_tools()
greet_tool = next(t for t in tools if t.name == "greet")
assert greet_tool.title == "Greet"
assert greet_tool.meta == {"fastmcp": {"tags": ["greet"]}}
@ -255,7 +255,7 @@ class TestTools:
)
proxy = create_proxy(server)
tools = await proxy.get_tools()
tools = await proxy.list_tools()
assert any(t.name == "add_transformed" for t in tools)
assert not any(t.name == "add" for t in tools)
@ -281,7 +281,7 @@ class TestTools:
assert result.data == 3
async def test_tool_without_description(self, proxy_server):
tools = await proxy_server.get_tools()
tools = await proxy_server.list_tools()
tool = next(t for t in tools if t.name == "tool_without_description")
assert tool.description is None
@ -356,7 +356,7 @@ class TestTools:
class TestResources:
async def test_get_resources(self, proxy_server):
resources = await proxy_server.get_resources()
resources = await proxy_server.list_resources()
assert [r.uri for r in resources] == Contains(
AnyUrl("data://users"),
AnyUrl("resource://wave"),
@ -364,7 +364,7 @@ class TestResources:
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()
resources = await proxy_server.list_resources()
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"]}}
@ -471,11 +471,11 @@ class TestResources:
class TestResourceTemplates:
async def test_get_resource_templates(self, proxy_server):
templates = await proxy_server.get_resource_templates()
templates = await proxy_server.list_resource_templates()
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()
templates = await proxy_server.list_resource_templates()
get_user_template = next(
t for t in templates if t.uri_template == "data://user/{user_id}"
)
@ -585,11 +585,11 @@ class TestResourceTemplates:
class TestPrompts:
async def test_get_prompts_server_method(self, proxy_server: FastMCPProxy):
prompts = await proxy_server.get_prompts()
prompts = await proxy_server.list_prompts()
assert [p.name for p in prompts] == Contains("welcome")
async def test_get_prompts_meta(self, proxy_server):
prompts = await proxy_server.get_prompts()
prompts = await proxy_server.list_prompts()
welcome_prompt = next(p for p in prompts if p.name == "welcome")
assert welcome_prompt.title == "Welcome"
assert welcome_prompt.meta == {"fastmcp": {"tags": ["welcome"]}}
@ -666,9 +666,9 @@ async def test_proxy_handles_multiple_concurrent_tasks_correctly(
results[name] = await coro()
async with create_task_group() as tg:
tg.start_soon(get_and_store, "prompts", proxy_server.get_prompts)
tg.start_soon(get_and_store, "resources", proxy_server.get_resources)
tg.start_soon(get_and_store, "tools", proxy_server.get_tools)
tg.start_soon(get_and_store, "prompts", proxy_server.list_prompts)
tg.start_soon(get_and_store, "resources", proxy_server.list_resources)
tg.start_soon(get_and_store, "tools", proxy_server.list_tools)
assert list(results) == Contains("resources", "prompts", "tools")
assert [p.name for p in results["prompts"]] == Contains("welcome")
@ -687,7 +687,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()
tools = await proxy_server.list_tools()
tool = next(t for t in tools if t.name == "greet")
with pytest.raises(NotImplementedError, match="server.enable"):
@ -695,7 +695,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()
tools = await proxy_server.list_tools()
tool = next(t for t in tools if t.name == "greet")
with pytest.raises(NotImplementedError, match="server.disable"):
@ -703,7 +703,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()
resources = await proxy_server.list_resources()
resource = next(r for r in resources if str(r.uri) == "resource://wave")
with pytest.raises(NotImplementedError, match="server.enable"):
@ -711,7 +711,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()
resources = await proxy_server.list_resources()
resource = next(r for r in resources if str(r.uri) == "resource://wave")
with pytest.raises(NotImplementedError, match="server.disable"):
@ -719,7 +719,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()
prompts = await proxy_server.list_prompts()
prompt = next(p for p in prompts if p.name == "welcome")
with pytest.raises(NotImplementedError, match="server.enable"):
@ -727,7 +727,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()
prompts = await proxy_server.list_prompts()
prompt = next(p for p in prompts if p.name == "welcome")
with pytest.raises(NotImplementedError, match="server.disable"):

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

@ -366,7 +366,7 @@ class TestLocalProviderDecorators:
# Filtering happens at the server level, not provider level
server = FastMCP("Test", providers=[provider])
tools = await server.get_tools()
tools = await server.list_tools()
names = {t.name for t in tools}
assert "enabled_tool" in names
assert "disabled_tool" not in names
@ -455,7 +455,7 @@ class TestLocalProviderDecorators:
# Filtering happens at the server level, not provider level
server = FastMCP("Test", providers=[provider])
resources = await server.get_resources()
resources = await server.list_resources()
uris = {str(r.uri) for r in resources}
assert "resource://enabled" in uris
assert "resource://disabled" not in uris
@ -491,7 +491,7 @@ class TestLocalProviderDecorators:
# Filtering happens at the server level, not provider level
server = FastMCP("Test", providers=[provider])
templates = await server.get_resource_templates()
templates = await server.list_resource_templates()
uris = {t.uri_template for t in templates}
assert "items://{id}" in uris
assert "data://{id}" not in uris
@ -562,7 +562,7 @@ class TestLocalProviderDecorators:
# Filtering happens at the server level, not provider level
server = FastMCP("Test", providers=[provider])
prompts = await server.get_prompts()
prompts = await server.list_prompts()
names = {p.name for p in prompts}
assert "enabled_prompt" in names
assert "disabled_prompt" not in names
@ -778,7 +778,7 @@ class TestLocalProviderStandaloneUsage:
assert any(t.name == "shared_tool" for t in tools2)
async def test_tools_visible_via_server_get_tools(self):
"""Test that provider tools are visible via server.get_tools()."""
"""Test that provider tools are visible via server.list_tools()."""
provider = LocalProvider()
@provider.tool
@ -787,7 +787,7 @@ class TestLocalProviderStandaloneUsage:
server = FastMCP("Test", providers=[provider])
tools = await server.get_tools()
tools = await server.list_tools()
assert any(t.name == "provider_tool" for t in tools)
async def test_server_decorator_and_provider_tools_coexist(self):
@ -804,7 +804,7 @@ class TestLocalProviderStandaloneUsage:
def server_tool() -> str:
return "from server"
tools = await server.get_tools()
tools = await server.list_tools()
assert any(t.name == "provider_tool" for t in tools)
assert any(t.name == "server_tool" for t in tools)
@ -823,7 +823,7 @@ class TestLocalProviderStandaloneUsage:
return "from server"
# Server's LocalProvider is first, so its tool wins
tools = await server.get_tools()
tools = await server.list_tools()
assert any(t.name == "duplicate_tool" for t in tools)
async with Client(server) as client:

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

@ -53,7 +53,7 @@ class TestPromptDecorator:
def fn() -> str:
return "Hello, world!"
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "fn")
assert prompt.name == "fn"
@ -68,7 +68,7 @@ class TestPromptDecorator:
def fn() -> str:
return "Hello, world!"
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert any(p.name == "fn" for p in prompts)
result = await mcp.render_prompt("fn")
@ -83,7 +83,7 @@ class TestPromptDecorator:
def fn() -> str:
return "Hello, world!"
prompts_list = await mcp.get_prompts()
prompts_list = await mcp.list_prompts()
assert len(prompts_list) == 1
prompt = next(p for p in prompts_list if p.name == "custom_name")
assert prompt.name == "custom_name"
@ -98,7 +98,7 @@ class TestPromptDecorator:
def fn() -> str:
return "Hello, world!"
prompts_list = await mcp.get_prompts()
prompts_list = await mcp.list_prompts()
assert len(prompts_list) == 1
prompt = next(p for p in prompts_list if p.name == "fn")
assert prompt.description == "A custom description"
@ -113,7 +113,7 @@ class TestPromptDecorator:
def test_prompt(name: str, greeting: str = "Hello") -> str:
return f"{greeting}, {name}!"
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "test_prompt")
assert prompt.arguments is not None
@ -221,7 +221,7 @@ class TestPromptDecorator:
def sample_prompt() -> str:
return "Hello, world!"
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert len(prompts) == 1
prompt = next(p for p in prompts if p.name == "sample_prompt")
assert prompt.tags == {"example", "test-tag"}
@ -235,7 +235,7 @@ class TestPromptDecorator:
"""A function with a string name."""
return "Hello from string named prompt!"
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert any(p.name == "string_named_prompt" for p in prompts)
assert not any(p.name == "my_function" for p in prompts)
@ -264,7 +264,7 @@ class TestPromptDecorator:
assert decorated.__fastmcp__.name == "direct_call_prompt"
assert result_fn is standalone_function
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
prompt = next(p for p in prompts if p.name == "direct_call_prompt")
# Prompt is registered separately, not same object as decorated function
assert prompt.name == "direct_call_prompt"
@ -313,7 +313,7 @@ class TestPromptDecorator:
def test_prompt(message: str) -> str:
return f"Response: {message}"
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
prompt = next(p for p in prompts if p.name == "test_prompt")
assert prompt.meta == meta_data
@ -327,17 +327,17 @@ class TestPromptEnabled:
def sample_prompt() -> str:
return "Hello, world!"
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert any(p.name == "sample_prompt" for p in prompts)
mcp.disable(names={"sample_prompt"}, components=["prompt"])
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert not any(p.name == "sample_prompt" for p in prompts)
mcp.enable(names={"sample_prompt"}, components=["prompt"])
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert any(p.name == "sample_prompt" for p in prompts)
async def test_prompt_disabled(self):
@ -348,7 +348,7 @@ class TestPromptEnabled:
return "Hello, world!"
mcp.disable(names={"sample_prompt"}, components=["prompt"])
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert len(prompts) == 0
async def test_prompt_toggle_enabled(self):
@ -359,11 +359,11 @@ class TestPromptEnabled:
return "Hello, world!"
mcp.disable(names={"sample_prompt"}, components=["prompt"])
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert not any(p.name == "sample_prompt" for p in prompts)
mcp.enable(names={"sample_prompt"}, components=["prompt"])
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert len(prompts) == 1
async def test_prompt_toggle_disabled(self):
@ -374,7 +374,7 @@ class TestPromptEnabled:
return "Hello, world!"
mcp.disable(names={"sample_prompt"}, components=["prompt"])
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert len(prompts) == 0
# get_prompt() applies enabled transform, returns None for disabled
@ -392,7 +392,7 @@ class TestPromptEnabled:
assert prompt is not None
mcp.disable(names={"sample_prompt"}, components=["prompt"])
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert len(prompts) == 0
# get_prompt() applies enabled transform, returns None for disabled
@ -429,27 +429,27 @@ class TestPromptTags:
async def test_include_tags_all_prompts(self):
mcp = self.create_server(include_tags={"a", "b"})
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert {p.name for p in prompts} == {"prompt_1", "prompt_2"}
async def test_include_tags_some_prompts(self):
mcp = self.create_server(include_tags={"a"})
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert {p.name for p in prompts} == {"prompt_1"}
async def test_exclude_tags_all_prompts(self):
mcp = self.create_server(exclude_tags={"a", "b"})
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert {p.name for p in prompts} == set()
async def test_exclude_tags_some_prompts(self):
mcp = self.create_server(exclude_tags={"a"})
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert {p.name for p in prompts} == {"prompt_2"}
async def test_exclude_takes_precedence_over_include(self):
mcp = self.create_server(exclude_tags={"a"}, include_tags={"b"})
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert {p.name for p in prompts} == {"prompt_2"}
async def test_read_prompt_includes_tags(self):

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

@ -140,7 +140,7 @@ class TestResourceTemplates:
def add(x: int, y: int = 10) -> str:
return str(int(x) + y)
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert len(templates) == 1
assert templates[0].uri_template == "math://add/{x}"
@ -158,7 +158,7 @@ class TestResourceTemplates:
def get_data(name: str) -> str:
return f"Data for {name}"
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert len(templates) == 1
assert templates[0].uri_template == "resource://{name}/data"
@ -172,7 +172,7 @@ class TestResourceTemplates:
def template_resource(param: str) -> str:
return f"Template resource: {param}"
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
template = next(t for t in templates if t.uri_template == "resource://{param}")
assert template.tags == {"template", "test-tag"}
@ -250,7 +250,7 @@ class TestResourceTemplates:
def get_user(user_id: str) -> str:
return f"User {user_id} data"
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert len(templates) == 1
template = templates[0]
@ -330,7 +330,7 @@ class TestResourceDecorator:
def get_data() -> str:
return "Hello, world!"
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert len(resources) == 1
assert resources[0].name == "custom-data"
@ -344,7 +344,7 @@ class TestResourceDecorator:
def get_data() -> str:
return "Hello, world!"
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert len(resources) == 1
assert resources[0].description == "Data resource"
@ -356,7 +356,7 @@ class TestResourceDecorator:
def get_data() -> str:
return "Hello, world!"
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert len(resources) == 1
assert resources[0].tags == {"example", "test-tag"}
@ -456,7 +456,7 @@ class TestResourceDecorator:
def get_data() -> str:
return "Hello, world!"
resources = await mcp.get_resources()
resources = await mcp.list_resources()
resource = next(r for r in resources if str(r.uri) == "resource://data")
assert resource.meta == meta_data
@ -525,7 +525,7 @@ class TestTemplateDecorator:
def get_data(name: str) -> str:
return f"Data for {name}"
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert len(templates) == 1
assert templates[0].name == "get_data"
assert templates[0].uri_template == "resource://{name}/data"
@ -551,7 +551,7 @@ class TestTemplateDecorator:
def get_data(name: str) -> str:
return f"Data for {name}"
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert len(templates) == 1
assert templates[0].name == "custom-template"
@ -565,7 +565,7 @@ class TestTemplateDecorator:
def get_data(name: str) -> str:
return f"Data for {name}"
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert len(templates) == 1
assert templates[0].description == "Template description"
@ -640,7 +640,7 @@ class TestTemplateDecorator:
def template_resource(param: str) -> str:
return f"Template resource: {param}"
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
template = next(t for t in templates if t.uri_template == "resource://{param}")
assert template.tags == {"template", "test-tag"}
@ -651,7 +651,7 @@ class TestTemplateDecorator:
def template_resource(param: str) -> str:
return f"Template resource: {param}"
templates = await mcp.get_resource_templates()
templates = await mcp.list_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"
@ -666,7 +666,7 @@ class TestTemplateDecorator:
def get_template_data(param: str) -> str:
return f"Data for {param}"
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
template = next(
t for t in templates if t.uri_template == "resource://{param}/data"
)
@ -690,27 +690,27 @@ class TestResourceTags:
async def test_include_tags_all_resources(self):
mcp = self.create_server(include_tags={"a", "b"})
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert {r.name for r in resources} == {"resource_1", "resource_2"}
async def test_include_tags_some_resources(self):
mcp = self.create_server(include_tags={"a", "z"})
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert {r.name for r in resources} == {"resource_1"}
async def test_exclude_tags_all_resources(self):
mcp = self.create_server(exclude_tags={"a", "b"})
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert {r.name for r in resources} == set()
async def test_exclude_tags_some_resources(self):
mcp = self.create_server(exclude_tags={"a"})
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert {r.name for r in resources} == {"resource_2"}
async def test_exclude_precedence(self):
mcp = self.create_server(exclude_tags={"a"}, include_tags={"b"})
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert {r.name for r in resources} == {"resource_2"}
async def test_read_included_resource(self):
@ -735,17 +735,17 @@ class TestResourceEnabled:
def sample_resource() -> str:
return "Hello, world!"
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert any(str(r.uri) == "resource://data" for r in resources)
mcp.disable(names={"resource://data"}, components=["resource"])
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert not any(str(r.uri) == "resource://data" for r in resources)
mcp.enable(names={"resource://data"}, components=["resource"])
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert any(str(r.uri) == "resource://data" for r in resources)
async def test_resource_disabled(self):
@ -756,7 +756,7 @@ class TestResourceEnabled:
return "Hello, world!"
mcp.disable(names={"resource://data"}, components=["resource"])
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert len(resources) == 0
with pytest.raises(NotFoundError, match="Unknown resource"):
@ -770,11 +770,11 @@ class TestResourceEnabled:
return "Hello, world!"
mcp.disable(names={"resource://data"}, components=["resource"])
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert not any(str(r.uri) == "resource://data" for r in resources)
mcp.enable(names={"resource://data"}, components=["resource"])
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert len(resources) == 1
async def test_resource_toggle_disabled(self):
@ -785,7 +785,7 @@ class TestResourceEnabled:
return "Hello, world!"
mcp.disable(names={"resource://data"}, components=["resource"])
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert len(resources) == 0
with pytest.raises(NotFoundError, match="Unknown resource"):
@ -802,7 +802,7 @@ class TestResourceEnabled:
assert resource is not None
mcp.disable(names={"resource://data"}, components=["resource"])
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert len(resources) == 0
with pytest.raises(NotFoundError, match="Unknown resource"):
@ -837,7 +837,7 @@ class TestResourceTemplatesTags:
async def test_include_tags_all_resources(self):
mcp = self.create_server(include_tags={"a", "b"})
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert {t.name for t in templates} == {
"template_resource_1",
"template_resource_2",
@ -845,22 +845,22 @@ class TestResourceTemplatesTags:
async def test_include_tags_some_resources(self):
mcp = self.create_server(include_tags={"a"})
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert {t.name for t in templates} == {"template_resource_1"}
async def test_exclude_tags_all_resources(self):
mcp = self.create_server(exclude_tags={"a", "b"})
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert {t.name for t in templates} == set()
async def test_exclude_tags_some_resources(self):
mcp = self.create_server(exclude_tags={"a"})
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert {t.name for t in templates} == {"template_resource_2"}
async def test_exclude_takes_precedence_over_include(self):
mcp = self.create_server(exclude_tags={"a"}, include_tags={"b"})
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert {t.name for t in templates} == {"template_resource_2"}
async def test_read_resource_template_includes_tags(self):
@ -888,17 +888,17 @@ class TestResourceTemplateEnabled:
def sample_template(param: str) -> str:
return f"Template: {param}"
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert any(t.uri_template == "resource://{param}" for t in templates)
mcp.disable(names={"resource://{param}"}, components=["template"])
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert not any(t.uri_template == "resource://{param}" for t in templates)
mcp.enable(names={"resource://{param}"}, components=["template"])
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert any(t.uri_template == "resource://{param}" for t in templates)
async def test_template_disabled(self):
@ -909,7 +909,7 @@ class TestResourceTemplateEnabled:
return f"Template: {param}"
mcp.disable(names={"resource://{param}"}, components=["template"])
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert len(templates) == 0
with pytest.raises(NotFoundError, match="Unknown resource"):
@ -923,11 +923,11 @@ class TestResourceTemplateEnabled:
return f"Template: {param}"
mcp.disable(names={"resource://{param}"}, components=["template"])
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert not any(t.uri_template == "resource://{param}" for t in templates)
mcp.enable(names={"resource://{param}"}, components=["template"])
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert len(templates) == 1
async def test_template_toggle_disabled(self):
@ -938,7 +938,7 @@ class TestResourceTemplateEnabled:
return f"Template: {param}"
mcp.disable(names={"resource://{param}"}, components=["template"])
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert len(templates) == 0
with pytest.raises(NotFoundError, match="Unknown resource"):
@ -955,7 +955,7 @@ class TestResourceTemplateEnabled:
assert template is not None
mcp.disable(names={"resource://{param}"}, components=["template"])
templates = await mcp.get_resource_templates()
templates = await mcp.list_resource_templates()
assert len(templates) == 0
with pytest.raises(NotFoundError, match="Unknown resource"):

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

@ -379,7 +379,7 @@ class TestToolParameters:
"""A greeting tool"""
return f"Hello {title} {name}"
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
tool = tools[0]
@ -402,7 +402,7 @@ class TestToolParameters:
"""A greeting tool"""
return f"Hello {title} {name}"
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
tool = tools[0]
@ -731,7 +731,7 @@ class TestToolParameters:
def f(x: Annotated[int, "A number"]):
return x
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].parameters["properties"]["x"]["description"] == "A number"
@ -745,7 +745,7 @@ class TestToolOutputSchema:
def f() -> annotation:
return "hello"
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
type_schema = TypeAdapter(annotation).json_schema()
@ -768,7 +768,7 @@ class TestToolOutputSchema:
def f() -> annotation:
return {"name": "John", "age": 30}
tools = await mcp.get_tools()
tools = await mcp.list_tools()
type_schema = compress_schema(
TypeAdapter(annotation).json_schema(), prune_titles=True
@ -826,7 +826,7 @@ class TestToolOutputSchema:
def simple_tool() -> int:
return 42
tools = await mcp.get_tools()
tools = await mcp.list_tools()
tool = next(t for t in tools if t.name == "simple_tool")
assert tool.output_schema is None
@ -853,7 +853,7 @@ class TestToolOutputSchema:
def explicit_tool() -> dict[str, Any]:
return {"greeting": "Hello", "count": 42}
tools = await mcp.get_tools()
tools = await mcp.list_tools()
tool = next(t for t in tools if t.name == "explicit_tool")
expected_schema = {
"type": "object",
@ -876,7 +876,7 @@ class TestToolOutputSchema:
def primitive_tool() -> str:
return "Hello, primitives!"
tools = await mcp.get_tools()
tools = await mcp.list_tools()
tool = next(t for t in tools if t.name == "primitive_tool")
expected_schema = {
"type": "object",
@ -897,7 +897,7 @@ class TestToolOutputSchema:
def complex_tool() -> list[dict[str, int]]:
return [{"a": 1, "b": 2}, {"c": 3, "d": 4}]
tools = await mcp.get_tools()
tools = await mcp.list_tools()
tool = next(t for t in tools if t.name == "complex_tool")
expected_inner_schema = compress_schema(
TypeAdapter(list[dict[str, int]]).json_schema(), prune_titles=True
@ -927,7 +927,7 @@ class TestToolOutputSchema:
def dataclass_tool() -> User:
return User(name="Alice", age=30)
tools = await mcp.get_tools()
tools = await mcp.list_tools()
tool = next(t for t in tools if t.name == "dataclass_tool")
expected_schema = compress_schema(
TypeAdapter(User).json_schema(), prune_titles=True
@ -968,7 +968,7 @@ class TestToolOutputSchema:
def edge_case_tool() -> tuple[int, str]:
return (42, "hello")
tools = await mcp.get_tools()
tools = await mcp.list_tools()
tool = next(t for t in tools if t.name == "edge_case_tool")
assert tool.output_schema and "x-fastmcp-wrap-result" in tool.output_schema
@ -988,7 +988,7 @@ class TestToolContextInjection:
def tool_with_context(x: int, ctx: Context) -> str:
return f"Request: {x}"
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].name == "tool_with_context"
# Context param should not appear in schema
@ -1057,7 +1057,7 @@ class TestToolContextInjection:
def sample_tool(x: int) -> int:
return x * 2
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].tags == {"example", "test-tag"}
@ -1093,7 +1093,7 @@ class TestToolContextInjection:
assert isinstance(ctx, Context)
return f"query: {query}"
tools = await mcp.get_tools()
tools = await mcp.list_tools()
tool = next(t for t in tools if t.name == "decorated_tool")
assert "ctx" not in tool.parameters.get("properties", {})
@ -1128,7 +1128,7 @@ class TestToolDecorator:
def add(x: int, y: int) -> int:
return x + y
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert any(t.name == "add" for t in tools)
result = await mcp.call_tool("add", {"x": 1, "y": 2})
@ -1151,7 +1151,7 @@ class TestToolDecorator:
def add(x: int, y: int) -> int:
return x + y
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
tool = tools[0]
assert tool.description == "Add two numbers"
@ -1265,7 +1265,7 @@ class TestToolDecorator:
def sample_tool(x: int) -> int:
return x * 2
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].tags == {"example", "test-tag"}
@ -1279,7 +1279,7 @@ class TestToolDecorator:
mcp.add_tool(Tool.from_function(multiply, name="custom_multiply"))
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert any(t.name == "custom_multiply" for t in tools)
result = await mcp.call_tool("custom_multiply", {"a": 5, "b": 3})
@ -1298,7 +1298,7 @@ class TestToolDecorator:
) -> None:
pass
tools = await mcp.get_tools()
tools = await mcp.list_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"
@ -1314,7 +1314,7 @@ class TestToolDecorator:
) -> None:
pass
tools = await mcp.get_tools()
tools = await mcp.list_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"
@ -1339,7 +1339,7 @@ class TestToolDecorator:
assert decorated.__fastmcp__.name == "direct_call_tool"
assert result_fn is standalone_function
tools = await mcp.get_tools()
tools = await mcp.list_tools()
tool = next(t for t in tools if t.name == "direct_call_tool")
# Tool is registered separately, not same object as decorated function
assert tool.name == "direct_call_tool"
@ -1356,7 +1356,7 @@ class TestToolDecorator:
"""A function with a string name."""
return f"Result: {x}"
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert any(t.name == "string_named_tool" for t in tools)
assert not any(t.name == "my_function" for t in tools)
@ -1398,7 +1398,7 @@ class TestToolDecorator:
"""Multiply two numbers."""
return a * b
tools = await mcp.get_tools()
tools = await mcp.list_tools()
tool = next(t for t in tools if t.name == "multiply")
assert tool.meta == meta_data
@ -1420,27 +1420,27 @@ class TestToolTags:
async def test_include_tags_all_tools(self):
mcp = self.create_server(include_tags={"a", "b"})
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert {t.name for t in tools} == {"tool_1", "tool_2"}
async def test_include_tags_some_tools(self):
mcp = self.create_server(include_tags={"a", "z"})
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert {t.name for t in tools} == {"tool_1"}
async def test_exclude_tags_all_tools(self):
mcp = self.create_server(exclude_tags={"a", "b"})
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert {t.name for t in tools} == set()
async def test_exclude_tags_some_tools(self):
mcp = self.create_server(exclude_tags={"a", "z"})
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert {t.name for t in tools} == {"tool_2"}
async def test_exclude_precedence(self):
mcp = self.create_server(exclude_tags={"a"}, include_tags={"b"})
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert {t.name for t in tools} == {"tool_2"}
async def test_call_included_tool(self):
@ -1469,19 +1469,19 @@ class TestToolEnabled:
return x * 2
# Tool is enabled by default
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert any(t.name == "sample_tool" for t in tools)
# Disable via server
mcp.disable(names={"sample_tool"}, components=["tool"])
# Tool should not be in list when disabled
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert not any(t.name == "sample_tool" for t in tools)
# Re-enable via server
mcp.enable(names={"sample_tool"}, components=["tool"])
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert any(t.name == "sample_tool" for t in tools)
async def test_tool_disabled_via_server(self):
@ -1492,7 +1492,7 @@ class TestToolEnabled:
return x * 2
mcp.disable(names={"sample_tool"}, components=["tool"])
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 0
with pytest.raises(NotFoundError, match="Unknown tool"):
@ -1507,7 +1507,7 @@ class TestToolEnabled:
mcp.disable(names={"sample_tool"}, components=["tool"])
mcp.enable(names={"sample_tool"}, components=["tool"])
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
async def test_tool_toggle_disabled(self):
@ -1518,7 +1518,7 @@ class TestToolEnabled:
return x * 2
mcp.disable(names={"sample_tool"}, components=["tool"])
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 0
with pytest.raises(NotFoundError, match="Unknown tool"):
@ -1535,7 +1535,7 @@ class TestToolEnabled:
assert tool is not None
mcp.disable(names={"sample_tool"}, components=["tool"])
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 0
with pytest.raises(NotFoundError, match="Unknown tool"):

134
tests/server/test_mount.py
View file

@ -42,7 +42,7 @@ class TestBasicMount:
main_app.mount(sub_app, "sub")
# Get tools from main app, should include sub_app's tools
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert any(t.name == "sub_tool" for t in tools)
assert any(t.name == "sub_transformed_tool" for t in tools)
@ -62,7 +62,7 @@ class TestBasicMount:
main_app.mount(sub_app, "sub")
# Tool should be accessible with the default separator
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert any(t.name == "sub_greet" for t in tools)
# Call the tool
@ -81,7 +81,7 @@ class TestBasicMount:
# Mount with empty prefix but without deprecated separators
main_app.mount(sub_app, namespace=prefix)
tools = await main_app.get_tools()
tools = await main_app.list_tools()
# With empty prefix, the tool should keep its original name
assert any(t.name == "sub_tool" for t in tools)
@ -97,7 +97,7 @@ class TestBasicMount:
# Mount without providing a prefix (should be None)
main_app.mount(sub_app)
tools = await main_app.get_tools()
tools = await main_app.list_tools()
# Without prefix, the tool should keep its original name
assert any(t.name == "sub_tool" for t in tools)
@ -118,7 +118,7 @@ class TestBasicMount:
main_app.mount(sub_app)
# Verify tool is accessible with original name
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert any(t.name == "sub_tool" for t in tools)
# Test actual functionality
@ -138,7 +138,7 @@ class TestBasicMount:
main_app.mount(sub_app)
# Verify resource is accessible with original URI
resources = await main_app.get_resources()
resources = await main_app.list_resources()
assert any(str(r.uri) == "data://config" for r in resources)
# Test actual functionality
@ -158,7 +158,7 @@ class TestBasicMount:
main_app.mount(sub_app)
# Verify template is accessible with original URI template
templates = await main_app.get_resource_templates()
templates = await main_app.list_resource_templates()
assert any(t.uri_template == "users://{user_id}/info" for t in templates)
# Test actual functionality
@ -178,7 +178,7 @@ class TestBasicMount:
main_app.mount(sub_app)
# Verify prompt is accessible with original name
prompts = await main_app.get_prompts()
prompts = await main_app.list_prompts()
assert any(p.name == "sub_prompt" for p in prompts)
# Test actual functionality
@ -208,7 +208,7 @@ class TestMultipleServerMount:
main_app.mount(news_app, "news")
# Check both are accessible
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert any(t.name == "weather_get_forecast" for t in tools)
assert any(t.name == "news_get_headlines" for t in tools)
@ -234,12 +234,12 @@ class TestMultipleServerMount:
# Mount first app
main_app.mount(first_app, "api")
tools = await main_app.get_tools()
tools = await main_app.list_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()
tools = await main_app.list_tools()
# Both apps' tools should be accessible (new behavior)
assert any(t.name == "api_first_tool" for t in tools)
@ -344,11 +344,10 @@ class TestPrefixConflictResolution:
main_app.mount(first_app)
main_app.mount(second_app)
# Test that get_tools shows the tool
tools = await main_app.get_tools()
# list_tools returns all components; execution uses first match
tools = await main_app.list_tools()
tool_names = [t.name for t in tools]
assert "shared_tool" in tool_names
assert tool_names.count("shared_tool") == 1 # Should only appear once
# Test that calling the tool uses the first server's implementation
result = await main_app.call_tool("shared_tool", {})
@ -372,11 +371,10 @@ class TestPrefixConflictResolution:
main_app.mount(first_app, "api")
main_app.mount(second_app, "api")
# Test that get_tools shows the tool
tools = await main_app.get_tools()
# list_tools returns all components; execution uses first match
tools = await main_app.list_tools()
tool_names = [t.name for t in tools]
assert "api_shared_tool" in tool_names
assert tool_names.count("api_shared_tool") == 1 # Should only appear once
# Test that calling the tool uses the first server's implementation
result = await main_app.call_tool("api_shared_tool", {})
@ -400,11 +398,10 @@ class TestPrefixConflictResolution:
main_app.mount(first_app)
main_app.mount(second_app)
# Test that get_resources shows the resource
resources = await main_app.get_resources()
# list_resources returns all components; execution uses first match
resources = await main_app.list_resources()
resource_uris = [str(r.uri) for r in resources]
assert "shared://data" in resource_uris
assert resource_uris.count("shared://data") == 1 # Should only appear once
# Test that reading the resource uses the first server's implementation
result = await main_app.read_resource("shared://data")
@ -428,11 +425,10 @@ class TestPrefixConflictResolution:
main_app.mount(first_app, "api")
main_app.mount(second_app, "api")
# Test that get_resources shows the resource
resources = await main_app.get_resources()
# list_resources returns all components; execution uses first match
resources = await main_app.list_resources()
resource_uris = [str(r.uri) for r in resources]
assert "shared://api/data" in resource_uris
assert resource_uris.count("shared://api/data") == 1 # Should only appear once
# Test that reading the resource uses the first server's implementation
result = await main_app.read_resource("shared://api/data")
@ -456,13 +452,10 @@ class TestPrefixConflictResolution:
main_app.mount(first_app)
main_app.mount(second_app)
# Test that get_resource_templates shows the template
templates = await main_app.get_resource_templates()
# list_resource_templates returns all components; execution uses first match
templates = await main_app.list_resource_templates()
template_uris = [t.uri_template for t in templates]
assert "users://{user_id}/profile" in template_uris
assert (
template_uris.count("users://{user_id}/profile") == 1
) # Should only appear once
# Test that reading the resource uses the first server's implementation
result = await main_app.read_resource("users://123/profile")
@ -486,13 +479,10 @@ class TestPrefixConflictResolution:
main_app.mount(first_app, "api")
main_app.mount(second_app, "api")
# Test that get_resource_templates shows the template
templates = await main_app.get_resource_templates()
# list_resource_templates returns all components; execution uses first match
templates = await main_app.list_resource_templates()
template_uris = [t.uri_template for t in templates]
assert "users://api/{user_id}/profile" in template_uris
assert (
template_uris.count("users://api/{user_id}/profile") == 1
) # Should only appear once
# Test that reading the resource uses the first server's implementation
result = await main_app.read_resource("users://api/123/profile")
@ -516,11 +506,10 @@ class TestPrefixConflictResolution:
main_app.mount(first_app)
main_app.mount(second_app)
# Test that get_prompts shows the prompt
prompts = await main_app.get_prompts()
# list_prompts returns all components; execution uses first match
prompts = await main_app.list_prompts()
prompt_names = [p.name for p in prompts]
assert "shared_prompt" in prompt_names
assert prompt_names.count("shared_prompt") == 1 # Should only appear once
# Test that getting the prompt uses the first server's implementation
result = await main_app.render_prompt("shared_prompt")
@ -546,11 +535,10 @@ class TestPrefixConflictResolution:
main_app.mount(first_app, "api")
main_app.mount(second_app, "api")
# Test that get_prompts shows the prompt
prompts = await main_app.get_prompts()
# list_prompts returns all components; execution uses first match
prompts = await main_app.list_prompts()
prompt_names = [p.name for p in prompts]
assert "api_shared_prompt" in prompt_names
assert prompt_names.count("api_shared_prompt") == 1 # Should only appear once
# Test that getting the prompt uses the first server's implementation
result = await main_app.render_prompt("api_shared_prompt")
@ -571,7 +559,7 @@ class TestDynamicChanges:
main_app.mount(sub_app, "sub")
# Initially, there should be no tools from sub_app
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert not any(t.name.startswith("sub_") for t in tools)
# Add a tool to the sub-app after mounting
@ -580,7 +568,7 @@ class TestDynamicChanges:
return "Added after mounting"
# The tool should be accessible through the main app
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert any(t.name == "sub_dynamic_tool" for t in tools)
# Call the dynamically added tool
@ -600,14 +588,14 @@ class TestDynamicChanges:
main_app.mount(sub_app, "sub")
# Initially, the tool should be accessible
tools = await main_app.get_tools()
tools = await main_app.list_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()
tools = await main_app.list_tools()
assert not any(t.name == "sub_temp_tool" for t in tools)
@ -627,7 +615,7 @@ class TestResourcesAndTemplates:
main_app.mount(data_app, "data")
# Resource should be accessible through main app
resources = await main_app.get_resources()
resources = await main_app.list_resources()
assert any(str(r.uri) == "data://data/users" for r in resources)
# Check that resource can be accessed
@ -650,7 +638,7 @@ class TestResourcesAndTemplates:
main_app.mount(user_app, "api")
# Template should be accessible through main app
templates = await main_app.get_resource_templates()
templates = await main_app.list_resource_templates()
assert any(t.uri_template == "users://api/{user_id}/profile" for t in templates)
# Check template instantiation
@ -674,7 +662,7 @@ class TestResourcesAndTemplates:
return json.dumps({"version": "1.0"})
# Resource should be accessible through main app
resources = await main_app.get_resources()
resources = await main_app.list_resources()
assert any(str(r.uri) == "data://data/config" for r in resources)
# Check access to the resource
@ -700,7 +688,7 @@ class TestPrompts:
main_app.mount(assistant_app, "assistant")
# Prompt should be accessible through main app
prompts = await main_app.get_prompts()
prompts = await main_app.list_prompts()
assert any(p.name == "assistant_greeting" for p in prompts)
# Render the prompt
@ -722,7 +710,7 @@ class TestPrompts:
return f"Goodbye, {name}!"
# Prompt should be accessible through main app
prompts = await main_app.get_prompts()
prompts = await main_app.list_prompts()
assert any(p.name == "assistant_farewell" for p in prompts)
# Render the prompt
@ -751,7 +739,7 @@ class TestProxyServer:
main_app.mount(proxy_server, "proxy")
# Tool should be accessible through main app
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert any(t.name == "proxy_get_data" for t in tools)
# Call the tool
@ -776,7 +764,7 @@ class TestProxyServer:
return "Dynamic data"
# Tool should be accessible through main app via proxy
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert any(t.name == "proxy_dynamic_data" for t in tools)
# Call the tool
@ -849,7 +837,7 @@ class TestAsProxyKwarg:
assert isinstance(provider._inner, FastMCPProvider)
assert provider._inner.server is sub
# Verify namespace is applied
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert {t.name for t in tools} == {"sub_sub_tool"}
async def test_as_proxy_false(self):
@ -872,7 +860,7 @@ class TestAsProxyKwarg:
assert isinstance(provider._inner, FastMCPProvider)
assert provider._inner.server is sub
# Verify namespace is applied
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert {t.name for t in tools} == {"sub_sub_tool"}
async def test_as_proxy_true(self):
@ -896,7 +884,7 @@ class TestAsProxyKwarg:
assert provider._inner.server is not sub
assert isinstance(provider._inner.server, FastMCPProxy)
# Verify namespace is applied
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert {t.name for t in tools} == {"sub_sub_tool"}
async def test_lifespan_server_mounted_directly(self):
@ -930,7 +918,7 @@ class TestAsProxyKwarg:
assert isinstance(provider._inner, FastMCPProvider)
assert provider._inner.server is sub
# Verify namespace is applied
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert {t.name for t in tools} == {"sub_sub_tool"}
async def test_as_proxy_ignored_for_proxy_mounts_default(self):
@ -990,13 +978,13 @@ class TestAsProxyKwarg:
mcp.mount(sub, "sub", as_proxy=True)
assert len(await mcp.get_tools()) == 0
assert len(await mcp.list_tools()) == 0
@sub.tool
def hello():
return "hi"
assert len(await mcp.get_tools()) == 1
assert len(await mcp.list_tools()) == 1
async def test_sub_lifespan_is_executed(self):
lifespan_check = []
@ -1043,7 +1031,7 @@ class TestResourceUriPrefixing:
main_app.mount(sub_app, "prefix")
# Get resources from main app
resources = await main_app.get_resources()
resources = await main_app.list_resources()
# Should have prefixed key (using path format: resource://prefix/resource_name)
assert any(str(r.uri) == "resource://prefix/my_resource" for r in resources)
@ -1069,7 +1057,7 @@ class TestResourceUriPrefixing:
main_app.mount(sub_app, "prefix")
# Get resource templates from main app
templates = await main_app.get_resource_templates()
templates = await main_app.list_resource_templates()
# Should have prefixed key (using path format: resource://prefix/template_uri)
assert any(
@ -1101,7 +1089,7 @@ class TestParentTagFiltering:
parent.mount(mounted)
tools = await parent.get_tools()
tools = await parent.list_tools()
tool_names = {t.name for t in tools}
assert "allowed_tool" in tool_names
assert "blocked_tool" not in tool_names
@ -1128,7 +1116,7 @@ class TestParentTagFiltering:
parent.mount(mounted)
tools = await parent.get_tools()
tools = await parent.list_tools()
tool_names = {t.name for t in tools}
assert "production_tool" in tool_names
assert "blocked_tool" not in tool_names
@ -1148,7 +1136,7 @@ class TestParentTagFiltering:
parent.mount(mounted)
resources = await parent.get_resources()
resources = await parent.list_resources()
resource_uris = {str(r.uri) for r in resources}
assert "resource://allowed" in resource_uris
assert "resource://blocked" not in resource_uris
@ -1168,7 +1156,7 @@ class TestParentTagFiltering:
parent.mount(mounted)
prompts = await parent.get_prompts()
prompts = await parent.list_prompts()
prompt_names = {p.name for p in prompts}
assert "allowed_prompt" in prompt_names
assert "blocked_prompt" not in prompt_names
@ -1236,7 +1224,7 @@ class TestCustomRouteForwarding:
assert provider2._inner.server == sub_server2
# Verify namespacing is applied by checking tool names
tools = await main_server.get_tools()
tools = await main_server.list_tools()
tool_names = {t.name for t in tools}
assert tool_names == {"sub1_tool1", "sub2_tool2"}
@ -1389,7 +1377,7 @@ class TestDeeplyNestedMount:
root.mount(level1, namespace="l1")
# Verify tool is listed
tools = await root.get_tools()
tools = await root.list_tools()
tool_names = [t.name for t in tools]
assert "l1_l2_l3_deep_tool" in tool_names
@ -1422,7 +1410,7 @@ class TestToolNameOverrides:
)
# Server introspection shows renamed + namespaced names
tools = await main.get_tools()
tools = await main.list_tools()
tool_names = [t.name for t in tools]
assert "prefix_custom_name" in tool_names
assert "original_tool" not in tool_names
@ -1444,7 +1432,7 @@ class TestToolNameOverrides:
tool_names={"original_tool": "custom_name"},
)
tools = await main.get_tools()
tools = await main.list_tools()
tool_names = [t.name for t in tools]
assert "prefix_custom_name" in tool_names
assert "prefix_original_tool" not in tool_names
@ -1541,18 +1529,18 @@ class TestComponentServicePrefixLess:
main_app.mount(sub_app)
# Initially the tool is enabled
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert any(t.name == "my_tool" for t in tools)
# Disable and re-enable
main_app.disable(names={"my_tool"}, components=["tool"])
# Verify tool is now disabled
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert not any(t.name == "my_tool" for t in tools)
main_app.enable(names={"my_tool"}, components=["tool"])
# Verify tool is now enabled
tools = await main_app.get_tools()
tools = await main_app.list_tools()
assert any(t.name == "my_tool" for t in tools)
async def test_enable_resource_prefixless_mount(self):
@ -1570,12 +1558,12 @@ class TestComponentServicePrefixLess:
# Disable and re-enable
main_app.disable(names={"data://test"}, components=["resource"])
# Verify resource is now disabled
resources = await main_app.get_resources()
resources = await main_app.list_resources()
assert not any(str(r.uri) == "data://test" for r in resources)
main_app.enable(names={"data://test"}, components=["resource"])
# Verify resource is now enabled
resources = await main_app.get_resources()
resources = await main_app.list_resources()
assert any(str(r.uri) == "data://test" for r in resources)
async def test_enable_prompt_prefixless_mount(self):
@ -1593,10 +1581,10 @@ class TestComponentServicePrefixLess:
# Disable and re-enable
main_app.disable(names={"my_prompt"}, components=["prompt"])
# Verify prompt is now disabled
prompts = await main_app.get_prompts()
prompts = await main_app.list_prompts()
assert not any(p.name == "my_prompt" for p in prompts)
main_app.enable(names={"my_prompt"}, components=["prompt"])
# Verify prompt is now enabled
prompts = await main_app.get_prompts()
prompts = await main_app.list_prompts()
assert any(p.name == "my_prompt" for p in prompts)

16
tests/server/test_providers.py
View file

@ -136,7 +136,7 @@ class TestProvider:
provider = SimpleToolProvider(tools=dynamic_tools)
base_server.add_provider(provider)
tools = await base_server.get_tools()
tools = await base_server.list_tools()
# Should have all tools: 2 static + 2 dynamic
assert len(tools) == 4
@ -154,9 +154,9 @@ class TestProvider:
base_server.add_provider(provider)
# Call get_tools multiple times
await base_server.get_tools()
await base_server.get_tools()
await base_server.get_tools()
await base_server.list_tools()
await base_server.list_tools()
await base_server.list_tools()
# Provider should have been called 3 times (once per get_tools call)
assert provider.list_tools_call_count == 3
@ -250,7 +250,7 @@ class TestProvider:
provider = SimpleToolProvider(tools=dynamic_tools)
base_server.add_provider(provider)
tools = await base_server.get_tools()
tools = await base_server.list_tools()
tool_names = [tool.name for tool in tools]
# Local tools should come first (LocalProvider is first in _providers)
@ -261,7 +261,7 @@ class TestProvider:
provider = SimpleToolProvider(tools=[])
base_server.add_provider(provider)
tools = await base_server.get_tools()
tools = await base_server.list_tools()
# Should only have static tools
assert len(tools) == 2
@ -330,7 +330,7 @@ class TestDynamicToolUpdates:
provider = SimpleToolProvider(tools=initial_tools)
mcp.add_provider(provider)
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].name == "tool_v1"
@ -351,7 +351,7 @@ class TestDynamicToolUpdates:
]
# List tools again - should see new tools
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 2
tool_names = [t.name for t in tools]
assert "tool_v1" not in tool_names

6
tests/server/test_server.py
View file

@ -112,7 +112,7 @@ class TestServerDelegation:
def local_tool() -> str:
return "local"
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert any(t.name == "local_tool" for t in tools)
@ -141,8 +141,8 @@ class TestResourcePrefixMounting:
main_server.mount(server, "prefix")
# Check that the resources are mounted with the correct prefixes
resources = await main_server.get_resources()
templates = await main_server.get_resource_templates()
resources = await main_server.list_resources()
templates = await main_server.list_resource_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)

6
tests/server/test_tool_annotations.py
View file

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

8
tests/server/test_tool_transformation.py
View file

@ -16,7 +16,7 @@ async def test_tool_transformation_via_layer():
ToolTransform({"echo": ToolTransformConfig(name="echo_transformed")})
)
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert any(t.name == "echo_transformed" for t in tools)
tool = next(t for t in tools if t.name == "echo_transformed")
@ -45,7 +45,7 @@ async def test_transformed_tool_filtering():
# Enable only tools with the enabled_tools tag
mcp.enable(tags={"enabled_tools"}, only=True)
tools = await mcp.get_tools()
tools = await mcp.list_tools()
# With transformation applied, the tool now has the enabled_tools tag
assert len(tools) == 1
@ -92,7 +92,7 @@ async def test_layer_based_transforms():
ToolTransform({"my_tool": ToolTransformConfig(name="renamed_tool")})
)
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].name == "renamed_tool"
@ -113,7 +113,7 @@ async def test_server_level_transforms_apply_to_mounted_servers():
ToolTransform({"sub_tool": ToolTransformConfig(name="renamed_sub_tool")})
)
tools = await main.get_tools()
tools = await main.list_tools()
tool_names = [t.name for t in tools]
assert "renamed_sub_tool" in tool_names

376
tests/server/test_versioning.py
View file

@ -92,7 +92,7 @@ class TestComponentVersioning:
def my_tool(x: int) -> int:
return x * 2
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].name == "my_tool"
assert tools[0].version == "2.0"
@ -106,7 +106,7 @@ class TestComponentVersioning:
def my_tool(x: int) -> int:
return x * 2
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].version is None
# Keys always have @ sentinel for unambiguous parsing
@ -120,7 +120,7 @@ class TestComponentVersioning:
def my_tool(x: int) -> int:
return x * 2
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].version == "2"
assert tools[0].key == "tool:my_tool@2"
@ -133,13 +133,13 @@ class TestComponentVersioning:
def my_tool(x: int) -> int:
return x * 2
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].version == "0"
assert tools[0].key == "tool:my_tool@0" # Not "tool:my_tool@"
async def test_multiple_tool_versions_deduplicated(self):
"""Multiple versions of same tool should deduplicate to highest."""
async def test_multiple_tool_versions_all_returned(self):
"""list_tools returns all versions; get_tool returns highest."""
mcp = FastMCP()
@mcp.tool(version="1.0")
@ -150,11 +150,16 @@ class TestComponentVersioning:
def add(x: int, y: int, z: int = 0) -> int:
return x + y + z
tools = await mcp.get_tools()
# Should only show the highest version
assert len(tools) == 1
assert tools[0].name == "add"
assert tools[0].version == "2.0"
# list_tools returns all versions
tools = await mcp.list_tools()
assert len(tools) == 2
versions = {t.version for t in tools}
assert versions == {"1.0", "2.0"}
# get_tool returns highest version
tool = await mcp.get_tool("add")
assert tool is not None
assert tool.version == "2.0"
async def test_call_tool_invokes_highest_version(self):
"""Calling a tool by name should invoke the highest version."""
@ -211,17 +216,24 @@ class TestComponentVersioning:
"""Resource version should work like tool version."""
mcp = FastMCP()
@mcp.resource("file://config", version="1.0")
@mcp.resource("file:///config", version="1.0")
def config_v1() -> str:
return "config v1"
@mcp.resource("file://config", version="2.0")
@mcp.resource("file:///config", version="2.0")
def config_v2() -> str:
return "config v2"
resources = await mcp.get_resources()
assert len(resources) == 1
assert resources[0].version == "2.0"
# list_resources returns all versions
resources = await mcp.list_resources()
assert len(resources) == 2
versions = {r.version for r in resources}
assert versions == {"1.0", "2.0"}
# get_resource returns highest version
resource = await mcp.get_resource("file:///config")
assert resource is not None
assert resource.version == "2.0"
async def test_prompt_with_version(self):
"""Prompt version should work like tool version."""
@ -235,9 +247,16 @@ class TestComponentVersioning:
def greet(name: str) -> str:
return f"Greetings, {name}!"
prompts = await mcp.get_prompts()
assert len(prompts) == 1
assert prompts[0].version == "2.0"
# list_prompts returns all versions
prompts = await mcp.list_prompts()
assert len(prompts) == 2
versions = {p.version for p in prompts}
assert versions == {"1.0", "2.0"}
# get_prompt returns highest version
prompt = await mcp.get_prompt("greet")
assert prompt is not None
assert prompt.version == "2.0"
class TestVersionSorting:
@ -260,11 +279,18 @@ class TestVersionSorting:
def count() -> int:
return 2
tools = await mcp.get_tools()
# Should keep v10 as highest (semantic: 10 > 2 > 1)
assert len(tools) == 1
assert tools[0].version == "10"
# list_tools returns all versions
tools = await mcp.list_tools()
assert len(tools) == 3
versions = {t.version for t in tools}
assert versions == {"1", "2", "10"}
# get_tool returns highest (semantic: 10 > 2 > 1)
tool = await mcp.get_tool("count")
assert tool is not None
assert tool.version == "10"
# call_tool uses highest version
result = await mcp.call_tool("count", {})
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "10"
@ -285,10 +311,16 @@ class TestVersionSorting:
def info() -> str:
return "1.10.1"
tools = await mcp.get_tools()
assert len(tools) == 1
# 1.10.1 > 1.2.10 > 1.2.3 (semantic)
assert tools[0].version == "1.10.1"
# list_tools returns all versions
tools = await mcp.list_tools()
assert len(tools) == 3
versions = {t.version for t in tools}
assert versions == {"1.2.3", "1.2.10", "1.10.1"}
# get_tool returns highest: 1.10.1 > 1.2.10 > 1.2.3 (semantic)
tool = await mcp.get_tool("info")
assert tool is not None
assert tool.version == "1.10.1"
async def test_v_prefix_normalized(self):
"""Versions with 'v' prefix should compare correctly."""
@ -302,9 +334,16 @@ class TestVersionSorting:
def calc() -> int:
return 2
tools = await mcp.get_tools()
assert len(tools) == 1
assert tools[0].version == "v2.0"
# list_tools returns all versions
tools = await mcp.list_tools()
assert len(tools) == 2
versions = {t.version for t in tools}
assert versions == {"v1.0", "v2.0"}
# get_tool returns highest
tool = await mcp.get_tool("calc")
assert tool is not None
assert tool.version == "v2.0"
class TestMountedServerVersioning:
@ -321,7 +360,7 @@ class TestMountedServerVersioning:
parent = FastMCP("Parent")
parent.mount(child, "child")
tools = await parent.get_tools()
tools = await parent.list_tools()
assert len(tools) == 1
assert tools[0].name == "child_add"
assert tools[0].version == "2.0"
@ -330,14 +369,14 @@ class TestMountedServerVersioning:
"""Mounted resources should preserve their version info."""
child = FastMCP("Child")
@child.resource("file://config", version="1.5")
@child.resource("file:///config", version="1.5")
def config() -> str:
return "config data"
parent = FastMCP("Parent")
parent.mount(child, "child")
resources = await parent.get_resources()
resources = await parent.list_resources()
assert len(resources) == 1
assert resources[0].version == "1.5"
@ -352,7 +391,7 @@ class TestMountedServerVersioning:
parent = FastMCP("Parent")
parent.mount(child, "child")
prompts = await parent.get_prompts()
prompts = await parent.list_prompts()
assert len(prompts) == 1
assert prompts[0].name == "child_greet"
assert prompts[0].version == "3.0"
@ -382,8 +421,8 @@ class TestMountedServerVersioning:
assert tool_v1 is not None
assert tool_v1.version == "1.0"
async def test_mounted_multiple_versions_deduplicates(self):
"""Mounted server with multiple versions should show only highest."""
async def test_mounted_multiple_versions_all_returned(self):
"""Mounted server with multiple versions should show all versions."""
child = FastMCP("Child")
@child.tool(version="1.0")
@ -401,9 +440,16 @@ class TestMountedServerVersioning:
parent = FastMCP("Parent")
parent.mount(child, "child")
tools = await parent.get_tools()
assert len(tools) == 1
assert tools[0].version == "3.0"
# list_tools returns all versions
tools = await parent.list_tools()
assert len(tools) == 3
versions = {t.version for t in tools}
assert versions == {"1.0", "2.0", "3.0"}
# get_tool returns highest
tool = await parent.get_tool("child_my_tool")
assert tool is not None
assert tool.version == "3.0"
async def test_mounted_call_tool_uses_highest_version(self):
"""Calling mounted tool should use highest version."""
@ -425,6 +471,124 @@ class TestMountedServerVersioning:
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "110"
async def test_mounted_tool_wrapper_executes_correct_version(self):
"""Calling a specific versioned tool wrapper should execute that version."""
child = FastMCP("Child")
@child.tool(version="1.0")
def calc(x: int) -> int:
return x * 10 # v1.0 multiplies by 10
@child.tool(version="2.0")
def calc(x: int) -> int:
return x * 100 # v2.0 multiplies by 100
parent = FastMCP("Parent")
parent.mount(child, "child")
# Get the v1.0 wrapper specifically
tools = await parent.list_tools()
v1_tool = next(
t for t in tools if t.name == "child_calc" and t.version == "1.0"
)
# Calling the v1.0 wrapper should execute v1.0's logic
result = await v1_tool.run({"x": 5})
assert result.content[0].text == "50" # 5 * 10, not 5 * 100
async def test_mounted_resource_wrapper_reads_correct_version(self):
"""Reading a specific versioned resource should read that version."""
from fastmcp.utilities.versions import VersionSpec
child = FastMCP("Child")
@child.resource("data:///config", version="1.0")
def config_v1() -> str:
return "config-v1-content"
@child.resource("data:///config", version="2.0")
def config_v2() -> str:
return "config-v2-content"
parent = FastMCP("Parent")
parent.mount(child, "child")
# Reading with version=1.0 should read v1.0's content
result = await parent.read_resource(
"data://child//config", version=VersionSpec(eq="1.0")
)
assert result.contents[0].content == "config-v1-content"
# Reading with version=2.0 should read v2.0's content
result = await parent.read_resource(
"data://child//config", version=VersionSpec(eq="2.0")
)
assert result.contents[0].content == "config-v2-content"
async def test_mounted_prompt_wrapper_renders_correct_version(self):
"""Rendering a specific versioned prompt should render that version."""
from fastmcp.utilities.versions import VersionSpec
child = FastMCP("Child")
@child.prompt(version="1.0")
def greeting(name: str) -> str:
return f"Hello, {name}!" # v1.0 says Hello
@child.prompt(version="2.0")
def greeting(name: str) -> str:
return f"Greetings, {name}!" # v2.0 says Greetings
parent = FastMCP("Parent")
parent.mount(child, "child")
# Rendering with version=1.0 should render v1.0's content
result = await parent.render_prompt(
"child_greeting", {"name": "World"}, version=VersionSpec(eq="1.0")
)
content = result.messages[0].content
assert isinstance(content, TextContent) and "Hello, World!" in content.text
# Rendering with version=2.0 should render v2.0's content
result = await parent.render_prompt(
"child_greeting", {"name": "World"}, version=VersionSpec(eq="2.0")
)
content = result.messages[0].content
assert isinstance(content, TextContent) and "Greetings, World!" in content.text
async def test_deeply_nested_version_forwarding(self):
"""Verify version is correctly forwarded through multiple mount levels."""
level3 = FastMCP("Level3")
@level3.tool(version="1.0")
def calc(x: int) -> int:
return x * 10 # v1.0 multiplies by 10
@level3.tool(version="2.0")
def calc(x: int) -> int:
return x * 100 # v2.0 multiplies by 100
level2 = FastMCP("Level2")
level2.mount(level3, "l3")
level1 = FastMCP("Level1")
level1.mount(level2, "l2")
# All versions should be visible through two levels of mounting
tools = await level1.list_tools()
calc_tools = [t for t in tools if "calc" in t.name]
assert len(calc_tools) == 2
versions = {t.version for t in calc_tools}
assert versions == {"1.0", "2.0"}
# Get v1.0 wrapper through two levels of mounting
v1_tool = next(t for t in tools if "calc" in t.name and t.version == "1.0")
# Should execute v1.0 logic, not v2.0
result = await v1_tool.run({"x": 5})
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "50" # 5 * 10, not 5 * 100
class TestVersionFilter:
"""Tests for VersionFilter transform."""
@ -447,16 +611,21 @@ class TestVersionFilter:
def calc() -> int:
return 3
# Without filter, should show v3 (highest)
tools = await mcp.get_tools()
assert len(tools) == 1
assert tools[0].version == "3.0"
# Without filter, list_tools returns all versions
tools = await mcp.list_tools()
versions = {t.version for t in tools}
assert versions == {"1.0", "2.0", "3.0"}
# With filter, should show v2 (highest below 3.0)
# With filter, only v1 and v2 are visible
mcp.add_transform(VersionFilter(version_lt="3.0"))
tools = await mcp.get_tools()
assert len(tools) == 1
assert tools[0].version == "2.0"
tools = await mcp.list_tools()
versions = {t.version for t in tools}
assert versions == {"1.0", "2.0"}
# get_tool returns highest matching version
tool = await mcp.get_tool("calc")
assert tool is not None
assert tool.version == "2.0"
async def test_version_gte_filters_low_versions(self):
"""VersionFilter(version_gte='2.0') hides v1, shows v2 and v3."""
@ -478,12 +647,17 @@ class TestVersionFilter:
mcp.add_transform(VersionFilter(version_gte="2.0"))
# Should show v3 (highest >= 2.0)
tools = await mcp.get_tools()
assert len(tools) == 1
assert tools[0].version == "3.0"
# list_tools shows all matching versions (v2 and v3)
tools = await mcp.list_tools()
versions = {t.version for t in tools}
assert versions == {"2.0", "3.0"}
# Can request specific versions in range (use get_tool to apply transforms)
# get_tool returns highest matching version
tool = await mcp.get_tool("add")
assert tool is not None
assert tool.version == "3.0"
# Can request specific versions in range
tool_v2 = await mcp.get_tool("add", VersionSpec(eq="2.0"))
assert tool_v2 is not None
assert tool_v2.version == "2.0"
@ -515,12 +689,17 @@ class TestVersionFilter:
mcp.add_transform(VersionFilter(version_gte="2.0", version_lt="3.0"))
# Should show v2.5 (highest in range)
tools = await mcp.get_tools()
assert len(tools) == 1
assert tools[0].version == "2.5"
# list_tools shows all versions in range
tools = await mcp.list_tools()
versions = {t.version for t in tools}
assert versions == {"2.0", "2.5"}
# Can request specific versions in range (use get_tool to apply transforms)
# get_tool returns highest in range
tool = await mcp.get_tool("calc")
assert tool is not None
assert tool.version == "2.5"
# Can request specific versions in range
tool_v2 = await mcp.get_tool("calc", VersionSpec(eq="2.0"))
assert tool_v2 is not None
assert tool_v2.version == "2.0"
@ -546,7 +725,7 @@ class TestVersionFilter:
# Filter that would exclude v5.0
mcp.add_transform(VersionFilter(version_lt="3.0"))
tools = await mcp.get_tools()
tools = await mcp.list_tools()
names = [t.name for t in tools]
assert "unversioned_tool" in names
assert "versioned_tool" not in names
@ -572,7 +751,7 @@ class TestVersionFilter:
# Q1 API: before April
mcp.add_transform(VersionFilter(version_lt="2025-04-01"))
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
assert tools[0].version == "2025-01-01"
@ -607,17 +786,17 @@ class TestVersionFilter:
mcp = FastMCP()
@mcp.resource("file://config", version="1.0")
@mcp.resource("file:///config", version="1.0")
def config_v1() -> str:
return "v1"
@mcp.resource("file://config", version="2.0")
@mcp.resource("file:///config", version="2.0")
def config_v2() -> str:
return "v2"
mcp.add_transform(VersionFilter(version_lt="2.0"))
resources = await mcp.get_resources()
resources = await mcp.list_resources()
assert len(resources) == 1
assert resources[0].version == "1.0"
@ -637,7 +816,7 @@ class TestVersionFilter:
mcp.add_transform(VersionFilter(version_lt="2.0"))
prompts = await mcp.get_prompts()
prompts = await mcp.list_prompts()
assert len(prompts) == 1
assert prompts[0].version == "1.0"
@ -664,13 +843,13 @@ class TestVersionMixingValidation:
mcp = FastMCP()
@mcp.resource("file://config", version="1.0")
@mcp.resource("file:///config", version="1.0")
def config_v1() -> str:
return "v1"
with pytest.raises(ValueError, match="unversioned.*versioned"):
@mcp.resource("file://config")
@mcp.resource("file:///config")
def config_unversioned() -> str:
return "unversioned"
@ -706,10 +885,16 @@ class TestVersionMixingValidation:
def calc() -> int:
return 3
# All versioned - this should work
tools = await mcp.get_tools()
assert len(tools) == 1
assert tools[0].version == "3.0"
# All versioned - list_tools returns all
tools = await mcp.list_tools()
assert len(tools) == 3
versions = {t.version for t in tools}
assert versions == {"1.0", "2.0", "3.0"}
# get_tool returns highest
tool = await mcp.get_tool("calc")
assert tool is not None
assert tool.version == "3.0"
class TestMountedVersionFiltering:
@ -808,7 +993,7 @@ class TestMountedVersionFiltering:
parent.add_transform(VersionFilter(version_lt="3.0"))
# Unversioned should pass through
tools = await parent.get_tools()
tools = await parent.list_tools()
assert len(tools) == 1
assert tools[0].name == "child_unversioned_tool"
assert tools[0].version is None
@ -828,7 +1013,7 @@ class TestMountedVersionFiltering:
parent.add_transform(VersionFilter(version_lt="3.0"))
# v5.0 is outside the filter range, so it should be hidden
tools = await parent.get_tools()
tools = await parent.list_tools()
assert len(tools) == 0
# get_tool should also return None (respects filter, applies transforms)
@ -914,7 +1099,7 @@ class TestUnversionedExemption:
# Filter that would exclude v5.0
mcp.add_transform(VersionFilter(version_lt="3.0"))
tools = await mcp.get_tools()
tools = await mcp.list_tools()
names = [t.name for t in tools]
# Unversioned passes through (exempt from filtering)
@ -972,7 +1157,7 @@ class TestVersionMetadata:
"""Tests for version metadata exposure in list operations."""
async def test_tool_versions_in_meta(self):
"""List tools should include versions list in meta."""
"""Each version has its own version in metadata."""
mcp = FastMCP()
@mcp.tool(version="1.0")
@ -983,16 +1168,17 @@ class TestVersionMetadata:
def add(x: int, y: int) -> int: # noqa: F811
return x + y
tools = await mcp.get_tools()
assert len(tools) == 1
# list_tools returns all versions
tools = await mcp.list_tools()
assert len(tools) == 2
tool = tools[0]
meta = tool.get_meta()
assert meta["fastmcp"]["version"] == "2.0"
assert meta["fastmcp"]["versions"] == ["2.0", "1.0"]
# Each version has its own version in metadata
by_version = {t.version: t for t in tools}
assert by_version["1.0"].get_meta()["fastmcp"]["version"] == "1.0"
assert by_version["2.0"].get_meta()["fastmcp"]["version"] == "2.0"
async def test_resource_versions_in_meta(self):
"""List resources should include versions list in meta."""
"""Each version has its own version in metadata."""
mcp = FastMCP()
@mcp.resource("data://config", version="1.0")
@ -1003,16 +1189,17 @@ class TestVersionMetadata:
def config_v2() -> str: # noqa: F811
return "v2"
resources = await mcp.get_resources()
assert len(resources) == 1
# list_resources returns all versions
resources = await mcp.list_resources()
assert len(resources) == 2
resource = resources[0]
meta = resource.get_meta()
assert meta["fastmcp"]["version"] == "2.0"
assert meta["fastmcp"]["versions"] == ["2.0", "1.0"]
# Each version has its own version in metadata
by_version = {r.version: r for r in resources}
assert by_version["1.0"].get_meta()["fastmcp"]["version"] == "1.0"
assert by_version["2.0"].get_meta()["fastmcp"]["version"] == "2.0"
async def test_prompt_versions_in_meta(self):
"""List prompts should include versions list in meta."""
"""Each version has its own version in metadata."""
mcp = FastMCP()
@mcp.prompt(version="1.0")
@ -1023,13 +1210,14 @@ class TestVersionMetadata:
def greet() -> str: # noqa: F811
return "Hello v2"
prompts = await mcp.get_prompts()
assert len(prompts) == 1
# list_prompts returns all versions
prompts = await mcp.list_prompts()
assert len(prompts) == 2
prompt = prompts[0]
meta = prompt.get_meta()
assert meta["fastmcp"]["version"] == "2.0"
assert meta["fastmcp"]["versions"] == ["2.0", "1.0"]
# Each version has its own version in metadata
by_version = {p.version: p for p in prompts}
assert by_version["1.0"].get_meta()["fastmcp"]["version"] == "1.0"
assert by_version["2.0"].get_meta()["fastmcp"]["version"] == "2.0"
async def test_unversioned_no_versions_list(self):
"""Unversioned components should not have versions list in meta."""
@ -1039,7 +1227,7 @@ class TestVersionMetadata:
def simple() -> str:
return "simple"
tools = await mcp.get_tools()
tools = await mcp.list_tools()
assert len(tools) == 1
tool = tools[0]

2
tests/tools/test_tool_timeout.py
View file

@ -141,7 +141,7 @@ class TestToolTimeout:
return "completed"
# Tool should be registered successfully
tools = await mcp.get_tools()
tools = await mcp.list_tools()
tool = next((t for t in tools if t.name == "task_with_timeout"), None)
assert tool is not None
assert tool.timeout == 1.0