auth_gateway · 网关 API Key 鉴权

本页收录可运行示例，配套本文档「指南 08」(指南 08)。 代码为随框架交付的示例，已一并收录到本站供直接对照使用。

说明与运行

示例: 网关 API Key 鉴权(后端自研接入). 背景: 框架不内置认证(需求能力面, 由需求方决定启用, 见概要设计 4.1/4.2). 本示例展示后端如何接管启动, 给网关挂 API Key 鉴权 middleware: - 不调用 serve(), 改用 create_app() 拿网关实例 → 挂鉴权 middleware → 自己 uvicorn.run - 请求头按接口契约: Authorization: Bearer [apikey] - 校验失败返回 401 + 契约 jsonRes 错误结构({code, statusText, message, data}), 前端无感 运行: cd agentframework PYTHONPATH=src AGENT_API_KEY=dev-key-123 python examples/auth_gateway.py 然后: curl -H "Authorization: Bearer dev-key-123" ... /api/v2/chat/completions

运行前需在已安装 agentframework 的 Python ≥ 3.11 环境；示例多为自带本地模型，无需真实 API Key；需要外部依赖（如 chromadb / 数据库驱动）会在说明中注明。

完整源码

"""示例: 网关 API Key 鉴权(后端自研接入).

背景: 框架不内置认证(需求能力面, 由需求方决定启用, 见概要设计 4.1/4.2).
本示例展示后端如何接管启动, 给网关挂 API Key 鉴权 middleware:

- 不调用 serve(), 改用 create_app() 拿网关实例 → 挂鉴权 middleware → 自己 uvicorn.run
- 请求头按接口契约: `Authorization: Bearer [apikey]`
- 校验失败返回 401 + 契约 jsonRes 错误结构({code, statusText, message, data}), 前端无感

运行:
    cd agentframework
    PYTHONPATH=src AGENT_API_KEY=dev-key-123 python examples/auth_gateway.py
    然后: curl -H "Authorization: Bearer dev-key-123" ... /api/v2/chat/completions
"""

from __future__ import annotations

import os
from collections.abc import Callable

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

from agentframework.gateway.app import create_app

# 1. API Key 来源: 演示用环境变量; 真实场景从自己的数据库/配置中心读取
VALID_KEYS = {os.getenv("AGENT_API_KEY", "dev-key-123")}


def _unauthorized(message: str) -> JSONResponse:
    """401 + 契约 jsonRes 错误结构(与网关 GatewayError.to_response 一致, 前端无感)."""
    return JSONResponse(
        status_code=401,
        content={"code": 401, "statusText": "", "message": message, "data": None},
    )


def install_api_key_auth(app: FastAPI) -> FastAPI:
    """给网关挂 API Key 鉴权 middleware.

    校验逻辑: Authorization 头必须为 `Bearer <key>`, 且 key 在 VALID_KEYS 内.
    可按需扩展: 查数据库 / 按 appId 区分 key 权限 / 加限流等.
    """

    @app.middleware("http")
    async def check_api_key(request: Request, call_next: Callable):
        auth = request.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            return _unauthorized("缺少 Authorization: Bearer [apikey] 请求头")
        api_key = auth[len("Bearer ") :].strip()
        if api_key not in VALID_KEYS:
            return _unauthorized("无效的 API Key")
        return await call_next(request)

    return app


def main() -> None:
    """启动带鉴权的网关(应用注册 + 鉴权 + uvicorn)."""
    from agentframework import Workflow
    from agentframework.nodes import Answer, ChatLLM

    # 2. 构建并注册一个应用(示例: 客服工作流)
    class _EchoModel:
        def invoke(self, messages):
            return "鉴权通过的客服回复"

    wf = Workflow(name="customer_service")
    wf.add_node(
        ChatLLM(
            name="llm",
            chat_model=_EchoModel(),
            output=False,  # 严格收口: 由 Answer 收口(骨架设计 5.2)
        )
    )
    wf.add_node(Answer(name="ans", text=wf.ref("llm.reply")))
    wf.add_edge("llm", "ans")
    app_graph = wf.compile()

    # 3. 网关 = create_app() + 鉴权 middleware + 自己启动
    from agentframework.gateway.registry import AppRegistry

    registry = AppRegistry()
    registry.register("customer_service", app_graph)

    app = install_api_key_auth(create_app(registry))

    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8080)


if __name__ == "__main__":
    main()

---

回到示例索引。