用 Manus Structured Output 与 JSON Schema 校验任务结果

Manus API 可以在 task.create 时附上 structured_output_schema，任务结束后按你的 JSON Schema 抽取结果。 这篇做 schema_lint.py：在调用 API 前按官方规则检查 Schema（这些规则比普通 JSON Schema 严格得多），并在拿到结果后只在 success: true 时使用数据。

先看清成品

• 我按习惯写的第一版 Schema 被检查出 6 个问题，改成第二版后通过。
• 真实调用发现：带着不合规 Schema 用无效 Key 请求，服务端先返回 401，不会告诉你 Schema 的问题——没有 Key 时只能靠本地检查。
• 7 项测试通过，包括官方文档里的两个示例 Schema（对象规则、递归树）。

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

做完之后你会得到什么？

需要准备哪些环境？

关键事实

“5 层”我按“每遇到一个 object 或 array 算一层，根对象算第 1 层”实现，这是我的理解；以服务端返回为准。

文件结构

81-manus-C-src/
├── schema_lint.py
├── schema_v1.json
├── schema_v2.json
└── test_schema_lint.py

第 1 步：检查脚本

"""在调用 task.create 之前，按 Manus 官方 Structured Output 规则检查 JSON Schema；
任务结束后，只在 success=true 时使用 value。"""
import json, sys

UNSUPPORTED = {"pattern", "format", "minLength", "maxLength", "minimum", "maximum",
               "exclusiveMinimum", "exclusiveMaximum", "multipleOf", "minItems", "maxItems",
               "uniqueItems", "allOf", "oneOf", "not", "if", "then", "else"}
MAX_DEPTH = 5


def lint(schema):
    errs = []
    if schema.get("type") != "object":
        errs.append("$: 根节点必须是 type=object")
    defs = schema.get("$defs", {})

    def walk(node, path, depth, seen_refs):
        if not isinstance(node, dict):
            return
        if "$ref" in node:
            name = node["$ref"].rsplit("/", 1)[-1]
            if name in seen_refs:      # 递归结构：不再展开
                return
            return walk(defs.get(name, {}), path + f"->{name}", depth, seen_refs | {name})
        for k in node:
            if k in UNSUPPORTED:
                errs.append(f"{path}: 不支持的关键字 `{k}`")
        types = node.get("type")
        types = types if isinstance(types, list) else [types]
        if "object" in types or "array" in types:
            depth += 1
            if depth > MAX_DEPTH:
                errs.append(f"{path}: 嵌套超过 {MAX_DEPTH} 层")
                return
        if "object" in types:
            props = node.get("properties", {})
            if node.get("additionalProperties") is not False:
                errs.append(f"{path}: additionalProperties 必须为 false")
            missing = sorted(set(props) - set(node.get("required", [])))
            if missing:
                errs.append(f"{path}: required 必须列出全部字段，缺 {missing}（可选字段改用可空类型）")
            for k, v in props.items():
                walk(v, f"{path}.{k}", depth, seen_refs)
        if "array" in types:
            if "items" not in node:
                errs.append(f"{path}: array 必须指定 items")
            walk(node.get("items"), f"{path}[]", depth, seen_refs)
        for sub in node.get("anyOf", []):
            walk(sub, path, depth, seen_refs)

    walk(schema, "$", 0, frozenset())
    return errs


def use_result(result):
    """result 是 structured_output_result 或 webhook 里的 structured_output。"""
    if not result.get("success"):
        # value 此时是“零值兜底”（空字符串、0、空数组……），不能当真实数据入库
        raise ValueError(f"结构化输出失败：{result.get('error')}")
    return result["value"]


if __name__ == "__main__":
    s = json.load(open(sys.argv[1], encoding="utf-8"))
    errs = lint(s)
    for e in errs:
        print("✗", e)
    print("✓ 通过" if not errs else f"共 {len(errs)} 个问题")
    sys.exit(1 if errs else 0)

第 2 步：第一版 Schema（不合规）

{
  "type": "object",
  "properties": {
    "apps": {
      "type": "array",
      "minItems": 3,
      "items": {
        "type": "object",
        "properties": {
          "name": {"type": "string", "minLength": 1},
          "license": {"type": "string"},
          "homepage": {"type": "string", "format": "uri"},
          "multi_currency": {"type": "boolean"},
          "last_release": {"type": "string", "description": "YYYY-MM-DD"}
        },
        "required": ["name", "license", "homepage"]
      }
    }
  },
  "required": ["apps"]
}

第 3 步：修正后的 Schema

{
  "type": "object",
  "properties": {
    "apps": {
      "type": "array",
      "description": "恰好 3 款；查不到的字段填 null",
      "items": {
        "type": "object",
        "properties": {
          "name": {"type": "string"},
          "license": {"type": ["string", "null"]},
          "homepage": {"type": "string", "description": "官网完整 URL"},
          "multi_currency": {"type": ["boolean", "null"]},
          "last_release": {"type": ["string", "null"], "description": "YYYY-MM-DD，未找到填 null"},
          "sources": {"type": "array", "items": {"type": "string"}}
        },
        "required": ["name", "license", "homepage", "multi_currency", "last_release", "sources"],
        "additionalProperties": false
      }
    }
  },
  "required": ["apps"],
  "additionalProperties": false
}

第 4 步：测试

import json, pytest
from schema_lint import lint, use_result


def obj(props):
    return {"type": "object", "properties": props, "required": list(props), "additionalProperties": False}


def test_v1_fails_v2_passes():
    assert len(lint(json.load(open("schema_v1.json", encoding="utf-8")))) == 6
    assert lint(json.load(open("schema_v2.json", encoding="utf-8"))) == []


def test_official_example_passes():   # 官方文档 Object Rules 示例
    s = obj({"name": {"type": "string"}, "age": {"type": "integer"}, "nickname": {"type": ["string", "null"]}})
    assert lint(s) == []


def test_official_recursive_tree_passes():
    s = {"type": "object", "properties": {"root": {"$ref": "#/$defs/TreeNode"}}, "required": ["root"],
         "additionalProperties": False,
         "$defs": {"TreeNode": {"type": "object", "properties": {"label": {"type": "string"},
                   "children": {"type": "array", "items": {"$ref": "#/$defs/TreeNode"}}},
                   "required": ["label", "children"], "additionalProperties": False}}}
    assert lint(s) == []


def test_depth_5_ok_6_rejected():
    five = obj({"a": obj({"b": obj({"c": obj({"d": obj({"e": {"type": "string"}})})})})})
    assert lint(five) == []
    six = obj({"a": obj({"b": obj({"c": obj({"d": obj({"e": obj({"f": {"type": "string"}})})})})})})
    assert any("嵌套超过" in e for e in lint(six))


def test_root_must_be_object_and_array_needs_items():
    assert lint({"type": "array", "items": {"type": "string"}})[0].startswith("$: 根节点")
    assert any("items" in e for e in lint(obj({"x": {"type": "array"}})))


def test_oneof_rejected_anyof_ok():
    assert lint(obj({"x": {"anyOf": [{"type": "string"}, {"type": "null"}]}})) == []
    assert lint(obj({"x": {"oneOf": [{"type": "string"}]}}))


def test_use_result():
    assert use_result({"success": True, "value": {"a": 1}, "error": None}) == {"a": 1}
    with pytest.raises(ValueError, match="Failed to extract"):
        use_result({"success": False, "value": {"country": "", "capital": ""}, "error": "Failed to extract structured output"})

第 5 步：创建任务时带上 Schema

import json, os, requests
schema = json.load(open("schema_v2.json", encoding="utf-8"))
r = requests.post("https://api.manus.ai/v2/task.create",
    headers={"x-manus-api-key": os.environ["MANUS_API_KEY"].strip(), "User-Agent": "binbho-tutorial/1.0"},
    json={"message": {"content": open("../81-manus-A-src/prompt.txt", encoding="utf-8").read()},
          "structured_output_schema": schema}, timeout=60)
print(r.status_code, r.json())
# 任务结束后，在 listMessages 的 structured_output_result 或 webhook 的
# task_detail.structured_output 里拿到结果，交给 use_result()

第 6 步：运行

$ python schema_lint.py schema_v1.json
✗ $: additionalProperties 必须为 false
✗ $.apps: 不支持的关键字 `minItems`
✗ $.apps[]: additionalProperties 必须为 false
✗ $.apps[]: required 必须列出全部字段，缺 ['last_release', 'multi_currency']（可选字段改用可空类型）
✗ $.apps[].name: 不支持的关键字 `minLength`
✗ $.apps[].homepage: 不支持的关键字 `format`
共 6 个问题
$ python schema_lint.py schema_v2.json
✓ 通过

# 真实请求：不合规 Schema + 无效 Key（2026-10-21 21:13 UTC+8）
$ curl -s -X POST https://api.manus.ai/v2/task.create -H 'x-manus-api-key: invalid-demo-key' -H 'Content-Type: application/json' --data-binary @body_with_schema_v1.json
{"error":{"code":"unauthenticated","message":"invalid api key"},"ok":false,"request_id":"ff28b0e5-0c69-4908-a49c-233a6f594fff"}   ← HTTP 401

$ python -m pytest -q
7 passed

怎么检查结果？

• 任何 success: false 的结果都不入库：use_result() 会抛出异常。零值兜底的空字符串看起来像“合法数据”，混进表格最难发现。
• success: true 也只说明结构符合，内容仍要核对来源（sources 字段就是为此加的）。
• 原来用 minItems、format 表达的约束，改写进 description，并在拿到结果后用代码检查（例如条目数是否为 3）。

调试示例

1. 习惯写法几乎全不合规。 第一版 Schema 是按普通 JSON Schema 习惯写的：用 minItems: 3 要求 3 条、format: uri 约束网址、minLength 防止空名字、只把必填字段写进 required。检查结果 6 个问题（见运行输出）。修改：删掉所有不支持的关键字，把约束写进 description；所有对象加 additionalProperties: false；“查不到”的字段改成可空类型并放进 required。

2. 服务端先验认证，不会提示 Schema 问题。 我想看看服务端对 v1 Schema 的 400 报错长什么样，带着无效 Key 发了真实请求，得到的是 401 invalid api key。也就是说认证在前，Schema 校验在后：没有可用 Key 时，只能靠本地检查。有 Key 时 400 会扣请求次数（频率限制按分钟计），本地先检查也能少浪费几次。

常见问题

模板规则

• 调用前本地检查 Schema。
• 约束写进 description，结果出来再用代码检查。
• success:false 一律不用 value。
• 给每条数据留 sources 字段。

项目资源

结构化输出让结果能直接进表格，但“能解析”不等于“是对的”。