python-sdk/tests/server/fastmcp/test_runtime_tools.py at eaec20aa96650811754d8b5ce97efde4c19814c1 · NikBellini/python-sdk · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
"""Integration tests for runtime tools functionality."""

import pytest

from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.server import Context
from mcp.server.fastmcp.tools.base import Tool
from mcp.shared.memory import create_connected_server_and_client_session
from mcp.types import TextContent


@pytest.mark.anyio
async def test_runtime_tools():
    """Test that runtime tools work correctly."""

    async def runtime_mcp_tools_generator(ctx: Context) -> list[Tool]:
        """Generate runtime tools."""

        def runtime_tool_1(message: str):
            return message

        def runtime_tool_2(message: str):
            return message

        def runtime_tool_3(message: str):
            return message

        tools = [Tool.from_function(runtime_tool_1), Tool.from_function(runtime_tool_2)]

        # Tool added only after authorization
        request = ctx.request_context.request
        if request and request.header.get("Authorization") == "Bearer test_auth":
            tools.append(Tool.from_function(runtime_tool_3))

        return tools

    # Create server with various tool configurations, both static and runtime
    mcp = FastMCP(name="RuntimeToolsTestServer", runtime_mcp_tools_generator=runtime_mcp_tools_generator)

    # Static tool
    @mcp.tool(description="Static tool")
    def static_tool(message: str) -> str:
        return message

    # Start server and connect client without authorization
    async with create_connected_server_and_client_session(mcp._mcp_server) as client:
        await client.initialize()

        # List tools
        tools_result = await client.list_tools()
        tool_names = {tool.name: tool for tool in tools_result.tools}

        # Verify both tools
        assert "static_tool" in tool_names
        assert "runtime_tool_1" in tool_names
        assert "runtime_tool_2" in tool_names

        # Check static tool
        result = await client.call_tool("static_tool", {"message": "This is a test"})
        assert len(result.content) == 1
        content = result.content[0]
        assert isinstance(content, TextContent)
        assert content.text == "This is a test"

        # Check runtime tool 1
        result = await client.call_tool("runtime_tool_1", {"message": "This is a test"})
        assert len(result.content) == 1
        content = result.content[0]
        assert isinstance(content, TextContent)
        assert content.text == "This is a test"

        # Check runtime tool 2
        result = await client.call_tool("runtime_tool_2", {"message": "This is a test"})
        assert len(result.content) == 1
        content = result.content[0]
        assert isinstance(content, TextContent)
        assert content.text == "This is a test"

        # Check non existing tool
        result = await client.call_tool("non_existing_tool", {"message": "This is a test"})
        assert len(result.content) == 1
        content = result.content[0]
        assert isinstance(content, TextContent)
        assert content.text == "Unknown tool: non_existing_tool"

        # Check not authorized tool
        result = await client.call_tool("runtime_tool_3", {"message": "This is a test"})
        assert len(result.content) == 1
        content = result.content[0]
        assert isinstance(content, TextContent)
        assert content.text == "Unknown tool: runtime_tool_3"