Function _handle_call_tool

tests/shared/test_streamable_http.py:209–316

(ctx: ServerRequestContext[ServerState], params: CallToolRequestParams)

async def _handle_call_tool(ctx: ServerRequestContext[ServerState], params: CallToolRequestParams) -> CallToolResult:
    name = params.name

    # When the tool is called, send a notification to test GET stream
    if name == "test_tool_with_standalone_notification":
        await ctx.session.send_resource_updated(uri="http://test_resource")
        return CallToolResult(content=[TextContent(type="text", text=f"Called {name}")])

    elif name == "test_sampling_tool":
        sampling_result = await ctx.session.create_message(  # pyright: ignore[reportDeprecated]
            messages=[
                types.SamplingMessage(
                    role="user",
                    content=types.TextContent(type="text", text="Server needs client sampling"),
                )
            ],
            max_tokens=100,
            related_request_id=ctx.request_id,
        )

        assert sampling_result.content.type == "text"
        return CallToolResult(
            content=[
                TextContent(
                    type="text",
                    text=f"Response from sampling: {sampling_result.content.text}",
                )
            ]
        )

    elif name == "wait_for_lock_with_notification":
        await ctx.session.send_log_message(  # pyright: ignore[reportDeprecated]
            level="info",
            data="First notification before lock",
            logger="lock_tool",
            related_request_id=ctx.request_id,
        )

        await ctx.lifespan_context.lock.wait()

        await ctx.session.send_log_message(  # pyright: ignore[reportDeprecated]
            level="info",
            data="Second notification after lock",
            logger="lock_tool",
            related_request_id=ctx.request_id,
        )

        return CallToolResult(content=[TextContent(type="text", text="Completed")])

    elif name == "release_lock":
        ctx.lifespan_context.lock.set()
        return CallToolResult(content=[TextContent(type="text", text="Lock released")])

    elif name == "tool_with_stream_close":
        await ctx.session.send_log_message(  # pyright: ignore[reportDeprecated]
            level="info",
            data="Before close",
            logger="stream_close_tool",