AstrBot MCP

Instructions

Implementation Reference

• astrbot_mcp/tools/config_tools.py:512-620 (handler)

  The core handler function that fetches an AstrBot config by ID, applies batch operations (set, add_key, append) using flexible path parsing, updates the config via API, and returns success/error with changed paths.

  async def apply_astrbot_config_ops(
      *,
      conf_id: str,
      ops: List[Dict[str, Any]],
      create_missing: bool = False,
  ) -> Dict[str, Any]:
      """
      Apply multiple edits to an AstrBot config, then save + hot reload.
  
      Supported ops (batch in a single tool call):
        - {"op":"set","path":<path>,"value":<any>}
        - {"op":"add_key","path":<parent_path>,"key":<str>,"value":<any>}
        - {"op":"append","path":<list_path>,"value":<any>}
  
      `path` accepts dot path, JSON Pointer, or segment list.
      """
      client = AstrBotClient.from_env()
  
      if not conf_id:
          return {"status": "error", "message": "conf_id is required"}
      if not isinstance(ops, list) or not ops:
          return {"status": "error", "message": "ops must be a non-empty list"}
  
      try:
          api_result = await client.get_abconf(conf_id=conf_id, system_config=False)
      except Exception as e:
          return {
              "status": "error",
              "message": _astrbot_connect_hint(client),
              "base_url": client.base_url,
              "detail": _httpx_error_detail(e),
          }
  
      status = api_result.get("status")
      if status != "ok":
          return {"status": status, "message": api_result.get("message"), "raw": api_result}
  
      config = (api_result.get("data") or {}).get("config")
      if not isinstance(config, dict):
          return {"status": "error", "message": "AstrBot returned invalid config payload", "raw": api_result}
  
      changed: List[List[JsonPathSegment]] = []
      counters = {"set": 0, "add_key": 0, "append": 0}
      try:
          for i, op in enumerate(ops):
              if not isinstance(op, dict):
                  raise ValueError(f"ops[{i}] must be an object")
              op_name = (op.get("op") or "").strip()
              if op_name == "set":
                  path_segments = _parse_path(op.get("path"))
                  if not path_segments:
                      raise ValueError("set op requires non-empty path")
                  _set_value(
                      config,
                      path_segments,
                      op.get("value"),
                      create_missing=create_missing,
                  )
                  changed.append(path_segments)
                  counters["set"] += 1
              elif op_name == "add_key":
                  parent_segments = _parse_path(op.get("path"))
                  key = op.get("key")
                  if not isinstance(key, str) or not key.strip():
                      raise ValueError("add_key op requires non-empty string 'key'")
                  _add_key(
                      config,
                      parent_segments,
                      key=key,
                      value=op.get("value"),
                      create_missing=create_missing,
                  )
                  changed.append(parent_segments + [key])
                  counters["add_key"] += 1
              elif op_name == "append":
                  list_segments = _parse_path(op.get("path"))
                  _append_list_item(
                      config,
                      list_segments,
                      value=op.get("value"),
                      create_missing=create_missing,
                  )
                  changed.append(list_segments + ["-"])
                  counters["append"] += 1
              else:
                  raise ValueError(f"Unsupported op: {op_name!r}")
      except Exception as e:
          return {"status": "error", "message": str(e), "op_index": i, "op": ops[i]}
  
      try:
          update_result = await client.update_astrbot_config(conf_id=conf_id, config=config)
      except Exception as e:
          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),
          }
  
      status = update_result.get("status")
      if status != "ok":
          return {"status": status, "message": update_result.get("message"), "raw": update_result}
  
      return {
          "message": update_result.get("message") or "ok",
          "conf_id": conf_id,
          "applied": counters,
          "changed_paths": changed,
      }

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

  Registers the 'apply_astrbot_config_ops' tool with the FastMCP server instance, making it available to MCP clients.

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

• astrbot_mcp/tools/config_tools.py:33-106 (helper)

  Key helper function to parse input paths into JSONPath segments, supporting multiple formats (dot, JSON Pointer, lists). Used by the handler for all path-based operations.

  def _parse_path(path: str | List[Any] | None) -> List[JsonPathSegment]:
      """
      Parse a JSON path.
  
      Supported formats:
        - None / "" -> []
        - List segments: ["provider", 0, "model_config", "temperature"]
        - JSON Pointer: "/provider/0/model_config/temperature"
        - Dot path: "provider.0.model_config.temperature"
  
      Note:
        - JSON Pointer unescapes "~1" -> "/" and "~0" -> "~".
        - Numeric segments are converted to int.
        - "-" is kept as string (used for list append in some ops).
      """
      if path is None or path == "":
          return []
  
      if isinstance(path, list):
          segments: List[JsonPathSegment] = []
          for seg in path:
              if isinstance(seg, bool):
                  raise ValueError("Path segment cannot be bool")
              if isinstance(seg, int):
                  if seg < 0:
                      raise ValueError("List index cannot be negative")
                  segments.append(seg)
                  continue
              if isinstance(seg, str):
                  s = seg.strip()
                  if s == "":
                      raise ValueError("Path segment cannot be empty string")
                  if s.isdigit():
                      segments.append(int(s))
                  else:
                      segments.append(s)
                  continue
              raise ValueError(f"Unsupported path segment type: {type(seg).__name__}")
          return segments
  
      if not isinstance(path, str):
          raise ValueError(f"Unsupported path type: {type(path).__name__}")
  
      raw = path.strip()
      if raw == "":
          return []
  
      # JSON Pointer
      if raw.startswith("/"):
          parts = raw.split("/")[1:]
          out: List[JsonPathSegment] = []
          for part in parts:
              part = part.replace("~1", "/").replace("~0", "~")
              if part == "":
                  raise ValueError("Invalid JSON pointer: empty segment")
              if part != "-" and part.isdigit():
                  out.append(int(part))
              else:
                  out.append(part)
          return out
  
      # Dot path
      parts = raw.split(".")
      out = []
      for part in parts:
          part = part.strip()
          if part == "":
              raise ValueError("Invalid dot path: empty segment")
          if part.isdigit():
              out.append(int(part))
          else:
              out.append(part)
      return out

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

  Re-exports 'apply_astrbot_config_ops' from config_tools.py in tools.__init__.py, allowing easy import as astrbot_tools.apply_astrbot_config_ops in server.py.

  "apply_astrbot_config_ops",

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

  Also listed in the astrbot://info resource endpoint which advertises available tools to MCP hosts.

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