speed_test

Instructions

Tool call template

This tool declares no required parameters. Add optional parameters from the input schema as needed.

{
  "id": 1,
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "arguments": {},
    "name": "speed_test"
  }
}

Implementation Reference

• src/sounding/server.py:527-599 (handler)

  The main handler for the speed_test tool. Downloads a 10 MB test payload from Cloudflare, measures throughput (Mbps) and TCP connect latency (ms). Returns a dict with test_server, latency stats (min/avg/max ms), and download info (speed_mbps, bytes_transferred, duration_ms).

  @mcp.tool()
  async def speed_test() -> dict:
      """Measure network download speed and latency.
  
      Downloads a 10 MB test payload from Cloudflare and measures throughput.
      Also measures latency with multiple TCP connect pings.
      Returns download speed in Mbps and latency stats in milliseconds.
      """
      # ── Latency measurement (5 TCP pings to Cloudflare) ──
      latencies: list[float] = []
      for _ in range(5):
          start = time.perf_counter()
          try:
              reader, writer = await asyncio.wait_for(
                  asyncio.open_connection("speed.cloudflare.com", 443),
                  timeout=5,
              )
              elapsed = (time.perf_counter() - start) * 1000
              latencies.append(elapsed)
              writer.close()
              await writer.wait_closed()
          except (OSError, asyncio.TimeoutError):
              pass
  
      latency_stats: dict = {}
      if latencies:
          avg = sum(latencies) / len(latencies)
          latency_stats = {
              "min_ms": round(min(latencies), 2),
              "avg_ms": round(avg, 2),
              "max_ms": round(max(latencies), 2),
          }
  
      # ── Download speed measurement ──
      test_server, test_url = _SPEED_TEST_URLS[0]
      download_mbps: float | None = None
      download_bytes: int = 0
      download_ms: float = 0.0
      dl_error: str | None = None
  
      try:
          async with httpx.AsyncClient(
              timeout=httpx.Timeout(60.0),
              follow_redirects=True,
          ) as client:
              start = time.perf_counter()
              response = await client.get(test_url)
              download_ms = (time.perf_counter() - start) * 1000
              download_bytes = len(response.content)
  
              if download_bytes > 0 and download_ms > 0:
                  # bits / milliseconds → megabits per second
                  download_mbps = round(
                      (download_bytes * 8) / (download_ms / 1000) / 1_000_000, 2
                  )
      except Exception as exc:
          dl_error = str(exc)
  
      result: dict = {
          "test_server": test_server,
          "latency": latency_stats if latency_stats else {"error": "All pings failed"},
      }
  
      if dl_error:
          result["download"] = {"error": dl_error}
      else:
          result["download"] = {
              "speed_mbps": download_mbps,
              "bytes_transferred": download_bytes,
              "duration_ms": round(download_ms, 2),
          }
  
      return result

• src/sounding/server.py:521-524 (helper)

  Configuration constant defining the speed test server and URL (Cloudflare 10 MB payload) used by the speed_test handler.

  _SPEED_TEST_URLS = [
      # 10 MB payload from Cloudflare
      ("Cloudflare", "https://speed.cloudflare.com/__down?bytes=10000000"),
  ]

• src/sounding/server.py:527-528 (registration)

  The @mcp.tool() decorator registers speed_test as a tool with the MCP server (line 33: mcp = FastMCP('sounding')).

  @mcp.tool()
  async def speed_test() -> dict: