purge_cache

Instructions

Implementation Reference

• src/cloudflare_mcp_server/__init__.py:234-263 (registration)

  Tool registration for 'purge_cache' in the list_tools handler. Defines schema with zone_id (required), purge_everything (boolean), files, tags, and hosts (arrays of strings).

  Tool(
      name="purge_cache",
      description="Purge Cloudflare's cache for a zone. Can purge everything or specific files/tags/hosts.",
      inputSchema={
          "type": "object",
          "properties": {
              "zone_id": {"type": "string", "description": "The zone ID"},
              "purge_everything": {
                  "type": "boolean",
                  "description": "Purge all cached content (use cautiously!)",
              },
              "files": {
                  "type": "array",
                  "items": {"type": "string"},
                  "description": "Array of URLs to purge",
              },
              "tags": {
                  "type": "array",
                  "items": {"type": "string"},
                  "description": "Array of cache tags to purge",
              },
              "hosts": {
                  "type": "array",
                  "items": {"type": "string"},
                  "description": "Array of hosts to purge",
              },
          },
          "required": ["zone_id"],
      },
  ),

• src/cloudflare_mcp_server/__init__.py:237-262 (schema)

  Input schema for purge_cache tool: requires zone_id, accepts purge_everything (boolean), and optional arrays files, tags, hosts.

  inputSchema={
      "type": "object",
      "properties": {
          "zone_id": {"type": "string", "description": "The zone ID"},
          "purge_everything": {
              "type": "boolean",
              "description": "Purge all cached content (use cautiously!)",
          },
          "files": {
              "type": "array",
              "items": {"type": "string"},
              "description": "Array of URLs to purge",
          },
          "tags": {
              "type": "array",
              "items": {"type": "string"},
              "description": "Array of cache tags to purge",
          },
          "hosts": {
              "type": "array",
              "items": {"type": "string"},
              "description": "Array of hosts to purge",
          },
      },
      "required": ["zone_id"],
  },

• src/cloudflare_mcp_server/__init__.py:527-543 (handler)

  Handler function for purge_cache tool. Builds data payload based on purge_everything flag (preferred, purges all) or selective files/tags/hosts arrays, then POSTs to Cloudflare API /zones/{zone_id}/purge_cache.

  async def _purge_cache(self, args: dict) -> Any:
      """Purge cache."""
      data = {}
  
      if args.get("purge_everything"):
          data["purge_everything"] = True
      else:
          if args.get("files"):
              data["files"] = args["files"]
          if args.get("tags"):
              data["tags"] = args["tags"]
          if args.get("hosts"):
              data["hosts"] = args["hosts"]
  
      return await self._make_request(
          f"/zones/{args['zone_id']}/purge_cache", method="POST", data=data
      )

• src/cloudflare_mcp_server/__init__.py:418-419 (registration)

  Routing: call_tool dispatcher maps 'purge_cache' name to _purge_cache handler method.

  elif name == "purge_cache":
      result = await self._purge_cache(arguments)

• src/cloudflare_mcp_server/__init__.py:27-60 (helper)

  Helper _make_request used by _purge_cache to make HTTP calls to the Cloudflare API. Handles auth, JSON serialization, error checking, and result extraction.

  async def _make_request(
      self,
      endpoint: str,
      method: str = "GET",
      data: Optional[dict] = None,
      params: Optional[dict] = None,
  ) -> Any:
      """Make a request to the Cloudflare API."""
      url = f"{CLOUDFLARE_API_BASE}{endpoint}"
      headers = {
          "Authorization": f"Bearer {self.api_token}",
          "Content-Type": "application/json",
      }
  
      try:
          response = await self.client.request(
              method=method,
              url=url,
              json=data,
              params=params,
              headers=headers,
          )
          response.raise_for_status()
          result = response.json()
  
          if not result.get("success"):
              errors = result.get("errors", [])
              raise Exception(f"Cloudflare API error: {json.dumps(errors)}")
  
          return result.get("result")
  
      except httpx.HTTPError as e:
          raise Exception(f"HTTP error occurred: {str(e)}")