ping

Instructions

Implementation Reference

• src/sounding/server.py:60-107 (handler)

  The async `ping` handler function decorated with `@mcp.tool()`. It pings a host using TCP connect (port 80) as a non-root ICMP alternative. Validates host via `validate_host`, runs `count` async TCP connections, measures latency in ms, and returns min/avg/max/jitter plus packet loss stats.

  @mcp.tool()
  async def ping(host: str, count: int = 4, timeout: int = 5) -> dict:
      """Ping a host using TCP connect (port 80) as a non-root ICMP alternative.
  
      Returns min/avg/max/jitter latency in milliseconds.
      """
      host = validate_host(host)
      if count < 1 or count > 100:
          raise ValueError("count must be between 1 and 100")
  
      latencies: list[float] = []
      errors: list[str] = []
  
      for _ in range(count):
          start = time.perf_counter()
          try:
              reader, writer = await asyncio.wait_for(
                  asyncio.open_connection(host, 80),
                  timeout=timeout,
              )
              elapsed = (time.perf_counter() - start) * 1000  # ms
              latencies.append(elapsed)
              writer.close()
              await writer.wait_closed()
          except (OSError, asyncio.TimeoutError) as exc:
              errors.append(str(exc))
  
      if not latencies:
          return {"host": host, "success": False, "errors": errors}
  
      avg = sum(latencies) / len(latencies)
      jitter = (
          (sum(abs(l - avg) for l in latencies) / len(latencies))
          if len(latencies) > 1
          else 0.0
      )
  
      return {
          "host": host,
          "success": True,
          "packets_sent": count,
          "packets_received": len(latencies),
          "packet_loss_pct": round((1 - len(latencies) / count) * 100, 1),
          "min_ms": round(min(latencies), 2),
          "avg_ms": round(avg, 2),
          "max_ms": round(max(latencies), 2),
          "jitter_ms": round(jitter, 2),
      }

• src/sounding/server.py:57-58 (registration)

  The `@mcp.tool()` decorator on line 60 registers `ping` as an MCP tool. Comment section header marks this as tool #2 (ping).

  # 2. ping
  # ---------------------------------------------------------------------------

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

  Import of `Optional` from typing, used in function signatures across the file (though not directly used in ping).

  from typing import Optional

• src/sounding/validators.py:101-143 (helper)

  `validate_host` is called by the ping handler to sanitize/validate the host argument. It strips whitespace, rejects shell metacharacters, validates IP/hostname format, and optionally blocks internal IPs (but `allow_internal=True` by default, so ping can target internal networks).

  def validate_host(host: str, *, allow_internal: bool = True) -> str:
      """Validate a hostname or IP address.
  
      Args:
          host: The hostname or IP to validate.
          allow_internal: If False, reject internal/private/loopback IPs and
              hostnames that resolve to them (SSRF protection). Defaults to
              True since tools like ping and traceroute legitimately target
              internal networks.
  
      Returns the cleaned host string.
      Raises ``ValueError`` on anything suspicious.
      """
      host = host.strip()
      if not host:
          raise ValueError("Host must not be empty")
  
      if _SHELL_META.search(host):
          raise ValueError(f"Host contains forbidden characters: {host!r}")
  
      # Accept valid IP addresses directly.
      try:
          addr = ipaddress.ip_address(host)
          if not allow_internal and is_internal_ip(host):
              raise ValueError(
                  f"Host {host} is an internal/private/loopback address — "
                  "requests to internal addresses are blocked"
              )
          return host
      except ValueError as exc:
          # Re-raise if it's our own SSRF block, not an ip_address parse error
          if "internal" in str(exc):
              raise
          pass
  
      if not _HOSTNAME_RE.match(host):
          raise ValueError(f"Invalid hostname: {host!r}")
  
      # If internal not allowed, resolve and check the IP
      if not allow_internal:
          _resolve_and_check(host)
  
      return host

• tests/test_tools.py:43-61 (helper)

  Test cases for the ping tool: `test_ping_localhost` and `test_ping_web_target` verify the handler works end-to-end. Boundary tests at lines 275-289 validate count range enforcement.

  # ── ping ───────────────────────────────────────────────────────────────────
  
  
  @pytest.mark.asyncio
  async def test_ping_localhost():
      """Ping localhost — TCP connect to port 80 may fail, but the call
      should not raise."""
      result = await ping("localhost", count=2, timeout=2)
      assert result["host"] == "localhost"
      assert "success" in result
  
  
  @pytest.mark.asyncio
  async def test_ping_web_target(web_target):
      """Ping the Docker web target — port 80 is open."""
      host, _port = web_target.split(":")
      result = await ping(host, count=2, timeout=5)
      assert result["host"] == host