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", ],