get_query_log

Instructions

Implementation Reference

• src/pihole_mcp/tools/queries.py:7-40 (handler)

  The main handler function for 'get_query_log' tool. Accepts filters (length, from_ts, until_ts, domain, client_ip, upstream, cursor) and calls client.get('/queries', params=params) to fetch DNS query data from Pi-hole API.

  @mcp.tool()
  async def get_query_log(
      length: int = 100,
      from_ts: int | None = None,
      until_ts: int | None = None,
      domain: str | None = None,
      client_ip: str | None = None,
      upstream: str | None = None,
      cursor: str | None = None,
  ) -> dict:
      """Fetch recent DNS queries with optional filters.
  
      Filters:
      - length: number of queries (default 100)
      - from_ts / until_ts: Unix timestamps
      - domain: exact or wildcard (*) domain match
      - client_ip: exact or wildcard (*) client IP match
      - upstream: one of 'cache', 'blocklist', 'permitted'
      - cursor: pagination cursor from previous response
      """
      params: dict = {"length": length}
      if from_ts is not None:
          params["from"] = from_ts
      if until_ts is not None:
          params["until"] = until_ts
      if domain:
          params["domain"] = domain
      if client_ip:
          params["client"] = client_ip
      if upstream:
          params["upstream"] = upstream
      if cursor:
          params["cursor"] = cursor
      return await client.get("/queries", params=params)

• src/pihole_mcp/tools/queries.py:6-47 (registration)

  Registration function that registers the tool via @mcp.tool() decorator on the FastMCP instance. Called from register_all() in __init__.py.

  def register(mcp: FastMCP, client: PiholeClient) -> int:
      @mcp.tool()
      async def get_query_log(
          length: int = 100,
          from_ts: int | None = None,
          until_ts: int | None = None,
          domain: str | None = None,
          client_ip: str | None = None,
          upstream: str | None = None,
          cursor: str | None = None,
      ) -> dict:
          """Fetch recent DNS queries with optional filters.
  
          Filters:
          - length: number of queries (default 100)
          - from_ts / until_ts: Unix timestamps
          - domain: exact or wildcard (*) domain match
          - client_ip: exact or wildcard (*) client IP match
          - upstream: one of 'cache', 'blocklist', 'permitted'
          - cursor: pagination cursor from previous response
          """
          params: dict = {"length": length}
          if from_ts is not None:
              params["from"] = from_ts
          if until_ts is not None:
              params["until"] = until_ts
          if domain:
              params["domain"] = domain
          if client_ip:
              params["client"] = client_ip
          if upstream:
              params["upstream"] = upstream
          if cursor:
              params["cursor"] = cursor
          return await client.get("/queries", params=params)
  
      @mcp.tool()
      async def get_query_suggestions() -> dict:
          """Get available filter values for the query log (domains, clients, upstreams, types, statuses)."""
          return await client.get("/queries/suggestions")
  
      return 2

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

  Top-level registration loop that calls queries.register(mcp, client) to register the get_query_log tool.

  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/server.py:15-15 (registration)

  Server entry point calls register_all(mcp, _client) to register all tools including get_query_log.

  _tool_count = register_all(mcp, _client)

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

  The client.get() helper method that the handler delegates to for making the GET request to /queries.

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