Gelbooru MCP

Instructions

Implementation Reference

• gelbooru_mcp.py:288-329 (registration)

  Tool registration for search_posts with name, description, and inputSchema defining tags, limit, pid, id, and cid parameters

  Tool(
      name="search_posts",
      description=(
          "Search Gelbooru posts by tags, page, limit, or ID. "
          "Supports all Gelbooru tag syntax: AND (tag1 tag2), OR ({t1~t2}), "
          "NOT (-tag), wildcards (*tag / tag*), meta-tags like "
          "rating:safe/questionable/explicit, score:>=N, width:>=N, "
          "user:name, sort:random, sort:score:desc, etc."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "tags": {
                  "type": "string",
                  "description": (
                      "Tag query string. Examples: 'cat_ears blue_eyes', "
                      "'touhou -rating:explicit', 'score:>=50 sort:score:desc'"
                  ),
              },
              "limit": {
                  "type": "integer",
                  "description": "Number of posts to return (default 20, max 100).",
                  "default": 20,
                  "minimum": 1,
                  "maximum": 100,
              },
              "pid": {
                  "type": "integer",
                  "description": "Page number (0-indexed).",
                  "default": 0,
              },
              "id": {
                  "type": "integer",
                  "description": "Fetch a single post by its Gelbooru ID.",
              },
              "cid": {
                  "type": "integer",
                  "description": "Fetch posts by change ID (Unix timestamp).",
              },
          },
      },
  ),

• gelbooru_mcp.py:503-515 (handler)

  Handler for search_posts that constructs API parameters from arguments and calls the _get helper function to fetch results

  if name == "search_posts":
      params = {"page": "dapi", "s": "post", "q": "index"}
      if "tags" in arguments:
          params["tags"] = arguments["tags"]
      if "limit" in arguments:
          params["limit"] = arguments["limit"]
      if "pid" in arguments:
          params["pid"] = arguments["pid"]
      if "id" in arguments:
          params["id"] = arguments["id"]
      if "cid" in arguments:
          params["cid"] = arguments["cid"]
      result = await loop.run_in_executor(None, _get, params)

• gelbooru_mcp.py:44-59 (helper)

  Helper function _get that performs synchronous HTTP GET requests to the Gelbooru API, injects credentials, and returns parsed JSON

  def _get(params: dict) -> Any:
      """Perform a synchronous HTTP GET and return parsed JSON."""
      params = {**params, "json": "1"}   # copy — never mutate the caller's dict
      _build_auth(params)
      url = f"{BASE_URL}?{urlencode(params)}"
      req = Request(url, headers={"User-Agent": "GelbooruMCP/1.0"})
      try:
          with urlopen(req, timeout=15) as resp:
              raw = resp.read().decode("utf-8")
      except URLError as exc:
          return {"error": str(exc)}
      try:
          return json.loads(raw)
      except json.JSONDecodeError:
          # Some endpoints return XML/empty on error; surface the raw text
          return {"raw": raw}

• gelbooru_mcp.py:33-41 (helper)

  Helper function _build_auth that injects API credentials from environment variables (GELBOORU_API_KEY and GELBOORU_USER_ID) into request parameters

  def _build_auth(params: dict) -> dict:
      """Inject API credentials from environment variables if present."""
      api_key = os.getenv("GELBOORU_API_KEY")
      user_id = os.getenv("GELBOORU_USER_ID")
      if api_key:
          params["api_key"] = api_key
      if user_id:
          params["user_id"] = user_id
      return params

• gelbooru_mcp.py:499-561 (handler)

  Main call_tool handler that dispatches to specific tool implementations based on the tool name, including search_posts, and returns results as TextContent

  @server.call_tool()
  async def call_tool(name: str, arguments: dict) -> list[TextContent]:
      loop = asyncio.get_event_loop()
  
      if name == "search_posts":
          params = {"page": "dapi", "s": "post", "q": "index"}
          if "tags" in arguments:
              params["tags"] = arguments["tags"]
          if "limit" in arguments:
              params["limit"] = arguments["limit"]
          if "pid" in arguments:
              params["pid"] = arguments["pid"]
          if "id" in arguments:
              params["id"] = arguments["id"]
          if "cid" in arguments:
              params["cid"] = arguments["cid"]
          result = await loop.run_in_executor(None, _get, params)
  
      elif name == "get_deleted_posts":
          params = {"page": "dapi", "s": "post", "q": "index", "deleted": "show"}
          if "last_id" in arguments:
              params["last_id"] = arguments["last_id"]
          if "limit" in arguments:
              params["limit"] = arguments["limit"]
          result = await loop.run_in_executor(None, _get, params)
  
      elif name == "search_tags":
          params = {"page": "dapi", "s": "tag", "q": "index"}
          for key in ("name", "names", "name_pattern", "id", "after_id", "limit", "order", "orderby"):
              if key in arguments:
                  params[key] = arguments[key]
          result = await loop.run_in_executor(None, _get, params)
  
      elif name == "search_users":
          params = {"page": "dapi", "s": "user", "q": "index"}
          for key in ("name", "name_pattern", "limit", "pid"):
              if key in arguments:
                  params[key] = arguments[key]
          result = await loop.run_in_executor(None, _get, params)
  
      elif name == "get_comments":
          params = {"page": "dapi", "s": "comment", "q": "index", "post_id": arguments["post_id"]}
          result = await loop.run_in_executor(None, _get, params)
  
      elif name == "get_character_tags":
          character_name = arguments["character_name"]
          max_images = arguments.get("max_images", 300)
          result = await loop.run_in_executor(
              None, _fetch_character_tags, character_name, max_images
          )
  
      elif name == "build_prompt":
          character_name = arguments["character_name"]
          max_images = arguments.get("max_images", 300)
          include_other = arguments.get("include_other", True)
          result = await loop.run_in_executor(
              None, _build_prompt, character_name, max_images, include_other
          )
  
      else:
          result = {"error": f"Unknown tool: {name}"}
  
      return [TextContent(type="text", text=json.dumps(result, indent=2))]