enable_blocking

Instructions

Implementation Reference

• src/pihole_mcp/tools/blocking.py:12-15 (handler)

  The handler function for the 'enable_blocking' tool. Sends a POST request to Pi-hole's /dns/blocking endpoint with {"blocking": true} to enable DNS blocking immediately.

  @mcp.tool()
  async def enable_blocking() -> dict:
      """Enable DNS blocking immediately."""
      return await client.post("/dns/blocking", json={"blocking": True})

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

  The register() function in blocking.py decorates enable_blocking with @mcp.tool(), which registers it as an MCP tool. The module is then imported and invoked via register_all() in tools/__init__.py.

  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)

  register_all() iterates over tool modules (including blocking) and calls module.register(mcp, client), which triggers the @mcp.tool() decorations.

  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:63-111 (helper)

  PiholeClient.request() and the convenience post() method are the helpers that enable_blocking uses to make the HTTP call. request() handles auth, retry on 401, and error handling.

  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)