Server Bot 接入指南

什么是 Server Bot

Server Bot 允许你把对话能力托管在自己的后端服务上：平台负责账号、计费、会话管理与前端呈现，你的后端负责生成回复。

平台按 GBP-v1 协议把用户消息转发给你的服务，并以 SSE 流式收回回复，按用量向提问用户收取积分。适用场景：

• 接入私有 / 自研模型
• 行业知识与内部工具编排
• 需要自定义 RAG、检索或工作流的 Bot

快速开始

用 FastAPI 实现三个协议端点，10 分钟即可跑通最小后端（完整示例见仓库 examples/gbp_bot/main.py）：

"""GBP-v1 示例 Bot 后端"""
import hashlib
import hmac
import json
import os
import time
from collections.abc import AsyncIterator

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse

SERVER_KEY = os.environ.get(
    "GBP_SERVER_KEY", "gbpk_dev_example_key_000000000000000000000000"
)

app = FastAPI(title="GBP-v1 Example Bot")

MAX_CLOCK_SKEW = 300  # ±5min


def _sse(event: str, data: dict) -> str:
    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"


def _verify(request: Request, raw_body: bytes) -> tuple[int, str] | None:
    """签名校验 + 版本 + timestamp 防重放。失败返回 (http_status, message)。"""
    if request.headers.get("X-GBP-Version") != "1":
        return 400, "unsupported X-GBP-Version"
    signature = request.headers.get("X-GBP-Signature", "")
    expected = hmac.new(SERVER_KEY.encode(), raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(signature, expected):
        return 403, "bad signature"
    try:
        body = json.loads(raw_body.decode("utf-8"))
    except ValueError:
        return 400, "bad json"
    ts = body.get("timestamp")
    if not isinstance(ts, int) or abs(int(time.time()) - ts) > MAX_CLOCK_SKEW:
        return 401, "timestamp out of window"
    return None


@app.post("/gbp/handshake")
async def handshake(request: Request):
    body = await request.json()
    challenge = body.get("challenge")
    if not isinstance(challenge, str) or not challenge:
        return JSONResponse({"error": "missing challenge"}, status_code=400)
    return {"challenge": challenge}


@app.get("/gbp/meta")
async def meta():
    return {
        "name": "GBP-v1 Example Bot",
        "capabilities": ["chat", "stream"],
        "max_context": 32768,
        "protocol": "gbp-v1",
    }


@app.post("/gbp/chat")
async def chat(request: Request):
    raw = await request.body()
    failure = _verify(request, raw)
    if failure is not None:
        status, message = failure
        return JSONResponse({"error": message}, status_code=status)

    body = json.loads(raw.decode("utf-8"))
    messages = body.get("messages") or []
    last_user = ""
    for m in reversed(messages):
        if m.get("role") == "user":
            content = m.get("content")
            last_user = content if isinstance(content, str) else ""
            break

    async def stream() -> AsyncIterator[str]:
        reply = f"[GBP Example Bot] echo: {last_user[:200]}"
        for ch in reply:
            yield _sse("delta", {"text": ch})
        yield _sse("done", {"finish_reason": "stop"})

    return StreamingResponse(stream(), media_type="text/event-stream")

1. 安装依赖：pip install fastapi uvicorn
2. 启动服务：uvicorn main:app --host 0.0.0.0 --port 9002
3. 在「创建 Bot → Server Bot」填入你的服务地址，点击「检测并握手」

本地联调可用内网穿透工具暴露 https 地址；生产环境平台强制 https 且禁止内网 / 保留地址。

协议参考

端点一览

请求签名

所有平台 → Bot 的 POST 请求携带签名头，算法为 hex(HMAC-SHA256(server_key, raw_body))。请使用原始请求体计算，并务必用常量时间比较函数防止时序攻击：

X-GBP-Version: 1 · X-GBP-Signature: <hex>

Python

import hashlib
import hmac

def verify(raw_body: bytes, signature: str, server_key: str) -> bool:
    expected = hmac.new(
        server_key.encode(), raw_body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected)

Node.js

import { createHmac, timingSafeEqual } from 'node:crypto'

function verify(rawBody: Buffer, signature: string, serverKey: string): boolean {
  const expected = createHmac('sha256', serverKey).update(rawBody).digest('hex')
  const a = Buffer.from(signature)
  const b = Buffer.from(expected)
  return a.length === b.length && timingSafeEqual(a, b)
}

Go

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
)

func verify(rawBody []byte, signature, serverKey string) bool {
	mac := hmac.New(sha256.New, []byte(serverKey))
	mac.Write(rawBody)
	expected := hex.EncodeToString(mac.Sum(nil))
	return hmac.Equal([]byte(signature), []byte(expected))
}

防重放

/gbp/chat 请求体携带 timestamp（Unix 秒），后端应拒绝与当前时间相差超过 ±5 分钟的请求，防止抓包重放。

消息语义

POST /gbp/chat 请求体示例：

{
  "protocol": "gbp-v1",
  "conversation_id": 1024,
  "timestamp": 1761187200,
  "messages": [
    { "role": "user", "content": "Hello" }
  ],
  "params": null,
  "user_context": { "locale": "zh-CN", "timezone": "Asia/Shanghai" }
}

• conversation_id — 平台会话 ID，可用于你自己的会话状态管理
• messages — 标准 OpenAI 消息数组（role/content），多模态消息 content 为 parts 数组
• params — 当次媒体参数覆盖（媒体 Bot 场景），可能为 null
• user_context — 脱敏用户上下文（语言 / 时区），不含任何可识别身份的信息

SSE 事件规范

回复必须以 text/event-stream 流式返回，事件序列：delta×N → done（或 error）：

event: delta
data: {"text": "你好"}

event: delta
data: {"text": "，世界"}

event: done
data: {"finish_reason": "stop"}

• delta — 增量文本，data 的 text 字段承载增量内容
• done — 结束，data 的 finish_reason 字段为 stop
• error — 错误，data 的 message 字段描述原因，用户侧将看到失败提示

密钥管理

server_key 仅在首次握手成功时明文展示一次，请立即妥善保存；平台只存储其 SHA-256 哈希，无法找回。

密钥丢失时，在 Bot 编辑页点击「重新生成密钥」，新密钥同样只展示一次，旧密钥立即失效。

后端热更新：将密钥注入环境变量（如 GBP_SERVER_KEY），重启或滚动更新你的服务进程即可。

上线检查清单

• https 部署（强制，禁止内网 / 保留地址）
• 签名校验已开启（HMAC-SHA256 + 常量时间比较）
• timestamp ±5 分钟窗口校验
• 60 秒内完成响应（尽早 flush delta 保持连接活跃）
• /gbp/meta 声明 chat + stream 能力

计费说明

平台按「每请求固定积分 + 回传 token 估算」向提问用户收取积分，定价由平台配置。

当前平台无支付体系，创建者不参与分成。

可在「统计」页查看自己 Bot 的调用量与积分消耗趋势。

常见问题