Notion Agent API：创建会话并读取助手回复

Notion Agent API：POST /v1/sessions（agent_id + message），头 Authorization + Notion-Version（如 2026-03-11）。轮询 GET /v1/sessions/{id}，再用 POST .../events/query 读 agent.message。

先看清成品

mock 完成会话并返回 Mock reply。

官方无效 token HTTP 401。

1 项 pytest 通过。

合规提醒：Notion AI / Custom Agents 只能看到你有权限的页面；回答会过时若资料过时。会议录音前告知参会者。企业需确认管理员的 AI 与数据设置。Agent API 为 public beta，能力与套餐（Business/Enterprise）以官方为准。本文不编造价格。

运行说明：无正式 NOTION_API_KEY / AGENT_ID。A 用 mock sessions；对 api.notion.com 无效 token 实测 HTTP 401。B 本地纪要抽取；C 提示词泄密扫描。2026-10-23（Asia/Shanghai）pytest 通过。

做完之后你会得到什么？

成果	说明
`agent_session.py`	会话客户端
`mock_notion_agent.py`	假服务
`test_112a.py`	测试

需要准备哪些环境？

项目	说明
Python	3.10+
环境变量	`NOTION_API_KEY`、`NOTION_AGENT_ID`；可选 `NOTION_VERSION`

关键事实

事实	来源
API 与已有 Custom Agent 对话，不负责在 API 里新建 Agent	官方 Quickstart
可用 PAT 或开启 Agents 能力的 internal connection token	官方 Quickstart
Custom Agents 常见于 Business/Enterprise	官方说明

文件结构

112-notion-ai-A-src/
├── agent_session.py
├── mock_notion_agent.py
└── test_112a.py

第 1 步：客户端

#!/usr/bin/env python3
"""Notion Agent API (public beta): POST /v1/sessions to chat with a Custom Agent.

Headers: Authorization Bearer, Notion-Version (e.g. 2026-03-11), Content-Type.
Body: agent_id + message. Poll GET /v1/sessions/{id} until completed, then query events.
"""
from __future__ import annotations
import argparse, json, os, sys, time, urllib.error, urllib.request

DEFAULT_VERSION = "2026-03-11"


def _headers(token, version):
    return {
        "Authorization": f"Bearer {token}",
        "Notion-Version": version,
        "Content-Type": "application/json",
    }


def create_session(base, token, agent_id, message, version=DEFAULT_VERSION):
    body = {"agent_id": agent_id, "message": message}
    req = urllib.request.Request(
        f"{base.rstrip('/')}/v1/sessions",
        data=json.dumps(body).encode(),
        headers=_headers(token, version),
        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 get_session(base, token, session_id, version=DEFAULT_VERSION):
    req = urllib.request.Request(
        f"{base.rstrip('/')}/v1/sessions/{session_id}",
        headers=_headers(token, version),
        method="GET",
    )
    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 query_events(base, token, session_id, version=DEFAULT_VERSION):
    body = {"filter": {"type": "agent.message"}}
    req = urllib.request.Request(
        f"{base.rstrip('/')}/v1/sessions/{session_id}/events/query",
        data=json.dumps(body).encode(),
        headers=_headers(token, version),
        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 wait_completed(base, token, session_id, version=DEFAULT_VERSION, timeout=20, interval=0.2):
    deadline = time.time() + timeout
    while time.time() < deadline:
        s = get_session(base, token, session_id, version)
        st = s.get("status") or (s.get("session") or {}).get("status")
        if st in ("completed", "complete", "idle"):
            return s
        time.sleep(interval)
    raise SystemExit("session 轮询超时")


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--base", default=os.environ.get("NOTION_BASE_URL", "https://api.notion.com"))
    ap.add_argument("--token", default=os.environ.get("NOTION_API_KEY", ""))
    ap.add_argument("--agent-id", default=os.environ.get("NOTION_AGENT_ID", ""))
    ap.add_argument("--version", default=os.environ.get("NOTION_VERSION", DEFAULT_VERSION))
    ap.add_argument("message")
    ap.add_argument("--poll", action="store_true")
    args = ap.parse_args(argv)
    if not args.token or not args.agent_id:
        print("设置 NOTION_API_KEY 与 NOTION_AGENT_ID（Business/Enterprise + Agents 能力）", file=sys.stderr)
        return 2
    created = create_session(args.base, args.token, args.agent_id, args.message, args.version)
    sid = created.get("id") or (created.get("session") or {}).get("id")
    if args.poll and sid:
        wait_completed(args.base, args.token, sid, args.version)
        ev = query_events(args.base, args.token, sid, args.version)
        print(json.dumps({"session_id": sid, "events": ev}, ensure_ascii=False, indent=2))
    else:
        print(json.dumps(created, ensure_ascii=False, indent=2))
    return 0


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

第 2 步：Mock

#!/usr/bin/env python3
"""Mock Notion Agent sessions API (public beta shape)."""
from __future__ import annotations
import json, uuid
from http.server import BaseHTTPRequestHandler, HTTPServer

TOKEN = "ntn_test"
SESSIONS = {}

class H(BaseHTTPRequestHandler):
    def log_message(self, *a): pass
    def _auth(self):
        return self.headers.get("Authorization") == f"Bearer {TOKEN}" and self.headers.get("Notion-Version")
    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, {"object": "error", "message": "Unauthorized"}); return
        n = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(n) or b"{}")
        path = self.path.split("?")[0]
        if path == "/v1/sessions":
            sid = str(uuid.uuid4())
            SESSIONS[sid] = {"id": sid, "object": "session", "status": "queued",
                             "agent_id": body.get("agent_id"), "message": body.get("message")}
            self._send(200, SESSIONS[sid]); return
        if path.endswith("/events/query"):
            sid = path.split("/")[3]
            s = SESSIONS.get(sid) or {}
            self._send(200, {"object": "list", "results": [
                {"type": "agent.message", "content": f"Mock reply about: {s.get('message', '')}"}
            ]}); return
        self.send_error(404)
    def do_GET(self):
        if not self._auth(): self._send(401, {"object": "error", "message": "Unauthorized"}); return
        parts = self.path.split("?")[0].strip("/").split("/")
        if len(parts) == 3 and parts[0] == "v1" and parts[1] == "sessions":
            sid = parts[2]
            s = SESSIONS.get(sid)
            if not s: self._send(404, {"message": "not found"}); return
            s = {**s, "status": "completed"}
            SESSIONS[sid] = s
            self._send(200, s); return
        self.send_error(404)

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

if __name__ == "__main__":
    main()

第 3 步：测试

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

def test_session_poll_events():
    port = 19113
    proc = subprocess.Popen([sys.executable, str(HERE/"mock_notion_agent.py"), "--port", str(port)],
                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        time.sleep(0.3)
        r = subprocess.run(
            [sys.executable, str(HERE/"agent_session.py"), "报销流程是什么？",
             "--base", f"http://127.0.0.1:{port}", "--token", "ntn_test",
             "--agent-id", "agent-1", "--poll"],
            capture_output=True, text=True, check=True)
        data = json.loads(r.stdout)
        assert data["session_id"]
        assert "Mock reply" in data["events"]["results"][0]["content"]
    finally:
        proc.terminate(); proc.wait(timeout=5)

第 4 步：运行

$ pytest 112a
.                                                                        [100%]
1 passed in 0.36s

$ invalid key api.notion.com sessions
HTTP 401
{"object":"error","status":401,"code":"unauthorized","message":"API token is invalid.","additional_data":{"reason":"token_invalid"},"request_id":"362de2d9-e415-41e6-a798-78cd87287bfc"}

怎么检查结果？

输出含 session_id 与 events。
正式调用前用 C 篇扫提示词。

调试示例

401 token_invalid。 在 Developer portal 重建 PAT，或给 connection 打开 View sessions and interact with agents 并分享该 Agent。

常见问题

现象	原因	处理
Notion-Version	跟文档当前值	2026-03-11 起 Agents beta
流式	Accept text/event-stream	见 Overview
无 Agents 菜单	套餐或权限	问工作区管理员

模板规则

token 最小权限。
核对引用页面再执行。
beta 接口以文档为准。

项目资源

用途	地址
Agent Quickstart（官方）	https://developers.notion.com/guides/notion-agent-apis/quickstart
Agent Overview（官方）	https://developers.notion.com/guides/notion-agent-apis/overview
Notion AI	https://www.notion.com/product/ai

会话通了，工作区权限与引用核对不能省。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

第 1 步：客户端

第 2 步：Mock

第 3 步：测试

第 4 步：运行

怎么检查结果？

调试示例

常见问题

模板规则

项目资源

相关教程