get_top_clients

Instructions

Implementation Reference

• src/pihole_mcp/tools/stats.py:22-28 (handler)

  The handler function for the 'get_top_clients' tool. It fetches top clients by query count from /stats/top_clients Pi-hole API endpoint. Accepts 'count' (default 10) and 'blocked' (default False) parameters.

  @mcp.tool()
  async def get_top_clients(count: int = 10, blocked: bool = False) -> dict:
      """Get top clients by query count. Set blocked=true for top clients by blocked query count."""
      params: dict = {"count": count}
      if blocked:
          params["blocked"] = "true"
      return await client.get("/stats/top_clients", params=params)

• src/pihole_mcp/tools/stats.py:6-50 (registration)

  The 'register' function defines all stat tools including 'get_top_clients' via @mcp.tool() decorator. The decorator registers the tool with the FastMCP server.

  def register(mcp: FastMCP, client: PiholeClient) -> int:
      @mcp.tool()
      async def get_stats() -> dict:
          """Get summary statistics (total queries, blocked queries, blocking percentage, unique domains, clients)."""
          return await client.get("/stats/summary")
  
      @mcp.tool()
      async def get_top_blocked(count: int = 10) -> dict:
          """Get top blocked domains by query count."""
          return await client.get("/stats/top_domains", params={"blocked": "true", "count": count})
  
      @mcp.tool()
      async def get_top_permitted(count: int = 10) -> dict:
          """Get top allowed (permitted) domains by query count."""
          return await client.get("/stats/top_domains", params={"blocked": "false", "count": count})
  
      @mcp.tool()
      async def get_top_clients(count: int = 10, blocked: bool = False) -> dict:
          """Get top clients by query count. Set blocked=true for top clients by blocked query count."""
          params: dict = {"count": count}
          if blocked:
              params["blocked"] = "true"
          return await client.get("/stats/top_clients", params=params)
  
      @mcp.tool()
      async def get_query_types() -> dict:
          """Breakdown of DNS query types (A, AAAA, PTR, SRV, etc)."""
          return await client.get("/stats/query_types")
  
      @mcp.tool()
      async def get_forward_destinations() -> dict:
          """Upstream DNS server stats (which forwarders served how many queries)."""
          return await client.get("/stats/upstreams")
  
      @mcp.tool()
      async def get_recent_blocked(count: int = 10) -> dict:
          """Recently blocked domains."""
          return await client.get("/stats/recent_blocked", params={"count": count})
  
      @mcp.tool()
      async def get_history() -> dict:
          """Time-series activity graph: timestamps, allowed/blocked/other counts per bucket."""
          return await client.get("/history")
  
      return 8

• src/pihole_mcp/tools/__init__.py:14-19 (registration)

  The 'register_all' function calls stats.register(mcp, client) which triggers the registration of get_top_clients and all other stat tools.

  def register_all(mcp: FastMCP, client: PiholeClient) -> int:
      """Register every tool module against the FastMCP instance. Returns tool count."""
      count = 0
      for module in (stats, queries, blocking, domains, local_dns, maintenance):
          count += module.register(mcp, client)
      return count

• src/pihole_mcp/client.py:100-101 (helper)

  The PiholeClient.get() method used by the handler to make the actual HTTP GET request to the Pi-hole API.

  async def get(self, path: str, *, params: dict[str, Any] | None = None) -> Any:
      return await self.request("GET", path, params=params)