get_query_types

Instructions

Implementation Reference

• src/pihole_mcp/tools/stats.py:31-33 (handler)

  The get_query_types tool handler: an async function decorated with @mcp.tool() that calls the Pi-hole API endpoint /stats/query_types to return a breakdown of DNS query types (A, AAAA, PTR, SRV, etc).

  async def get_query_types() -> dict:
      """Breakdown of DNS query types (A, AAAA, PTR, SRV, etc)."""
      return await client.get("/stats/query_types")

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

  The register() function in stats.py uses @mcp.tool() decorator to register get_query_types (and other tools) with the FastMCP server instance.

  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 orchestrates registration by calling stats.register(mcp, client), which registers get_query_types via the decorator.

  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-105 (helper)

  The PiholeClient.get() method used by get_query_types to make the 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)
  
  async def post(self, path: str, *, json: Any | None = None) -> Any:
      return await self.request("POST", path, json=json)