from __future__ import annotations import argparse import json from typing import Any from m2_mcp.config_api import ( get_command_log, get_ai_online_v2, search_ai_rcmd_operations, search_ai_systems, set_multi_values, ) def _parse_bool(raw: str) -> bool: text = str(raw).strip().lower() if text in {"1", "true", "yes", "y", "on"}: return True if text in {"0", "false", "no", "n", "off"}: return False raise argparse.ArgumentTypeError(f"invalid boolean value: {raw}") def _print(payload: Any) -> None: print(json.dumps(payload, ensure_ascii=False, indent=2)) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Smoke test for instrument config MCP upstream APIs") subparsers = parser.add_subparsers(dest="command", required=True) p = subparsers.add_parser("search-ai-systems") p.add_argument("--project-key", required=True) p.add_argument("--keyword", default="") p.add_argument("--page-size", type=int, default=20) p.add_argument("--page-num", type=int, default=1) p.add_argument("--order-by", nargs="*", default=["-id"]) p = subparsers.add_parser("search-ai-rcmd-operations") p.add_argument("--project-key", required=True) p.add_argument("--codes", nargs="+", required=True) p.add_argument("--end", required=True) p.add_argument("--page-size", type=int, default=20) p.add_argument("--page-num", type=int, default=1) p.add_argument("--order", default="-create_time") p = subparsers.add_parser("get-ai-online-v2") p.add_argument("--project-key", required=True) p.add_argument("--codes", nargs="+", required=True) p = subparsers.add_parser("get-command-log") p.add_argument("--project-key", required=True) p.add_argument("--point-id", required=True) p.add_argument("--begin", type=int, required=True) p.add_argument("--end", type=int, required=True) p.add_argument("--page-size", type=int, default=20) p.add_argument("--page-num", type=int, default=1) p.add_argument("--export", type=_parse_bool, default=False) p = subparsers.add_parser("set-multi-values") p.add_argument("--project-key", required=True) p.add_argument("--points-json", required=True) p.add_argument("--from", dest="from_", default="M2_BACKEND") return parser def main() -> None: parser = build_parser() args = parser.parse_args() if args.command == "search-ai-systems": _print( search_ai_systems( args.project_key, keyword=args.keyword, page_size=args.page_size, page_num=args.page_num, order_by=args.order_by, ) ) return if args.command == "search-ai-rcmd-operations": _print( search_ai_rcmd_operations( args.project_key, codes=args.codes, end=args.end, page_size=args.page_size, page_num=args.page_num, order=args.order, ) ) return if args.command == "get-ai-online-v2": _print(get_ai_online_v2(args.project_key, codes=args.codes)) return if args.command == "get-command-log": _print( get_command_log( args.project_key, point_id=args.point_id, begin=args.begin, end=args.end, page_size=args.page_size, page_num=args.page_num, export=args.export, ) ) return if args.command == "set-multi-values": _print(set_multi_values(args.project_key, points=json.loads(args.points_json), from_=args.from_)) return raise ValueError(f"unsupported command: {args.command}") if __name__ == "__main__": main()