用 Manus API 构建可审批的任务运行器：创建任务、跟踪状态与人工确认

Manus 是会自己规划、上网、写代码的 AI 智能体；它也有官方 API，可以从程序里布置任务。 这篇做 manus_task.py：创建任务、轮询进度、在任务停下来提问或请求确认时安全地交给人处理，最后打印结果。重点是“不自动确认”：Agent 想发邮件、部署或执行命令时，脚本只提醒你去 Manus 里看，不替你点同意。

先看清成品

• 用无效 Key 真实调用官方 API，拿到两种 401 返回，脚本能给出清楚的提示。
• 发现一个真实问题：Key 末尾带换行时，requests 在本地就报错，且报错信息里带着整段 Key。脚本已在读取时 strip。
• 10 项 pytest（假服务）覆盖 running/stopped/waiting/error、后台任务、429 退避，全部通过。

合规提醒：只使用 Manus 官方 API v2（https://api.manus.ai/v2/）和你自己账号创建的 API Key，不共享账号或 Key，不用任何非官方封装或网页自动化。API 任务和网页任务一样消耗积分，额度与价格以 manus.im 官网为准（本文不写价格）。Agent 发邮件、部署、执行命令这类动作必须由人确认。让 Agent 访问的网站也要遵守对方的使用条款。

运行说明：我没有 Manus API Key，所以没有创建过真实任务。真实跑过的是：用无效 Key 调用官方 API（2026-10-21 21:09–21:13，UTC+8，返回内容原样贴在文中），以及全部本地代码。正常任务的返回数据来自我写的 mock，字段照官方 v2 文档的示例和字段说明写，已在文中标出。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

各接口每分钟的具体上限见官方“频率限制”页，可能调整，本文不抄数字。

文件结构

81-manus-A-src/
├── manus_task.py
├── prompt.txt
└── test_manus_task.py

第 1 步：写任务描述

请调研 3 款开源的个人记账软件（只用它们的官网和 GitHub 页面），
用表格对比：许可证、支持平台、是否支持多币种、最近一次发布日期。
每一项都给出来源链接；查不到的写“未找到”，不要猜。

第 2 步：主脚本

"""用 Manus API v2 创建任务、轮询状态、安全处理 waiting。

- 只用官方端点：task.create / task.listMessages / task.detail / task.sendMessage
- 永远不自动确认动作（发邮件、部署、执行终端命令等），交给人在 Manus 里决定
- 429 时指数退避 + 随机抖动（官方建议）
"""
import argparse, json, os, random, sys, time
import requests

BASE = "https://api.manus.ai/v2/"
UA = "binbho-tutorial/1.0"
QUESTION_TYPES = {"messageAskUser", "cascadeAskUser"}


class ManusError(Exception):
    def __init__(self, status, code, message, request_id=None):
        super().__init__(f"HTTP {status} {code}: {message} (request_id={request_id})")
        self.status, self.code = status, code


class Client:
    def __init__(self, api_key, session=None, sleep=time.sleep, max_retries=5):
        key = (api_key or "").strip()          # .env 里常带换行
        if not key:
            raise SystemExit("缺少 MANUS_API_KEY（在 manus.im 设置 → Developers 创建）")
        self.s = session or requests.Session()
        self.s.headers.update({"x-manus-api-key": key, "User-Agent": UA})
        self.sleep, self.max_retries = sleep, max_retries

    def _call(self, method, endpoint, **kw):
        for attempt in range(self.max_retries + 1):
            r = self.s.request(method, BASE + endpoint, timeout=60, **kw)
            try:
                body = r.json()
            except ValueError:
                body = {}
            if r.status_code == 429 and attempt < self.max_retries:
                wait = min(60, 2 ** attempt) + random.uniform(0, 1)
                print(f"  429 rate_limited，{wait:.1f}s 后重试", file=sys.stderr)
                self.sleep(wait)
                continue
            if r.status_code >= 400 or body.get("ok") is False:
                e = body.get("error", {})
                raise ManusError(r.status_code, e.get("code"), e.get("message"), body.get("request_id"))
            return body
        raise AssertionError("unreachable")

    def create(self, content):
        return self._call("POST", "task.create", json={"message": {"content": content}})

    def latest_status(self, task_id):
        """order=desc：最新的事件在最前面。返回 (status_update, 本页消息)。"""
        body = self._call("GET", "task.listMessages",
                          params={"task_id": task_id, "order": "desc", "limit": 20})
        msgs = body.get("messages", [])
        for m in msgs:
            if m.get("type") == "status_update":
                return m["status_update"], msgs
        return None, msgs

    def detail(self, task_id):
        return self._call("GET", "task.detail", params={"task_id": task_id})["task"]

    def send(self, task_id, text):
        if not text.strip():
            raise ValueError("不能发送空回答")
        return self._call("POST", "task.sendMessage",
                          json={"task_id": task_id, "message": {"content": text}})


def decide(status_update):
    """把一个 status_update 变成下一步动作：poll / done / ask / confirm_by_human / failed"""
    st = status_update.get("agent_status")
    if st == "running":
        return "poll", None
    if st == "stopped":
        return "done", None
    if st == "error":
        return "failed", None
    if st == "waiting":
        d = status_update.get("status_detail") or {}
        t = d.get("waiting_for_event_type")
        if t in QUESTION_TYPES:
            return "ask", d
        # 动作确认或未知类型：一律不自动确认
        return "confirm_by_human", d
    return "poll", None


def wait(client, task_id, task_url="", interval=10, max_polls=90, sleep=time.sleep):
    for _ in range(max_polls):
        su, msgs = client.latest_status(task_id)
        action, d = decide(su or {"agent_status": "running"})
        if action == "done":
            t = client.detail(task_id)
            if t.get("has_running_background_jobs"):
                print("  主 Agent 已停止，但还有后台任务，继续等", file=sys.stderr)
                sleep(interval)
                continue
            for m in msgs:   # desc：第一条 assistant_message 就是最终回答
                if m.get("type") == "assistant_message":
                    return 0, m["assistant_message"].get("content", "")
            return 0, ""
        if action == "failed":
            err = next((m["error_message"].get("content") for m in msgs
                        if m.get("type") == "error_message"), "未知错误")
            return 2, f"任务失败：{err}"
        if action == "ask":
            q = next((m["assistant_message"].get("content") for m in msgs
                      if m.get("type") == "assistant_message"), "")
            return 3, f"Agent 在提问：{q}\n用 `python manus_task.py reply {task_id} \"你的回答\"` 回复"
        if action == "confirm_by_human":
            return 4, (f"Agent 等待确认：{d.get('waiting_for_event_type')} — "
                       f"{d.get('waiting_description', '')}\n本脚本不自动确认，请人工在 Manus 中查看后决定：{task_url}")
        sleep(interval)
    return 5, f"轮询 {max_polls} 次仍未结束，稍后再查：{task_url}"


def main(argv=None):
    ap = argparse.ArgumentParser()
    sub = ap.add_subparsers(dest="cmd", required=True)
    r = sub.add_parser("run"); r.add_argument("prompt_file"); r.add_argument("--interval", type=int, default=10)
    p = sub.add_parser("reply"); p.add_argument("task_id"); p.add_argument("text")
    a = ap.parse_args(argv)
    c = Client(os.environ.get("MANUS_API_KEY"))
    try:
        if a.cmd == "reply":
            c.send(a.task_id, a.text); print("已回复，重新运行 run 前可用 task.listMessages 查看进度"); return 0
        content = open(a.prompt_file, encoding="utf-8").read()
        t = c.create(content)
        print(f"任务已创建：{t['task_id']}  {t.get('task_url', '')}")
        code, text = wait(c, t["task_id"], t.get("task_url", ""), a.interval)
        print(text)
        return code
    except ManusError as e:
        print(f"Manus API 错误：{e}", file=sys.stderr)
        if e.status == 401:
            print("检查 MANUS_API_KEY 是否正确、是否已被删除", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())

第 3 步：测试（假服务）

"""假 Manus 服务（我写的 mock，字段按官方 v2 文档），不需要网络和 API Key。"""
import json
import pytest
import manus_task as mt


class Resp:
    def __init__(self, status, body):
        self.status_code, self._b = status, body
    def json(self):
        return self._b


class FakeSession:
    def __init__(self, script):
        self.script, self.calls, self.headers = list(script), [], {}
    def request(self, method, url, timeout=None, **kw):
        self.calls.append((method, url.rsplit("/", 1)[1], kw))
        return self.script.pop(0)


def su(status, detail=None):
    m = {"id": "e1", "type": "status_update", "status_update": {"agent_status": status}}
    if detail:
        m["status_update"]["status_detail"] = detail
    return m


def page(*msgs):
    return Resp(200, {"ok": True, "request_id": "r", "task_id": "t1", "messages": list(msgs), "has_more": False})


def detail(bg=False):
    return Resp(200, {"ok": True, "request_id": "r", "task": {"id": "t1", "status": "stopped", "has_running_background_jobs": bg}})


def client(script):
    s = FakeSession(script)
    return mt.Client("k\n", session=s, sleep=lambda x: None), s


def test_key_is_stripped_and_header_used():
    c, s = client([])
    assert s.headers["x-manus-api-key"] == "k"
    assert "Authorization" not in s.headers


def test_running_then_stopped_returns_answer():
    ans = {"id": "a", "type": "assistant_message", "assistant_message": {"content": "结果表格"}}
    c, s = client([page(su("running")), page(su("stopped"), ans), detail()])
    assert mt.wait(c, "t1", sleep=lambda x: None) == (0, "结果表格")


def test_background_jobs_keep_polling():
    ans = {"id": "a", "type": "assistant_message", "assistant_message": {"content": "最终"}}
    c, s = client([page(su("stopped")), detail(bg=True), page(su("stopped"), ans), detail()])
    assert mt.wait(c, "t1", sleep=lambda x: None) == (0, "最终")


def test_action_confirmation_never_auto_confirmed():
    d = {"waiting_for_event_id": "R0", "waiting_for_event_type": "gmailSendAction",
         "waiting_description": "Send email to user@example.com",
         "confirm_input_schema": {"type": "object", "properties": {"accept": {"type": "boolean"}}, "required": ["accept"]}}
    c, s = client([page(su("waiting", d))])
    code, text = mt.wait(c, "t1", "https://manus.im/app/t1", sleep=lambda x: None)
    assert code == 4 and "不自动确认" in text
    assert all(ep != "task.confirmAction" and ep != "task.sendMessage" for _, ep, _ in s.calls)


def test_unknown_waiting_type_goes_to_human():
    assert mt.decide({"agent_status": "waiting", "status_detail": {"waiting_for_event_type": "somethingNew"}})[0] == "confirm_by_human"
    assert mt.decide({"agent_status": "waiting"})[0] == "confirm_by_human"


def test_question_returns_reply_hint():
    q = {"id": "q", "type": "assistant_message", "assistant_message": {"content": "要导出 PDF 还是 DOCX？"}}
    c, s = client([page(su("waiting", {"waiting_for_event_type": "messageAskUser", "waiting_for_event_id": "q"}), q)])
    code, text = mt.wait(c, "t1", sleep=lambda x: None)
    assert code == 3 and "PDF" in text and "reply t1" in text


def test_error_reads_error_message():
    e = {"id": "x", "type": "error_message", "error_message": {"error_type": "internal", "content": "额度不足"}}
    c, s = client([page(su("error"), e)])
    assert mt.wait(c, "t1", sleep=lambda x: None) == (2, "任务失败：额度不足")


def test_429_backoff_then_success():
    rl = Resp(429, {"ok": False, "request_id": "r", "error": {"code": "rate_limited", "message": "slow down"}})
    ok = Resp(200, {"ok": True, "request_id": "r", "task_id": "t1", "task_title": "x", "task_url": "u"})
    c, s = client([rl, rl, ok])
    assert c.create("hi")["task_id"] == "t1"
    assert len(s.calls) == 3


def test_401_raises():
    bad = Resp(401, {"error": {"code": "unauthenticated", "message": "invalid api key"}, "ok": False, "request_id": "f9"})
    c, s = client([bad])
    with pytest.raises(mt.ManusError) as ei:
        c.create("hi")
    assert ei.value.status == 401 and ei.value.code == "unauthenticated"


def test_empty_reply_rejected():
    c, s = client([])
    with pytest.raises(ValueError):
        c.send("t1", "  ")

第 4 步：运行

# 1) 先用 curl 看官方 API 的真实报错（无效 Key / 不带 Key），2026-10-21 21:09 UTC+8
$ curl -s -X POST https://api.manus.ai/v2/task.create -H 'x-manus-api-key: invalid' -H 'Content-Type: application/json' -d '{"message":{"content":"hello"}}'
{"error":{"code":"unauthenticated","message":"invalid api key"},"ok":false,"request_id":"f99ddb3a-a72e-4985-93e3-60ad6e5d767b"}   ← HTTP 401
$ curl -s -X POST https://api.manus.ai/v2/task.create -H 'Content-Type: application/json' -d '{"message":{"content":"hello"}}'
{"error":{"code":"unauthenticated","message":"missing authentication: require either API Key or Bearer Token"},"ok":false,"request_id":"2d6a7db5-3b8f-41bc-8435-a9c393768782"}   ← HTTP 401

# 2) 脚本（无效 Key、空 Key），21:11 UTC+8
$ MANUS_API_KEY=invalid-demo-key python manus_task.py run prompt.txt
Manus API 错误：HTTP 401 unauthenticated: invalid api key (request_id=ef4bad5f-ab61-4618-841f-3ae272e7ec28)
检查 MANUS_API_KEY 是否正确、是否已被删除
$ echo $?
1
$ MANUS_API_KEY= python manus_task.py run prompt.txt
缺少 MANUS_API_KEY（在 manus.im 设置 → Developers 创建）

# 3) 测试
$ python -m pytest -q
10 passed

# 4) 有真实 Key 时的预期（我没跑过）：
#   任务已创建：<task_id>  https://manus.im/app/<task_id>
#   …轮询…
#   退出码 0 = 完成并打印回答；3 = Agent 提问；4 = 等待人工确认；2 = 任务失败；5 = 超过轮询次数

怎么检查结果？

• 退出码 4 时一定去 task_url 看 Agent 想做什么（发给谁、部署到哪），再在 Manus 里决定，不要写脚本自动 confirm。
• 结果里的每个对比项都要点开来源核对；Agent 可能把旧版本信息当成最新。
• 在 Manus 网页的用量页面看这次任务用了多少积分，再决定要不要批量跑。

调试示例

1. Key 末尾的换行把 Key 打进了日志。 我习惯用 MANUS_API_KEY=$(cat key.txt) 或从 .env 读取，文件末尾常有换行。在本地用 requests 准备这样一个请求（不发送）：

$ python -c "import requests; requests.Session().prepare_request(requests.Request('POST','https://api.manus.ai/v2/task.create',headers={'x-manus-api-key':'invalid-demo-key\n'}))"
requests.exceptions.InvalidHeader: Invalid leading whitespace, reserved character(s), or return character(s) in header value: 'invalid-demo-key\n'

请求根本没发出去，而且异常信息里带着完整的 Key——如果这段报错进了日志或 CI 输出，Key 就泄露了。修复：Client.__init__ 里先 .strip()，空字符串直接退出并提示去哪里创建。测试 test_key_is_stripped_and_header_used 固定了这个行为。

2. 两种 401 要分开提示。 真实返回里，不带 Key 是 missing authentication: require either API Key or Bearer Token，Key 错误是 invalid api key，code 都是 unauthenticated。脚本在本地先拦住“没带 Key”，服务端的 401 统一提示检查 Key，并打印 request_id，联系官方支持时用得上。

常见问题

模板规则

• Key 只放环境变量，读取时 strip。
• 轮询用 order=desc，取最新的 status_update。
• stopped 后再查 task.detail 的后台任务。
• 任何动作确认都交给人。
• 429 指数退避加抖动。

项目资源

先用小任务看清积分消耗和结果质量，再把它接进你的流程。