get_forward_destinations

Instructions

Implementation Reference

• src/pihole_mcp/tools/stats.py:35-38 (handler)

  The handler/implementation of the get_forward_destinations tool — makes a GET request to /stats/upstreams via the Pi-hole client.

  @mcp.tool()
  async def get_forward_destinations() -> dict:
      """Upstream DNS server stats (which forwarders served how many queries)."""
      return await client.get("/stats/upstreams")

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

  The register() function decorates get_forward_destinations with @mcp.tool(), registering it as an MCP tool. stats.py contributes 8 tools total.

  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)

  Top-level registration orchestrator — calls stats.register(mcp, client) which registers get_forward_destinations.

  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 that triggers registration of all tools including get_forward_destinations.

  _tool_count = register_all(mcp, _client)

• src/pihole_mcp/client.py:10-111 (helper)

  The PiholeClient.get() helper called by get_forward_destinations to issue the HTTP GET to /stats/upstreams.

  class PiholeClient:
      """Async HTTP client for Pi-hole v6 REST API with session auth and auto-refresh."""
  
      _REFRESH_BUFFER_SECONDS = 60
  
      def __init__(self, config: PiholeConfig):
          self._config = config
          self._http = httpx.AsyncClient(
              base_url=config.api_base,
              verify=config.verify_tls,
              timeout=config.timeout_seconds,
          )
          self._sid: str | None = None
          self._sid_expires_at: float = 0.0
  
      async def close(self) -> None:
          if self._sid:
              try:
                  await self._http.delete(
                      "/auth",
                      headers={"X-FTL-SID": self._sid},
                  )
              except Exception:
                  pass
              self._sid = None
          await self._http.aclose()
  
      async def _authenticate(self) -> None:
          resp = await self._http.post(
              "/auth",
              json={"password": self._config.password},
          )
          if resp.status_code != 200:
              raise PiholeAuthError(
                  f"Authentication failed: HTTP {resp.status_code} {resp.text}"
              )
          payload = resp.json()
          session = payload.get("session") or {}
          sid = session.get("sid")
          valid = session.get("valid") or session.get("validity")
          if not sid:
              raise PiholeAuthError(f"No session SID in auth response: {payload}")
          self._sid = sid
          validity_seconds = int(valid) if valid else 300
          self._sid_expires_at = time.time() + validity_seconds
  
      async def _ensure_session(self) -> str:
          now = time.time()
          if not self._sid or now >= (self._sid_expires_at - self._REFRESH_BUFFER_SECONDS):
              await self._authenticate()
          assert self._sid
          return self._sid
  
      async def request(
          self,
          method: str,
          path: str,
          *,
          params: dict[str, Any] | None = None,
          json: Any | None = None,
      ) -> Any:
          """Issue a request, auto-authenticating and retrying once on 401."""
          sid = await self._ensure_session()
          resp = await self._http.request(
              method,
              path,
              params=params,
              json=json,
              headers={"X-FTL-SID": sid},
          )
          if resp.status_code == 401:
              self._sid = None
              sid = await self._ensure_session()
              resp = await self._http.request(
                  method,
                  path,
                  params=params,
                  json=json,
                  headers={"X-FTL-SID": sid},
              )
          if resp.status_code >= 400:
              try:
                  body = resp.json()
              except ValueError:
                  body = resp.text
              raise PiholeAPIError(resp.status_code, f"{method} {path} failed", body)
          if resp.status_code == 204 or not resp.content:
              return None
          return resp.json()
  
      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)
  
      async def patch(self, path: str, *, json: Any | None = None) -> Any:
          return await self.request("PATCH", path, json=json)
  
      async def delete(self, path: str) -> Any:
          return await self.request("DELETE", path)