AstrBot MCP

Instructions

Implementation Reference

• astrbot_mcp/tools/plugin_market_tools.py:106-211 (handler)

  The primary handler function implementing the 'browse_plugin_market' tool. It fetches plugins from AstrBot client or fallback registry, supports 'latest' and 'search' modes, sorts by update time and stars, and paginates results.

  async def browse_plugin_market(
      mode: Literal["latest", "search"] = "latest",
      query: str | None = None,
      start: int = 1,
      count: int = 20,
      custom_registry: str | None = None,
      force_refresh: bool = False,
  ) -> Dict[str, Any]:
      """
      查看 AstrBot 插件市场（支持搜索与按时间排序）。
  
      用法：
        - mode="latest": 按 `updated_at` 倒序，返回第 start ~ start+count-1 条（start 从 1 开始）
        - mode="search": 按 query 搜索（名称/简介/标签/作者/仓库），再按 `updated_at` 倒序
  
      返回字段：
        - total_plugins: 插件市场总数（未过滤）
        - matched_plugins: 搜索命中数（mode=search）
        - plugins: 列表（包含 name/desc/tags/stars/updated_at）
      """
      if start < 1:
          return {"status": "error", "message": "start must be >= 1"}
      if count < 1 or count > 200:
          return {"status": "error", "message": "count must be in [1, 200]"}
      if mode not in ("latest", "search"):
          return {"status": "error", "message": "mode must be 'latest' or 'search'"}
      if mode == "search" and not (query or "").strip():
          return {"status": "error", "message": "query is required when mode='search'"}
  
      client = AstrBotClient.from_env()
      source = "astrbot"
      raw_data: Any | None = None
  
      try:
          result = await client.get_plugin_market_list(
              custom_registry=custom_registry,
              force_refresh=force_refresh,
          )
          if result.get("status") != "ok":
              return {
                  "status": result.get("status") or "error",
                  "message": result.get("message") or "AstrBot returned non-ok status.",
                  "raw": result,
              }
          raw_data = (result.get("data") or {})
      except Exception as e:
          source = "remote"
          try:
              raw_data = await _fetch_default_registry(timeout=client.timeout)
          except Exception as fallback_exc:
              return {
                  "status": "error",
                  "message": f"AstrBot API error: {e.response.status_code if hasattr(e, 'response') else 'Unknown'}",
                  "base_url": client.base_url,
                  "detail": _httpx_error_detail(e),
                  "hint": _astrbot_connect_hint(client),
                  "fallback_error": str(fallback_exc),
              }
  
      items = _normalize_plugin_items(raw_data)
      total = len(items)
  
      if mode == "search":
          items = [it for it in items if _matches_query(it, query or "")]
  
      def sort_key(it: Dict[str, Any]) -> Tuple[datetime, int, str]:
          dt = _parse_iso_datetime(it.get("updated_at") or it.get("update_time") or it.get("updated"))
          stars = _as_int(it.get("stars") or it.get("star") or 0)
          pid = str(it.get("id") or "")
          return (dt, stars, pid)
  
      items.sort(key=sort_key, reverse=True)
  
      matched = len(items)
      offset = start - 1
      page = items[offset : offset + count]
  
      plugins: List[Dict[str, Any]] = []
      for idx, it in enumerate(page, start=start):
          plugin_id = str(it.get("id") or "")
          display_name = it.get("display_name") or it.get("name") or plugin_id
          tags = it.get("tags") if isinstance(it.get("tags"), list) else []
          plugins.append(
              {
                  "rank": idx,
                  "id": plugin_id,
                  "name": str(display_name),
                  "desc": str(it.get("desc") or it.get("description") or ""),
                  "tags": [str(t) for t in tags],
                  "stars": _as_int(it.get("stars") or it.get("star") or 0),
                  "updated_at": it.get("updated_at"),
              }
          )
  
      return {
          "status": "ok",
          "source": source,
          "mode": mode,
          "query": query,
          "start": start,
          "count": count,
          "total_plugins": total,
          "matched_plugins": matched if mode == "search" else total,
          "returned_plugins": len(plugins),
          "plugins": plugins,
      }

• astrbot_mcp/server.py:33-33 (registration)

  Registers the browse_plugin_market tool handler with the FastMCP server using the re-exported function from astrbot_tools.

  server.tool(astrbot_tools.browse_plugin_market, name="browse_plugin_market")

• astrbot_mcp/tools/__init__.py:27-27 (registration)

  Re-exports the browse_plugin_market handler from plugin_market_tools.py into the tools package namespace for use in server.py.

  from .plugin_market_tools import browse_plugin_market

• astrbot_mcp/server.py:48-60 (registration)

  Lists 'browse_plugin_market' as one of the available tools in the astrbot_info resource.

  "tools": [
      "get_astrbot_logs",
      "get_message_platforms",
      "send_platform_message",
      "send_platform_message_direct",
      "restart_astrbot",
      "get_platform_session_messages",
      "browse_plugin_market",
      "list_astrbot_config_files",
      "inspect_astrbot_config",
      "apply_astrbot_config_ops",
      "search_astrbot_config_paths",
  ],