| 
 | 1 | +"""Tests for tool annotations in low-level server."""  | 
 | 2 | + | 
 | 3 | +import anyio  | 
 | 4 | +import pytest  | 
 | 5 | + | 
 | 6 | +from mcp.client.session import ClientSession  | 
 | 7 | +from mcp.server import Server  | 
 | 8 | +from mcp.server.lowlevel import NotificationOptions  | 
 | 9 | +from mcp.server.models import InitializationOptions  | 
 | 10 | +from mcp.server.session import ServerSession  | 
 | 11 | +from mcp.shared.session import RequestResponder  | 
 | 12 | +from mcp.types import (  | 
 | 13 | +    ClientResult,  | 
 | 14 | +    JSONRPCMessage,  | 
 | 15 | +    ServerNotification,  | 
 | 16 | +    ServerRequest,  | 
 | 17 | +    Tool,  | 
 | 18 | +    ToolAnnotations,  | 
 | 19 | +)  | 
 | 20 | + | 
 | 21 | + | 
 | 22 | +@pytest.mark.anyio  | 
 | 23 | +async def test_lowlevel_server_tool_annotations():  | 
 | 24 | +    """Test that tool annotations work in low-level server."""  | 
 | 25 | +    server = Server("test")  | 
 | 26 | + | 
 | 27 | +    # Create a tool with annotations  | 
 | 28 | +    @server.list_tools()  | 
 | 29 | +    async def list_tools():  | 
 | 30 | +        return [  | 
 | 31 | +            Tool(  | 
 | 32 | +                name="echo",  | 
 | 33 | +                description="Echo a message back",  | 
 | 34 | +                inputSchema={  | 
 | 35 | +                    "type": "object",  | 
 | 36 | +                    "properties": {  | 
 | 37 | +                        "message": {"type": "string"},  | 
 | 38 | +                    },  | 
 | 39 | +                    "required": ["message"],  | 
 | 40 | +                },  | 
 | 41 | +                annotations=ToolAnnotations(  | 
 | 42 | +                    title="Echo Tool",  | 
 | 43 | +                    readOnlyHint=True,  | 
 | 44 | +                ),  | 
 | 45 | +            )  | 
 | 46 | +        ]  | 
 | 47 | + | 
 | 48 | +    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream[  | 
 | 49 | +        JSONRPCMessage  | 
 | 50 | +    ](10)  | 
 | 51 | +    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream[  | 
 | 52 | +        JSONRPCMessage  | 
 | 53 | +    ](10)  | 
 | 54 | + | 
 | 55 | +    # Message handler for client  | 
 | 56 | +    async def message_handler(  | 
 | 57 | +        message: RequestResponder[ServerRequest, ClientResult]  | 
 | 58 | +        | ServerNotification  | 
 | 59 | +        | Exception,  | 
 | 60 | +    ) -> None:  | 
 | 61 | +        if isinstance(message, Exception):  | 
 | 62 | +            raise message  | 
 | 63 | + | 
 | 64 | +    # Server task  | 
 | 65 | +    async def run_server():  | 
 | 66 | +        async with ServerSession(  | 
 | 67 | +            client_to_server_receive,  | 
 | 68 | +            server_to_client_send,  | 
 | 69 | +            InitializationOptions(  | 
 | 70 | +                server_name="test-server",  | 
 | 71 | +                server_version="1.0.0",  | 
 | 72 | +                capabilities=server.get_capabilities(  | 
 | 73 | +                    notification_options=NotificationOptions(),  | 
 | 74 | +                    experimental_capabilities={},  | 
 | 75 | +                ),  | 
 | 76 | +            ),  | 
 | 77 | +        ) as server_session:  | 
 | 78 | +            async with anyio.create_task_group() as tg:  | 
 | 79 | + | 
 | 80 | +                async def handle_messages():  | 
 | 81 | +                    async for message in server_session.incoming_messages:  | 
 | 82 | +                        await server._handle_message(message, server_session, {}, False)  | 
 | 83 | + | 
 | 84 | +                tg.start_soon(handle_messages)  | 
 | 85 | +                await anyio.sleep_forever()  | 
 | 86 | + | 
 | 87 | +    # Run the test  | 
 | 88 | +    async with anyio.create_task_group() as tg:  | 
 | 89 | +        tg.start_soon(run_server)  | 
 | 90 | + | 
 | 91 | +        async with ClientSession(  | 
 | 92 | +            server_to_client_receive,  | 
 | 93 | +            client_to_server_send,  | 
 | 94 | +            message_handler=message_handler,  | 
 | 95 | +        ) as client_session:  | 
 | 96 | +            # Initialize the session  | 
 | 97 | +            await client_session.initialize()  | 
 | 98 | + | 
 | 99 | +            # List tools  | 
 | 100 | +            tools_result = await client_session.list_tools()  | 
 | 101 | + | 
 | 102 | +            # Cancel the server task  | 
 | 103 | +            tg.cancel_scope.cancel()  | 
 | 104 | + | 
 | 105 | +    # Verify results  | 
 | 106 | +    assert tools_result is not None  | 
 | 107 | +    assert len(tools_result.tools) == 1  | 
 | 108 | +    assert tools_result.tools[0].name == "echo"  | 
 | 109 | +    assert tools_result.tools[0].annotations is not None  | 
 | 110 | +    assert tools_result.tools[0].annotations.title == "Echo Tool"  | 
 | 111 | +    assert tools_result.tools[0].annotations.readOnlyHint is True  | 
0 commit comments