WebHook

{
  "schema_version": "v1",
  "event_type": "session.completed",
  "event_id": "2d3134c5-...",
  "occurred_at": "2026-02-15T14:30:00+00:00",
  "is_test": false,
  "user": {
    "user_id": "1234567890",
    "username": "alice",
    "full_name": "Alice",
    "department": "研发",
    "role": "user"
  },
  "webhook": {
    "id": 12,
    "owner_scope": "system"
  },
  "session": {
    "session_id": "5f6a...",
    "title": "日报总结",
    "kind": "chat",
    "status": "completed",
    "project_id": "97d2...",
    "created_at": "2026-02-15T14:20:00+00:00",
    "updated_at": "2026-02-15T14:30:00+00:00"
  },
  "run": {
    "run_id": "6dc9...",
    "turn_id": "20cf...",
    "status": "completed",
    "schedule_mode": "immediate",
    "scheduled_task_id": null,
    "started_at": "2026-02-15T14:20:10+00:00",
    "finished_at": "2026-02-15T14:29:58+00:00"
  },
  "summary": {
    "has_artifacts": true,
    "has_assistant_response": true,
    "message_count": 9,
    "assistant_message_count": 4,
    "tool_execution_count": 3,
    "total_cost_usd": 0.012345,
    "input_tokens": 2345,
    "output_tokens": 1234,
    "duration_ms": 588000
  },
  "meta": {
    "trigger": "run.done",
    "source": "poco.backend.callback.completed",
    "idempotency_key": "2d3134c5-...",
    "capability_profile_version": 3,
    "model_stop_reason": "end_turn"
  }
}

{
  "schema_version": "v1",
  "event_type": "publish_page.published",
  "event_id": "7ec2d1a1-...",
  "occurred_at": "2026-02-15T14:35:00+00:00",
  "is_test": false,
  "user": {
    "user_id": "1234567890",
    "username": "alice",
    "full_name": "Alice",
    "department": "研发",
    "role": "user"
  },
  "webhook": {
    "id": 13,
    "owner_scope": "user"
  },
  "publish_page": {
    "publish_center_id": "8c01...",
    "name": "日报发布",
    "publish_key": "daily-report",
    "source_type": "scheduled_task",
    "category": "日报",
    "public_to_board_enabled": true
  },
  "version": {
    "version_no": 3,
    "session_id": "5f6a...",
    "entry_path": "/report.html",
    "share_code": "AbC123xYz0",
    "published_url": "/share/p/daily-report",
    "trigger_type": "auto",
    "reason": "AI selected the first publishable report"
  },
  "session": {
    "session_id": "5f6a...",
    "title": "日报总结",
    "kind": "scheduled",
    "project_id": "97d2..."
  },
  "meta": {
    "trigger": "publish_page.published",
    "source": "poco.backend.publish_center",
    "idempotency_key": "7ec2d1a1-..."
  }
}

{
  "schema_version": "v1",
  "event_type": "workspace_export.completed",
  "event_id": "f1b5a6d4-...",
  "occurred_at": "2026-02-15T14:32:00+00:00",
  "is_test": false,
  "user": {
    "user_id": "1234567890",
    "username": "alice",
    "full_name": "Alice",
    "department": "研发",
    "role": "user"
  },
  "webhook": {
    "id": 14,
    "owner_scope": "user"
  },
  "workspace_export": {
    "status": "ready",
    "files_prefix": "workspaces/user-1/session/files",
    "manifest_key": "workspaces/user-1/session/manifest.json",
    "archive_key": "workspaces/user-1/session/archive.zip",
    "archive_available": true
  },
  "session": {
    "session_id": "5f6a...",
    "title": "日报总结",
    "kind": "chat",
    "status": "completed",
    "project_id": "97d2...",
    "created_at": "2026-02-15T14:20:00+00:00",
    "updated_at": "2026-02-15T14:32:00+00:00"
  },
  "run": {
    "run_id": "6dc9...",
    "turn_id": "20cf...",
    "status": "completed",
    "schedule_mode": "immediate",
    "scheduled_task_id": null,
    "started_at": "2026-02-15T14:20:10+00:00",
    "finished_at": "2026-02-15T14:29:58+00:00"
  },
  "meta": {
    "trigger": "workspace_export.completed",
    "source": "poco.backend.callback.workspace_export",
    "idempotency_key": "f1b5a6d4-...",
    "capability_profile_version": 3
  }
}

curl -X POST 'http://<backend>/api/v1/webhooks/<webhook_id>/test' \
  -H 'Content-Type: application/json' \
  -H 'Authorization: Bearer <token>' \
  -d '{"event_type":"session.completed","session_id":"<session_uuid>"}'

curl -X POST 'http://<backend>/api/v1/webhooks/<webhook_id>/test' \
  -H 'Content-Type: application/json' \
  -H 'Authorization: Bearer <token>' \
  -d '{"event_type":"publish_page.published","session_id":"<session_uuid>"}'

curl -X POST 'http://<backend>/api/v1/webhooks/<webhook_id>/test' \
  -H 'Content-Type: application/json' \
  -H 'Authorization: Bearer <token>' \
  -d '{"event_type":"workspace_export.completed","session_id":"<session_uuid>"}'

import hashlib
import hmac


def build_signature(secret: str, timestamp: str, body_bytes: bytes) -> str:
    signing_payload = timestamp.encode("utf-8") + b"." + body_bytes
    digest = hmac.new(
        secret.encode("utf-8"),
        signing_payload,
        hashlib.sha256,
    ).hexdigest()
    return f"sha256={digest}"

import hashlib
import hmac


def verify_signature(
    secret: str,
    timestamp: str,
    body_bytes: bytes,
    signature_from_header: str,
) -> bool:
    expected = build_signature(secret, timestamp, body_bytes)
    return hmac.compare_digest(expected, signature_from_header or "")

from __future__ import annotations

import hashlib
import hmac
import json
from datetime import datetime, timedelta, timezone
from typing import Any

from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()

# 请改成你自己的密钥；如果你在 Poco 中没配置密钥，可关闭验签逻辑。
WEBHOOK_SECRET = "replace-with-your-secret"

# 简单幂等缓存（生产建议换 Redis/DB，并设置过期时间）
PROCESSED_EVENT_IDS: set[str] = set()


def build_signature(secret: str, timestamp: str, body_bytes: bytes) -> str:
    signing_payload = timestamp.encode("utf-8") + b"." + body_bytes
    digest = hmac.new(
        secret.encode("utf-8"),
        signing_payload,
        hashlib.sha256,
    ).hexdigest()
    return f"sha256={digest}"


def verify_signature(secret: str, timestamp: str, body_bytes: bytes, signature: str) -> bool:
    expected = build_signature(secret, timestamp, body_bytes)
    return hmac.compare_digest(expected, signature or "")


def verify_timestamp(timestamp: str, max_skew_minutes: int = 5) -> None:
    try:
        ts = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
    except ValueError as exc:
        raise HTTPException(status_code=400, detail="Invalid X-Poco-Timestamp") from exc

    now = datetime.now(timezone.utc)
    if abs(now - ts.astimezone(timezone.utc)) > timedelta(minutes=max_skew_minutes):
        raise HTTPException(status_code=400, detail="Timestamp skew too large")


@app.post("/webhook/poco")
async def receive_poco_webhook(
    request: Request,
    x_poco_event: str | None = Header(default=None, alias="X-Poco-Event"),
    x_poco_event_id: str | None = Header(default=None, alias="X-Poco-Event-Id"),
    x_poco_timestamp: str | None = Header(default=None, alias="X-Poco-Timestamp"),
    x_poco_signature: str | None = Header(default=None, alias="X-Poco-Signature"),
) -> dict[str, Any]:
    if not x_poco_event or not x_poco_event_id or not x_poco_timestamp:
        raise HTTPException(status_code=400, detail="Missing webhook headers")

    body_bytes = await request.body()

    # 如果设置了密钥，执行验签
    if WEBHOOK_SECRET:
        if not x_poco_signature:
            raise HTTPException(status_code=401, detail="Missing signature")
        verify_timestamp(x_poco_timestamp, max_skew_minutes=5)
        if not verify_signature(WEBHOOK_SECRET, x_poco_timestamp, body_bytes, x_poco_signature):
            raise HTTPException(status_code=401, detail="Invalid signature")

    # 解析 JSON
    try:
        payload = json.loads(body_bytes.decode("utf-8"))
    except json.JSONDecodeError as exc:
        raise HTTPException(status_code=400, detail="Invalid JSON payload") from exc

    # 幂等处理
    if x_poco_event_id in PROCESSED_EVENT_IDS:
        return {"ok": True, "duplicate": True}
    PROCESSED_EVENT_IDS.add(x_poco_event_id)

    # 事件路由
    if x_poco_event == "session.completed":
        session_id = payload.get("session", {}).get("session_id")
        title = payload.get("session", {}).get("title")
        kind = payload.get("session", {}).get("kind")
        has_artifacts = payload.get("summary", {}).get("has_artifacts")
        # TODO: 在这里写你的业务逻辑（入库、推送消息、触发任务等）
        print(
            "[Webhook] session.completed",
            f"session_id={session_id}",
            f"title={title}",
            f"kind={kind}",
            f"has_artifacts={has_artifacts}",
        )
    elif x_poco_event == "publish_page.published":
        publish_page = payload.get("publish_page", {})
        version = payload.get("version", {})
        print(
            "[Webhook] publish_page.published",
            f"publish_key={publish_page.get('publish_key')}",
            f"version_no={version.get('version_no')}",
            f"entry_path={version.get('entry_path')}",
        )
    else:
        print("[Webhook] ignored event:", x_poco_event)

    return {"ok": True}

pip install fastapi uvicorn
uvicorn webhook_receiver:app --host 0.0.0.0 --port 8000 --reload