Function run_tool

tests/shared/test_streamable_http.py:1173–1185

()

                async with anyio.create_task_group() as tg:  # pragma: no branch

                    async def run_tool():
                        metadata = ClientMessageMetadata(
                            on_resumption_token_update=on_resumption_token_update,
                        )
                        await session.send_request(
                            types.CallToolRequest(
                                params=types.CallToolRequestParams(
                                    name="wait_for_lock_with_notification", arguments={}
                                ),
                            ),
                            types.CallToolResult,
                            metadata=metadata,
                        )

                    tg.start_soon(run_tool)