Skip to content

Conversation

@neelay-aign
Copy link
Collaborator

Merge when FastMCP releases a v3.x to PyPI and we switch to that in the pyproject.toml

Copilot AI review requested due to automatic review settings February 10, 2026 20:34
@atlantis-platform-engineering
Error: This repo is not allowlisted for Atlantis.

Copy link

Copilot AI left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Prepare the MCP integration code for FastMCP v3.x API changes (mount namespacing and async tool listing) ahead of switching the dependency in pyproject.toml.

Changes:

  • Update mcp.mount(..., prefix=...) to mcp.mount(..., namespace=...).
  • Replace get_tools() (dict) with list_tools() (list) and adjust tool formatting output.
  • Update docstrings/comments to reflect the new FastMCP API terminology.

seen_names.add(server.name)
logger.info(f"Mounting MCP server: {server.name}")
mcp.mount(server, prefix=server.name)
mcp.mount(server, namespace=server.name)
Copy link

Copilot AI Feb 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

namespace on mount() and list_tools() are FastMCP v3.x API calls; if this PR lands before the dependency is bumped to v3.x, this will raise at runtime (unexpected keyword argument / missing attribute). To make this change safe (and keep CI green) until pyproject.toml is updated, consider supporting both APIs via capability detection (e.g., try namespace then fallback to prefix, and try list_tools() then fallback to get_tools()).

Copilot uses AI. Check for mistakes.
# lazily initialize resources. We use asyncio.run() to bridge sync/async.
tools = asyncio.run(server.get_tools())
return [{"name": name, "description": tool.description or ""} for name, tool in tools.items()]
tools = asyncio.run(server.list_tools())
Copy link

Copilot AI Feb 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

namespace on mount() and list_tools() are FastMCP v3.x API calls; if this PR lands before the dependency is bumped to v3.x, this will raise at runtime (unexpected keyword argument / missing attribute). To make this change safe (and keep CI green) until pyproject.toml is updated, consider supporting both APIs via capability detection (e.g., try namespace then fallback to prefix, and try list_tools() then fallback to get_tools()).

Copilot uses AI. Check for mistakes.
tools = asyncio.run(server.get_tools())
return [{"name": name, "description": tool.description or ""} for name, tool in tools.items()]
tools = asyncio.run(server.list_tools())
return [{"name": tool.name, "description": tool.description or ""} for tool in tools]
Copy link

Copilot AI Feb 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

list_tools() returning a list can make output ordering depend on server/tool registration order, which may be non-deterministic across environments and can cause flaky CLI output/tests. Consider sorting the returned tools by tool.name before formatting the list so mcp_list_tools() has stable output.

Suggested change
return [{"name": tool.name, "description": tool.description or ""} for tool in tools]
# Sort tools by name to ensure deterministic output order across environments
sorted_tools = sorted(tools, key=lambda tool: tool.name)
return [{"name": tool.name, "description": tool.description or ""} for tool in sorted_tools]

Copilot uses AI. Check for mistakes.
Comment on lines +120 to +121
tools = asyncio.run(server.list_tools())
return [{"name": tool.name, "description": tool.description or ""} for tool in tools]
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: Tests in mcp_test.py use the outdated FastMCP v2 API (get_tools()), while production code uses the new v3 API (list_tools()), guaranteeing future test failures.
Severity: CRITICAL

Suggested Fix

Update the test file tests/aignostics/utils/mcp_test.py to use the new FastMCP v3.x API. Replace calls to server.get_tools() with server.list_tools(). Update the test logic to handle a list of Tool objects instead of a dictionary, accessing tool names via the tool.name attribute.

Prompt for AI Agent
Review the code at the location below. A potential bug has been identified by an AI
agent.
Verify if this is a real issue. If it is, propose a fix; if not, explain why it's not
valid.

Location: src/aignostics/utils/_mcp.py#L120-L121

Potential issue: The production code was updated to use the FastMCP v3.x API,
specifically changing from `server.get_tools()` to `server.list_tools()`. However, the
corresponding test file, `tests/aignostics/utils/mcp_test.py`, was not updated. It still
calls the old `get_tools()` method and expects a dictionary, while the new
`list_tools()` method returns a list of `Tool` objects. When the FastMCP dependency is
updated to v3.x as intended for this pull request, the tests will fail with an
`AttributeError`, blocking deployment.

Did we get this right? 👍 / 👎 to inform future reviews.

@codecov
Copy link

codecov bot commented Feb 10, 2026

❌ 6 Tests Failed:

Tests completed Failed Passed Skipped
512 6 506 3
View the full list of 6 ❄️ flaky test(s)
tests.aignostics.cli_test::test_cli_mcp_list_tools

Flake rate in main: 100.00% (Passed 0 times, Failed 1 times)

Stack Traces | 0.036s run time
runner = <typer.testing.CliRunner object at 0x7f89821ac710>
record_property = <function record_property.<locals>.append_property at 0x7f89821a3c10>

    @pytest.mark.unit
    def test_cli_mcp_list_tools(runner: CliRunner, record_property) -> None:
        """Test list-tools command displays discovered tools."""
        record_property("tested-item-id", "SPEC-UTILS-SERVICE")
        test_server = FastMCP("test")
    
        @test_server.tool
        def test_tool() -> str:
            """A test tool."""
            return "test"
    
        with patch(PATCH_MCP_LOCATE_IMPLEMENTATIONS, return_value=[test_server]):
            result = runner.invoke(cli, ["mcp", "list-tools"])
>           assert result.exit_code == 0
E           assert 1 == 0
E            +  where 1 = <Result TypeError("FastMCP.mount() got an unexpected keyword argument 'namespace'")>.exit_code

tests/aignostics/cli_test.py:321: AssertionError
tests.aignostics.cli_test::test_cli_mcp_list_tools_empty

Flake rate in main: 100.00% (Passed 0 times, Failed 1 times)

Stack Traces | 0.025s run time
runner = <typer.testing.CliRunner object at 0x7f89820057b0>
record_property = <function record_property.<locals>.append_property at 0x7f8982059220>

    @pytest.mark.unit
    def test_cli_mcp_list_tools_empty(runner: CliRunner, record_property) -> None:
        """Test list-tools command with no tools."""
        record_property("tested-item-id", "SPEC-UTILS-SERVICE")
        with patch(PATCH_MCP_LOCATE_IMPLEMENTATIONS, return_value=[]):
            result = runner.invoke(cli, ["mcp", "list-tools"])
>           assert result.exit_code == 0
E           assert 1 == 0
E            +  where 1 = <Result AttributeError("'FastMCP' object has no attribute 'list_tools'")>.exit_code

tests/aignostics/cli_test.py:332: AssertionError
tests.aignostics.utils.mcp_test::test_mcp_create_server_mounts_discovered

Flake rate in main: 100.00% (Passed 0 times, Failed 1 times)

Stack Traces | 0.019s run time
record_property = <function record_property.<locals>.append_property at 0x7f9763a15e80>

    @pytest.mark.unit
    def test_mcp_create_server_mounts_discovered(record_property) -> None:
        """Test that mcp_create_server mounts discovered servers with their tools."""
        record_property("tested-item-id", "SPEC-UTILS-SERVICE")
        plugin1 = FastMCP("plugin1")
        plugin2 = FastMCP("plugin2")
    
        @plugin1.tool
        def plugin1_tool() -> str:
            """Plugin 1 tool."""
            return "p1"
    
        @plugin2.tool
        def plugin2_tool() -> str:
            """Plugin 2 tool."""
            return "p2"
    
        with patch(PATCH_LOCATE_IMPLEMENTATIONS, return_value=[plugin1, plugin2]):
>           server = mcp_create_server()
                     ^^^^^^^^^^^^^^^^^^^

.../aignostics/utils/mcp_test.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

server_name = 'Central Aignostics MCP Server'

    def mcp_create_server(server_name: str = MCP_SERVER_NAME) -> FastMCP:
        """Create and configure the MCP server with all discovered plugins mounted.
    
        Creates a new FastMCP server instance and mounts all discovered MCP servers
        from the SDK and plugins. Each mounted server's tools are namespaced
        automatically using FastMCP's built-in namespace feature.
    
        Args:
            server_name: Human-readable name for the MCP server.
    
        Returns:
            FastMCP: Configured MCP server ready to run.
        """
        mcp = FastMCP(name=server_name, version=__version__)
    
        # Mount discovered servers
        servers = mcp_discover_servers()
        seen_names: set[str] = set()
        count = 0
    
        for server in servers:
            if server is not mcp:  # Don't mount self
                if server.name in seen_names:
                    logger.warning(f"Duplicate MCP server name '{server.name}' - skipping to avoid tool collision")
                    continue
                seen_names.add(server.name)
                logger.info(f"Mounting MCP server: {server.name}")
>               mcp.mount(server, namespace=server.name)
E               TypeError: FastMCP.mount() got an unexpected keyword argument 'namespace'

.../aignostics/utils/_mcp.py:79: TypeError
tests.aignostics.utils.mcp_test::test_mcp_create_server_skips_duplicate_names

Flake rate in main: 100.00% (Passed 0 times, Failed 1 times)

Stack Traces | 0.023s run time
caplog = <_pytest.logging.LogCaptureFixture object at 0x7f9763a87d20>
record_property = <function record_property.<locals>.append_property at 0x7f9763a7b110>

    @pytest.mark.unit
    def test_mcp_create_server_skips_duplicate_names(caplog: pytest.LogCaptureFixture, record_property) -> None:
        """Test that servers with duplicate names are skipped with warning."""
        record_property("tested-item-id", "SPEC-UTILS-SERVICE")
        dup1 = FastMCP("duplicate_name")
        dup2 = FastMCP("duplicate_name")
        unique = FastMCP("unique_name")
    
        @dup1.tool
        def dup1_tool() -> str:
            """Dup1 tool."""
            return "dup1"
    
        @dup2.tool
        def dup2_tool() -> str:
            """Dup2 tool - should NOT be mounted."""
            return "dup2"
    
        @unique.tool
        def unique_tool() -> str:
            """Unique tool."""
            return "unique"
    
        with patch(PATCH_LOCATE_IMPLEMENTATIONS, return_value=[dup1, dup2, unique]):
>           server = mcp_create_server()
                     ^^^^^^^^^^^^^^^^^^^

.../aignostics/utils/mcp_test.py:131: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

server_name = 'Central Aignostics MCP Server'

    def mcp_create_server(server_name: str = MCP_SERVER_NAME) -> FastMCP:
        """Create and configure the MCP server with all discovered plugins mounted.
    
        Creates a new FastMCP server instance and mounts all discovered MCP servers
        from the SDK and plugins. Each mounted server's tools are namespaced
        automatically using FastMCP's built-in namespace feature.
    
        Args:
            server_name: Human-readable name for the MCP server.
    
        Returns:
            FastMCP: Configured MCP server ready to run.
        """
        mcp = FastMCP(name=server_name, version=__version__)
    
        # Mount discovered servers
        servers = mcp_discover_servers()
        seen_names: set[str] = set()
        count = 0
    
        for server in servers:
            if server is not mcp:  # Don't mount self
                if server.name in seen_names:
                    logger.warning(f"Duplicate MCP server name '{server.name}' - skipping to avoid tool collision")
                    continue
                seen_names.add(server.name)
                logger.info(f"Mounting MCP server: {server.name}")
>               mcp.mount(server, namespace=server.name)
E               TypeError: FastMCP.mount() got an unexpected keyword argument 'namespace'

.../aignostics/utils/_mcp.py:79: TypeError
tests.aignostics.utils.mcp_test::test_mcp_list_tools_empty

Flake rate in main: 100.00% (Passed 0 times, Failed 1 times)

Stack Traces | 0.009s run time
record_property = <function record_property.<locals>.append_property at 0x7f9763ac5d20>

    @pytest.mark.unit
    def test_mcp_list_tools_empty(record_property) -> None:
        """Test mcp_list_tools with no discovered tools."""
        record_property("tested-item-id", "SPEC-UTILS-SERVICE")
        with patch(PATCH_LOCATE_IMPLEMENTATIONS, return_value=[]):
>           tools = mcp_list_tools()
                    ^^^^^^^^^^^^^^^^

.../aignostics/utils/mcp_test.py:175: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

server_name = 'Central Aignostics MCP Server'

    def mcp_list_tools(server_name: str = MCP_SERVER_NAME) -> list[dict[str, Any]]:
        """List all available MCP tools.
    
        Creates the server and returns information about all registered tools
        including those from mounted servers.
    
        Note:
            This function must be called from a synchronous context. Calling it
            from within an async function will raise RuntimeError.
    
        Args:
            server_name: Human-readable name for the MCP server.
    
        Returns:
            list[dict[str, Any]]: List of tool information dictionaries with
                'name' and 'description' keys.
        """
        server = mcp_create_server(server_name)
        # FastMCP's list_tools() is async because mounted servers may need to
        # lazily initialize resources. We use asyncio.run() to bridge sync/async.
>       tools = asyncio.run(server.list_tools())
                            ^^^^^^^^^^^^^^^^^
E       AttributeError: 'FastMCP' object has no attribute 'list_tools'. Did you mean: 'get_tools'?

.../aignostics/utils/_mcp.py:120: AttributeError
tests.aignostics.utils.mcp_test::test_mcp_list_tools_returns_tool_info

Flake rate in main: 100.00% (Passed 0 times, Failed 1 times)

Stack Traces | 0.014s run time
record_property = <function record_property.<locals>.append_property at 0x7f9763aa64b0>

    @pytest.mark.unit
    def test_mcp_list_tools_returns_tool_info(record_property) -> None:
        """Test that mcp_list_tools returns correct tool information."""
        record_property("tested-item-id", "SPEC-UTILS-SERVICE")
        test_server = FastMCP("test")
    
        @test_server.tool
        def test_tool(param: str) -> str:
            """A test tool."""
            return param
    
        with patch(PATCH_LOCATE_IMPLEMENTATIONS, return_value=[test_server]):
>           tools = mcp_list_tools()
                    ^^^^^^^^^^^^^^^^

.../aignostics/utils/mcp_test.py:164: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../aignostics/utils/_mcp.py:117: in mcp_list_tools
    server = mcp_create_server(server_name)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

server_name = 'Central Aignostics MCP Server'

    def mcp_create_server(server_name: str = MCP_SERVER_NAME) -> FastMCP:
        """Create and configure the MCP server with all discovered plugins mounted.
    
        Creates a new FastMCP server instance and mounts all discovered MCP servers
        from the SDK and plugins. Each mounted server's tools are namespaced
        automatically using FastMCP's built-in namespace feature.
    
        Args:
            server_name: Human-readable name for the MCP server.
    
        Returns:
            FastMCP: Configured MCP server ready to run.
        """
        mcp = FastMCP(name=server_name, version=__version__)
    
        # Mount discovered servers
        servers = mcp_discover_servers()
        seen_names: set[str] = set()
        count = 0
    
        for server in servers:
            if server is not mcp:  # Don't mount self
                if server.name in seen_names:
                    logger.warning(f"Duplicate MCP server name '{server.name}' - skipping to avoid tool collision")
                    continue
                seen_names.add(server.name)
                logger.info(f"Mounting MCP server: {server.name}")
>               mcp.mount(server, namespace=server.name)
E               TypeError: FastMCP.mount() got an unexpected keyword argument 'namespace'

.../aignostics/utils/_mcp.py:79: TypeError

To view more test analytics, go to the Test Analytics Dashboard
📋 Got 3 mins? Take this short survey to help us improve Test Analytics.

@sonarqubecloud
Copy link

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant