用 Manus Webhook 接收任务事件：RSA-SHA256 验签、重放保护与去重

轮询会消耗请求额度，官方建议用 webhook：任务创建、停止、删除时，Manus 主动 POST 到你的地址。 这篇做一个只用标准库的接收服务 webhook_server.py：按官方算法验证 RSA-SHA256 签名、拒绝 5 分钟前的请求、按 event_id 去重，把事件写进 events.jsonl，并打印一行摘要。

先看清成品

• 在 box 上真实跑通：用我自己生成的测试密钥对签名，4 次投递（完成、提问、未知类型、重复）全部按预期处理。
• 复现并修复真实问题：服务用 Host 头拼出 http://127.0.0.1 去验签，而签名用的是注册的 https 公网地址，结果 401。
• 另外两个真实小坑：pytest 把 sign_test.py 当测试收集导致报错；shell 里残留的 PORT 变量让服务监听错端口。pytest 7 项通过。

合规提醒：只使用 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，返回内容原样贴在文中），以及全部本地代码。webhook 内容和签名来自我写的 mock，字段照官方 v2 文档的示例和字段说明写，已在文中标出。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

81-manus-B-src/
├── webhook_server.py
├── make_test_keys.py
├── send_signed.py
├── samples/
│   ├── stopped_finish.json
│   ├── stopped_ask.json
│   └── unknown.json
└── test_webhook.py

第 1 步：接收服务

"""接收 Manus webhook：验签（RSA-SHA256）、防重放、去重、落盘。只用标准库 + cryptography。

签名内容（官方 Webhook Security）：{timestamp}.{url}.{sha256_hex(原始 body)}
url 是你在 Manus 注册的完整回调地址（含查询参数）。
"""
import base64, hashlib, json, os, sys, time
from http.server import BaseHTTPRequestHandler, HTTPServer
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

MAX_SKEW = 300  # 官方：5 分钟


def verify(public_key_pem, url, body, signature_b64, timestamp, now=None):
    now = int(time.time()) if now is None else now
    try:
        if abs(now - int(timestamp)) > MAX_SKEW:
            return False, "timestamp 超出 5 分钟"
        sig = base64.b64decode(signature_b64, validate=True)
    except (ValueError, TypeError):
        return False, "签名或时间戳格式错误"
    content = f"{timestamp}.{url}.{hashlib.sha256(body).hexdigest()}".encode()
    key = serialization.load_pem_public_key(public_key_pem)
    try:
        key.verify(sig, content, padding.PKCS1v15(), hashes.SHA256())
        return True, "ok"
    except InvalidSignature:
        return False, "签名不匹配"


class Store:
    def __init__(self, path):
        self.path, self.seen = path, set()
        if os.path.exists(path):
            for line in open(path, encoding="utf-8"):
                self.seen.add(json.loads(line)["event_id"])

    def add(self, event):
        if event["event_id"] in self.seen:
            return False
        self.seen.add(event["event_id"])
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps(event, ensure_ascii=False) + "\n")
        return True


def summarize(event):
    t, d = event.get("event_type"), event.get("task_detail", {})
    if t == "task_stopped":
        if d.get("stop_reason") == "ask":
            return f"需要回答：{d.get('message', '')[:60]} → {d.get('task_url')}"
        files = ", ".join(f"{a['file_name']}({a['size_bytes']}B)" for a in d.get("attachments") or [])
        so = d.get("structured_output")
        extra = "" if so is None else f"；结构化输出 success={so.get('success')}"
        return f"完成：{d.get('task_title')}；附件：{files or '无'}{extra}"
    if t in ("task_created", "task_deleted"):
        return f"{t}：{d.get('task_id') or event.get('task_id')}"
    return f"未知事件类型 {t}，已忽略"


def make_handler(public_key_pem, public_url, store, log=print):
    class H(BaseHTTPRequestHandler):
        def do_POST(self):
            body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
            sig, ts = self.headers.get("X-Webhook-Signature"), self.headers.get("X-Webhook-Timestamp")
            if not sig or not ts:
                return self._send(400, "Missing headers")
            # 关键：用注册时的公网 URL，而不是从 Host 头拼出来的本地地址
            url = public_url.rstrip("/") + self.path if public_url else f"http://{self.headers['Host']}{self.path}"
            ok, why = verify(public_key_pem, url, body, sig, ts)
            if not ok:
                log(f"拒绝：{why}（验签用的 url={url}）")
                return self._send(401, "Unauthorized")
            event = json.loads(body)
            if store.add(event):
                log(summarize(event))
            else:
                log(f"重复投递，跳过 {event['event_id']}")
            self._send(200, "OK")   # 未知类型也回 200，避免对方重试

        def _send(self, code, text):
            self.send_response(code); self.end_headers(); self.wfile.write(text.encode())

        def log_message(self, *a):
            pass
    return H


if __name__ == "__main__":
    pem = open(os.environ.get("MANUS_PUBLIC_KEY_FILE", "manus_public_key.pem"), "rb").read()
    base = os.environ.get("PUBLIC_BASE_URL", "")   # 例如 https://hooks.example.com
    port = int(os.environ.get("PORT", "8081"))
    print(f"监听 :{port}，验签 base={base or '(从 Host 头推断)'}", flush=True)
    HTTPServer(("127.0.0.1", port), make_handler(pem, base, Store("events.jsonl"),
               log=lambda s: print(s, flush=True))).serve_forever()

第 2 步：测试密钥和模拟投递

# make_test_keys.py
"""生成本地测试用的 2048 位 RSA 密钥对（仅用于模拟，真实环境用 webhook.publicKey 返回的公钥）。"""
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
k = rsa.generate_private_key(public_exponent=65537, key_size=2048)
open("test_private_key.pem", "wb").write(k.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()))
open("test_public_key.pem", "wb").write(k.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo))
print("已生成 test_private_key.pem / test_public_key.pem")

# send_signed.py
"""本地模拟 Manus 投递：用我自己生成的测试密钥对签名（不是 Manus 的私钥）。
用法：python send_signed.py <注册的完整URL> <发送到的本地URL> <payload.json>"""
import base64, hashlib, sys, time, urllib.request
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

signed_url, target, payload = sys.argv[1:4]
key = serialization.load_pem_private_key(open("test_private_key.pem", "rb").read(), None)
body = open(payload, "rb").read()
ts = str(int(time.time()))
content = f"{ts}.{signed_url}.{hashlib.sha256(body).hexdigest()}".encode()
sig = base64.b64encode(key.sign(content, padding.PKCS1v15(), hashes.SHA256())).decode()
req = urllib.request.Request(target, data=body, method="POST", headers={
    "Content-Type": "application/json", "X-Webhook-Signature": sig, "X-Webhook-Timestamp": ts})
try:
    print(urllib.request.urlopen(req).status)
except urllib.error.HTTPError as e:
    print(e.code)

第 3 步：示例事件（我写的，字段按官方文档）

{"event_id": "task_stopped_task_demo1", "event_type": "task_stopped", "task_detail": {"task_id": "task_demo1", "task_title": "整理三款记账软件对比", "task_url": "https://manus.im/app/task_demo1", "message": "已完成对比表。", "attachments": [{"file_name": "compare.md", "url": "https://example.com/compare.md", "size_bytes": 2310}], "stop_reason": "finish"}}
{"event_id": "task_stopped_task_demo2", "event_type": "task_stopped", "task_detail": {"task_id": "task_demo2", "task_title": "导出报告", "task_url": "https://manus.im/app/task_demo2", "message": "要导出 PDF 还是 DOCX？", "attachments": [], "stop_reason": "ask", "question_expectation": {"event_id": "ask_demo2", "options": ["PDF", "DOCX"], "selection_mode": "multiple", "response_method": "send_message"}}}
{"event_id": "evt_demo3", "event_type": "some_future_event", "task_detail": {"task_id": "task_demo3"}}

第 4 步：测试

"""用临时生成的测试密钥对（不是 Manus 的）验证验签逻辑。"""
import base64, hashlib, json
import pytest
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa
import webhook_server as ws

URL = "https://hooks.example.com/manus/webhook?src=prod"
KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
PEM = KEY.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)


def sign(body, ts, url=URL):
    c = f"{ts}.{url}.{hashlib.sha256(body).hexdigest()}".encode()
    return base64.b64encode(KEY.sign(c, padding.PKCS1v15(), hashes.SHA256())).decode()


BODY = b'{"event_id":"e1","event_type":"task_created","task_detail":{"task_id":"t1"}}'


def test_valid():
    assert ws.verify(PEM, URL, BODY, sign(BODY, 1000), "1000", now=1100) == (True, "ok")


def test_tampered_body():
    assert ws.verify(PEM, URL, BODY + b" ", sign(BODY, 1000), "1000", now=1000)[0] is False


def test_url_must_match_including_query_and_scheme():
    s = sign(BODY, 1000)
    assert ws.verify(PEM, "https://hooks.example.com/manus/webhook", BODY, s, "1000", now=1000)[0] is False
    assert ws.verify(PEM, URL.replace("https", "http"), BODY, s, "1000", now=1000)[0] is False


def test_old_timestamp_rejected():
    assert ws.verify(PEM, URL, BODY, sign(BODY, 1000), "1000", now=1000 + 301) == (False, "timestamp 超出 5 分钟")


def test_garbage_signature():
    assert ws.verify(PEM, URL, BODY, "%%%", "1000", now=1000)[0] is False
    assert ws.verify(PEM, URL, BODY, sign(BODY, 1000), "abc", now=1000)[0] is False


def test_store_dedupes(tmp_path):
    p = tmp_path / "e.jsonl"
    s = ws.Store(str(p))
    assert s.add({"event_id": "a"}) and not s.add({"event_id": "a"})
    assert not ws.Store(str(p)).add({"event_id": "a"})   # 重启后仍能去重


def test_summarize_structured_output_and_unknown():
    e = {"event_type": "task_stopped", "task_detail": {"task_title": "T", "stop_reason": "finish", "attachments": [],
         "structured_output": {"success": False, "value": {"a": ""}, "error": "Failed to extract structured output"}}}
    assert "success=False" in ws.summarize(e)
    assert "已忽略" in ws.summarize({"event_type": "new_thing"})

第 5 步：运行

$ python make_test_keys.py
已生成 test_private_key.pem / test_public_key.pem
$ export MANUS_PUBLIC_KEY_FILE=test_public_key.pem PORT=8081
$ PUBLIC_BASE_URL=https://hooks.example.com python webhook_server.py &
$ python send_signed.py https://hooks.example.com/manus/webhook http://127.0.0.1:8081/manus/webhook samples/stopped_finish.json
200
$ python send_signed.py https://hooks.example.com/manus/webhook http://127.0.0.1:8081/manus/webhook samples/stopped_ask.json
200
$ python send_signed.py https://hooks.example.com/manus/webhook http://127.0.0.1:8081/manus/webhook samples/unknown.json
200
$ python send_signed.py https://hooks.example.com/manus/webhook http://127.0.0.1:8081/manus/webhook samples/stopped_finish.json
200
# 服务端输出
监听 :8081，验签 base=https://hooks.example.com
完成：整理三款记账软件对比；附件：compare.md(2310B)
需要回答：要导出 PDF 还是 DOCX？ → https://manus.im/app/task_demo2
未知事件类型 some_future_event，已忽略
重复投递，跳过 task_stopped_task_demo1
$ wc -l events.jsonl
3 events.jsonl
$ python -m pytest -q
7 passed

# 真实部署：把 MANUS_PUBLIC_KEY_FILE 换成 webhook.publicKey 返回的公钥，
# 用 webhook.create 注册 https://你的域名/manus/webhook（我没有 Key，没跑过这一步）

怎么检查结果？

• 重复投递只记一次：events.jsonl 行数等于不同 event_id 的数量。
• 故意改一个字节的 body 再发，应返回 401。
• 收到 stop_reason: ask 时，回答要通过 task.sendMessage 发送（A 篇的 reply）。
• 附件 URL 会过期，需要的话收到后尽快下载保存。

调试示例

1. 验签全部失败：URL 不一致。 第一次运行时没设 PUBLIC_BASE_URL，服务从 Host 头拼地址：

$ python send_signed.py https://hooks.example.com/manus/webhook http://127.0.0.1:8081/manus/webhook samples/stopped_finish.json
401
# 服务端
监听 :8081，验签 base=(从 Host 头推断)
拒绝：签名不匹配（验签用的 url=http://127.0.0.1:8081/manus/webhook）

签名里的 url 是注册时的 https://hooks.example.com/manus/webhook，而服务看到的是反向代理转发后的 http://127.0.0.1:8081/...。协议、主机、端口任何一处不同，签名都不对。官方示例用的 request.url 在反向代理后面也会遇到同样的问题。修复：用环境变量 PUBLIC_BASE_URL 指定注册时的地址，只拼接 path 和查询参数；拒绝时把验签用的 url 打到日志里，一眼就能看出差异。

2. 服务连不上：残留的 PORT 变量。 第一次启动时日志是 监听 :3099，客户端报 ConnectionRefusedError——之前别的项目在 shell 里 export 过 PORT=3099。启动前显式写 PORT=8081。

3. pytest 收集报错。 模拟脚本最初叫 sign_test.py，pytest 把它当测试文件导入，脚本读取 sys.argv 拿到了 no:cacheprovider：FileNotFoundError: [Errno 2] No such file or directory: 'no:cacheprovider'。改名为 send_signed.py。

常见问题

模板规则

• 验签用原始 body 字节，不要先解析 JSON 再序列化。
• url 用注册时的完整地址。
• 先验签再解析，再去重。
• 未知事件返回 200。
• 公钥缓存，但验签失败多时重新获取。

项目资源

先在本机用测试密钥把验签跑通，再换成官方公钥上线。