disable_blocking

Instructions

Implementation Reference

• src/pihole_mcp/tools/blocking.py:17-23 (handler)

  The 'disable_blocking' tool handler function. It is an async function registered as an MCP tool that sends a POST to /dns/blocking with {"blocking": false} and optionally includes a "timer" field if duration_seconds is provided.

  @mcp.tool()
  async def disable_blocking(duration_seconds: int | None = None) -> dict:
      """Disable DNS blocking. If duration_seconds is given, re-enables after that timer; otherwise indefinite."""
      body: dict = {"blocking": False}
      if duration_seconds is not None:
          body["timer"] = duration_seconds
      return await client.post("/dns/blocking", json=body)

• src/pihole_mcp/tools/blocking.py:6-25 (registration)

  The 'register' function that registers all blocking tools (including disable_blocking) onto the FastMCP instance via the @mcp.tool() decorator. Returns 3 (the count of blocking tools).

  def register(mcp: FastMCP, client: PiholeClient) -> int:
      @mcp.tool()
      async def get_blocking_status() -> dict:
          """Check if Pi-hole DNS blocking is enabled. Returns status and any active timer."""
          return await client.get("/dns/blocking")
  
      @mcp.tool()
      async def enable_blocking() -> dict:
          """Enable DNS blocking immediately."""
          return await client.post("/dns/blocking", json={"blocking": True})
  
      @mcp.tool()
      async def disable_blocking(duration_seconds: int | None = None) -> dict:
          """Disable DNS blocking. If duration_seconds is given, re-enables after that timer; otherwise indefinite."""
          body: dict = {"blocking": False}
          if duration_seconds is not None:
              body["timer"] = duration_seconds
          return await client.post("/dns/blocking", json=body)
  
      return 3

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

  The 'register_all' function that aggregates tool registration by calling each module's register() function, including the blocking module.

  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)

  The top-level call that triggers registration of all tools (including disable_blocking) by calling register_all().

  _tool_count = register_all(mcp, _client)

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

  The PiholeClient 'request'/'post' helper methods used by the disable_blocking handler to issue the actual HTTP POST request to the Pi-hole API.

  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)