Glean Platform Search API：构建权限感知搜索客户端

新集成优先 Platform API：POST /api/search，Bearer token，JSON query。结果权限感知，可用 cursor 翻页。

先看清成品

• mock 返回带 datasource 的 results。
• 1 项 pytest 通过。

合规提醒：Glean 是企业产品，需管理员采购与部署；个人通常不能自助注册。搜索与聊天结果继承各数据源权限，禁止用管理员特权接口绕过 ACL 给无授权用户。Token 等同企业访问权，勿提交公开仓库。本文不编造报价。

运行说明：无公司 Glean 租户。A/B 用本地 mock 演示 Platform Search/Chat 形状；C 为发查询前的脱敏守卫。未连接真实 *-be.glean.com。2026-10-23（Asia/Shanghai）pytest 通过。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

108-glean-A-src/
├── glean_search.py
├── mock_glean.py
└── test_108a.py

第 1 步：客户端

#!/usr/bin/env python3
"""Glean Platform Search: POST /api/search (permission-aware)."""
from __future__ import annotations
import argparse, json, os, sys, urllib.error, urllib.request

def search(base, token, query, page_size=10):
    req = urllib.request.Request(
        f"{base.rstrip('/')}/api/search",
        data=json.dumps({"query": query, "pageSize": page_size}).encode(),
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e

def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("query")
    ap.add_argument("--base", default=os.environ.get("GLEAN_BASE_URL", "https://example-be.glean.com"))
    ap.add_argument("--token", default=os.environ.get("GLEAN_TOKEN", ""))
    args = ap.parse_args(argv)
    if not args.token:
        print("设置 GLEAN_TOKEN（需公司管理员发放的 Platform token）", file=sys.stderr); return 2
    if not args.query.strip():
        print("query 为空", file=sys.stderr); return 2
    print(json.dumps(search(args.base, args.token, args.query), ensure_ascii=False, indent=2))
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：Mock

#!/usr/bin/env python3
"""Mock Glean Platform API search + chat."""
from __future__ import annotations
import json
from http.server import BaseHTTPRequestHandler, HTTPServer

TOKEN = "glean-test-token"

class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass
    def _auth(self):
        return self.headers.get("Authorization") == f"Bearer {TOKEN}"
    def _send(self, code, obj):
        b = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(b)))
        self.end_headers()
        self.wfile.write(b)
    def do_POST(self):
        if not self._auth():
            self._send(401, {"error": "Unauthorized"}); return
        n = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(n) or b"{}")
        if self.path == "/api/search":
            q = body.get("query") or ""
            self._send(200, {
                "results": [
                    {"title": f"Doc about {q}", "url": "https://intranet.example/doc1", "datasource": "confluence", "snippet": "permission-aware mock"},
                ],
                "next_cursor": None,
            }); return
        if self.path == "/api/chat":
            msgs = body.get("messages") or body.get("query") or ""
            text = msgs if isinstance(msgs, str) else json.dumps(msgs)[:120]
            self._send(200, {"messages": [{"author": "GLEAN_AI", "fragments": [{"text": f"Grounded answer for: {text}"}]}]}); return
        self.send_error(404)

def main():
    import argparse
    ap = argparse.ArgumentParser(); ap.add_argument("--port", type=int, default=9108)
    args = ap.parse_args()
    HTTPServer(("127.0.0.1", args.port), H).serve_forever()

if __name__ == "__main__":
    main()

第 3 步：测试

import subprocess, sys, time
from pathlib import Path
HERE = Path(__file__).resolve().parent

def test_search():
    port = 19180
    proc = subprocess.Popen([sys.executable, str(HERE/"mock_glean.py"), "--port", str(port)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        time.sleep(0.3)
        r = subprocess.run([sys.executable, str(HERE/"glean_search.py"), "vacation", "--base", f"http://127.0.0.1:{port}", "--token", "glean-test-token"], capture_output=True, text=True, check=True)
        assert "confluence" in r.stdout
    finally:
        proc.terminate(); proc.wait(timeout=5)

第 4 步：运行

$ pytest glean search
.                                                                        [100%]
1 passed in 0.36s

怎么检查结果？

• pytest 绿。
• 正式环境向管理员申请最小权限 token。

调试示例

401/403。 检查 Bearer 是否过期、是否有 SEARCH scope，以及 base URL 是否指向本公司实例。

常见问题

模板规则

• 最小权限 token。
• 不绕过 ACL。
• 查询先过 C 篇守卫。

项目资源

企业搜索的第一课：权限对了，结果才安全。