diff --git a/docs/cn/open_source/open_source_api/scheduler/get_status.md b/docs/cn/open_source/open_source_api/scheduler/get_status.md index 87e1a4a5a..0a60d6fac 100644 --- a/docs/cn/open_source/open_source_api/scheduler/get_status.md +++ b/docs/cn/open_source/open_source_api/scheduler/get_status.md @@ -64,34 +64,48 @@ desc: 监控 MemOS 异步任务的生命周期，提供包括任务进度、队 ## 4. 快速上手示例 -使用 SDK 轮询任务状态直至完成： +这些接口由开源版 Server（`server_api`，路由前缀 `/product`）直接提供，使用标准 HTTP 请求即可访问。以下示例轮询任务状态直至完成： ```python -from memos.api.client import MemOSClient import time -client = MemOSClient(api_key="...", base_url="...") +import requests + +# 自部署 MemOS Server 的地址（如启用了鉴权，请自行补充 Authorization 请求头） +base_url = "http://localhost:8000" # 1. 系统级概览：查看整个 MemOS 系统的运行健康度 -global_res = client.get_all_scheduler_status() -if global_res: - print(f"系统运行概况: {global_res.data['scheduler_summary']}") +resp = requests.get(f"{base_url}/product/scheduler/allstatus", timeout=10) +resp.raise_for_status() +global_res = resp.json() +print(f"系统运行概况: {global_res['data']['scheduler_summary']}") # 2. 队列指标监控：检查特定用户的任务积压情况 -queue_res = client.get_task_queue_status(user_id="dev_user_01") -if queue_res: - print(f"待处理任务数: {queue_res.data['remaining_tasks_count']}") - print(f"已下发未完成任务数: {queue_res.data['pending_tasks_count']}") +resp = requests.get( + f"{base_url}/product/scheduler/task_queue_status", + params={"user_id": "dev_user_01"}, + timeout=10, +) +resp.raise_for_status() +queue_res = resp.json() +print(f"排队中任务数: {queue_res['data']['remaining_tasks_count']}") +print(f"已下发未确认任务数: {queue_res['data']['pending_tasks_count']}") # 3. 任务进度追踪：轮询特定任务直至结束 task_id = "task_888999" +active_states = {"waiting", "pending", "in_progress"} while True: - res = client.get_task_status(user_id="dev_user_01", task_id=task_id) - if res and res.code == 200: - current_status = res.data[0]['status'] # data 为状态列表 - print(f"任务 {task_id} 当前状态: {current_status}") - - if current_status in ['completed', 'failed', 'cancelled']: - break + resp = requests.get( + f"{base_url}/product/scheduler/status", + params={"user_id": "dev_user_01", "task_id": task_id}, + timeout=10, + ) + resp.raise_for_status() + items = resp.json().get("data", []) # data 为状态列表：[{"task_id": ..., "status": ...}] + statuses = {item["status"] for item in items} + print(f"任务 {task_id} 当前状态: {statuses or '空'}") + + if not statuses or statuses.isdisjoint(active_states): + break time.sleep(2) ``` diff --git a/docs/cn/open_source/open_source_api/scheduler/ wait.md b/docs/cn/open_source/open_source_api/scheduler/wait.md similarity index 61% rename from docs/cn/open_source/open_source_api/scheduler/ wait.md rename to docs/cn/open_source/open_source_api/scheduler/wait.md index 9849ffe68..52b0d79f6 100644 --- a/docs/cn/open_source/open_source_api/scheduler/ wait.md +++ b/docs/cn/open_source/open_source_api/scheduler/wait.md @@ -42,36 +42,47 @@ desc: 提供阻塞等待与流式进度观测能力，确保在执行后续操 ## 4. 快速上手示例 -使用开源版 SDK 进行阻塞式等待： +这些接口由开源版 Server（`server_api`，路由前缀 `/product`）直接提供，使用标准 HTTP 请求即可访问。注意：`user_name`、`timeout_seconds`、`poll_interval` 均为查询参数（Query），而非请求体（Body）。以下示例进行阻塞式等待： ```python -from memos.api.client import MemOSClient +import json -client = MemOSClient(api_key="...", base_url="...") +import requests + +# 自部署 MemOS Server 的地址（如启用了鉴权，请自行补充 Authorization 请求头） +base_url = "http://localhost:8000" user_name = "dev_user_01" # --- 场景 A：同步阻塞等待 (常用于 Python 自动化脚本) --- print(f"正在等待用户 {user_name} 的任务队列清空...") -res = client.wait_until_idle( - user_name=user_name, - timeout_seconds=300, - poll_interval=2 +resp = requests.post( + f"{base_url}/product/scheduler/wait", + params={"user_name": user_name, "timeout_seconds": 300, "poll_interval": 2}, + timeout=310, # HTTP 超时应大于 timeout_seconds ) -if res and res.code == 200: +resp.raise_for_status() +result = resp.json() # {"message": "idle" | "timeout", "data": {...}} +if result["message"] == "idle": print("✅ 任务已全部完成。") +else: + print(f"⚠️ 等待超时，仍有 {result['data']['running_tasks']} 个任务在执行。") # --- 场景 B：流式进度观测 (常用于前端进度条渲染) --- print("开始监听任务实时进度流...") -# 注意：SSE 接口在 SDK 中通常返回一个生成器 (Generator) -progress_stream = client.stream_scheduler_progress( - user_name=user_name, - timeout_seconds=300 -) - -for event in progress_stream: - # 实时打印剩余任务数 - print(f"当前排队任务数: {event['remaining_tasks_count']}") - if event['status'] == 'idle': - print("🎉 调度器已空闲") - break +with requests.get( + f"{base_url}/product/scheduler/wait/stream", + params={"user_name": user_name, "timeout_seconds": 300}, + stream=True, + timeout=310, +) as resp: + resp.raise_for_status() + for line in resp.iter_lines(decode_unicode=True): + if not line or not line.startswith("data:"): + continue + event = json.loads(line.removeprefix("data:").strip()) + # 实时打印仍在执行的任务数 + print(f"当前活跃任务数: {event['active_tasks']}，状态: {event['status']}") + if event["status"] in ("idle", "timeout"): + print("🎉 调度器已空闲" if event["status"] == "idle" else "⚠️ 监听超时") + break ``` diff --git a/docs/en/open_source/open_source_api/scheduler/get_status.md b/docs/en/open_source/open_source_api/scheduler/get_status.md index f2014d9e5..7d1565858 100644 --- a/docs/en/open_source/open_source_api/scheduler/get_status.md +++ b/docs/en/open_source/open_source_api/scheduler/get_status.md @@ -66,34 +66,48 @@ When you send a status request, **SchedulerHandler** performs the following oper ## 4. Quick Start -Poll task status with the SDK until completion: +These endpoints are served directly by the open-source Server (`server_api`, router prefix `/product`) and can be called with plain HTTP requests. The example below polls task status until completion: ```python -from memos.api.client import MemOSClient import time -client = MemOSClient(api_key="...", base_url="...") +import requests + +# Address of your self-hosted MemOS Server (add an Authorization header if auth is enabled) +base_url = "http://localhost:8000" # 1. System overview: inspect overall MemOS health. -global_res = client.get_all_scheduler_status() -if global_res: - print(f"System summary: {global_res.data['scheduler_summary']}") +resp = requests.get(f"{base_url}/product/scheduler/allstatus", timeout=10) +resp.raise_for_status() +global_res = resp.json() +print(f"System summary: {global_res['data']['scheduler_summary']}") # 2. Queue metrics: inspect backlog for a specific user. -queue_res = client.get_task_queue_status(user_id="dev_user_01") -if queue_res: - print(f"Remaining tasks: {queue_res.data['remaining_tasks_count']}") - print(f"Pending tasks: {queue_res.data['pending_tasks_count']}") +resp = requests.get( + f"{base_url}/product/scheduler/task_queue_status", + params={"user_id": "dev_user_01"}, + timeout=10, +) +resp.raise_for_status() +queue_res = resp.json() +print(f"Remaining tasks: {queue_res['data']['remaining_tasks_count']}") +print(f"Pending tasks: {queue_res['data']['pending_tasks_count']}") # 3. Task progress: poll a specific task until it finishes. task_id = "task_888999" +active_states = {"waiting", "pending", "in_progress"} while True: - res = client.get_task_status(user_id="dev_user_01", task_id=task_id) - if res and res.code == 200: - current_status = res.data[0]['status'] # data is a status list - print(f"Task {task_id} status: {current_status}") - - if current_status in ['completed', 'failed', 'cancelled']: - break + resp = requests.get( + f"{base_url}/product/scheduler/status", + params={"user_id": "dev_user_01", "task_id": task_id}, + timeout=10, + ) + resp.raise_for_status() + items = resp.json().get("data", []) # data is a status list: [{"task_id": ..., "status": ...}] + statuses = {item["status"] for item in items} + print(f"Task {task_id} status: {statuses or 'empty'}") + + if not statuses or statuses.isdisjoint(active_states): + break time.sleep(2) ``` diff --git a/docs/en/open_source/open_source_api/scheduler/wait.md b/docs/en/open_source/open_source_api/scheduler/wait.md index 9de0ff4be..6f356341a 100644 --- a/docs/en/open_source/open_source_api/scheduler/wait.md +++ b/docs/en/open_source/open_source_api/scheduler/wait.md @@ -41,36 +41,47 @@ Both endpoints share the following query parameters: ## 4. Quick Start -Use the open-source SDK for a blocking wait: +These endpoints are served directly by the open-source Server (`server_api`, router prefix `/product`) and can be called with plain HTTP requests. Note that `user_name`, `timeout_seconds`, and `poll_interval` are query parameters, not a request body. The example below performs a blocking wait: ```python -from memos.api.client import MemOSClient +import json -client = MemOSClient(api_key="...", base_url="...") +import requests + +# Address of your self-hosted MemOS Server (add an Authorization header if auth is enabled) +base_url = "http://localhost:8000" user_name = "dev_user_01" # Scenario A: blocking wait, commonly used in Python automation scripts. print(f"Waiting for user {user_name}'s task queue to drain...") -res = client.wait_until_idle( - user_name=user_name, - timeout_seconds=300, - poll_interval=2 +resp = requests.post( + f"{base_url}/product/scheduler/wait", + params={"user_name": user_name, "timeout_seconds": 300, "poll_interval": 2}, + timeout=310, # HTTP timeout should be larger than timeout_seconds ) -if res and res.code == 200: +resp.raise_for_status() +result = resp.json() # {"message": "idle" | "timeout", "data": {...}} +if result["message"] == "idle": print("All tasks have completed.") +else: + print(f"Timed out with {result['data']['running_tasks']} task(s) still running.") # Scenario B: streaming progress, commonly used by frontend progress bars. print("Listening to the live task progress stream...") -# The SSE endpoint usually returns a generator from the SDK. -progress_stream = client.stream_scheduler_progress( - user_name=user_name, - timeout_seconds=300 -) - -for event in progress_stream: - # Print the remaining queued tasks in real time. - print(f"Remaining queued tasks: {event['remaining_tasks_count']}") - if event['status'] == 'idle': - print("Scheduler is idle") - break +with requests.get( + f"{base_url}/product/scheduler/wait/stream", + params={"user_name": user_name, "timeout_seconds": 300}, + stream=True, + timeout=310, +) as resp: + resp.raise_for_status() + for line in resp.iter_lines(decode_unicode=True): + if not line or not line.startswith("data:"): + continue + event = json.loads(line.removeprefix("data:").strip()) + # Print the number of active tasks in real time. + print(f"Active tasks: {event['active_tasks']}, status: {event['status']}") + if event["status"] in ("idle", "timeout"): + print("Scheduler is idle" if event["status"] == "idle" else "Stream timed out") + break ```