marimo + FastAPI 鉴权实战：用纯 ASGI 中间件把用户身份传入 Notebook-开发者社区

marimo + FastAPI 鉴权实战：用纯 ASGI 中间件把用户身份传入 Notebook

【免费下载链接】marimoA reactive notebook for Python — run reproducible experiments, query with SQL, execute as a script, deploy as an app, and version with git. Stored as pure Python. All in a modern, AI-native editor.项目地址: https://gitcode.com/GitHub_Trending/ma/marimo

本文以官方示例 examples/frameworks/fastapi-auth 为主线，讲解如何在 FastAPI 中为 marimo notebook 实现登录/登出与会话管理，并通过mo.app_meta().request.user与mo.app_meta().request.meta把已认证用户信息传递进 notebook 单元格。读完后，你可以复制一套可运行的鉴权中间件（同时覆盖 HTTP 与 WebSocket 连接），并理解 marimo 是如何在 ASGIscope层面消费user/meta的，从而在自己的 FastAPI 应用中安全地为 marimo 应用做身份控制。

示例场景与文件构成

官方示例位于examples/frameworks/fastapi-auth/，包含三个文件：

README.md：说明推荐模式与运行方式
main.py：FastAPI 应用 + 鉴权中间件 + 登录页
notebook.py：被挂载的 marimo notebook，读取并展示用户信息

示例覆盖的能力包括：

基于会话 Cookie 的登录 / 登出；
一个纯 ASGI 中间件，为 HTTP 和 WebSocket 连接同时设置scope["user"]与scope["meta"]；
一个通过mo.app_meta().request读取用户信息的 marimo notebook。

为什么必须用纯 ASGI 中间件

这是整个示例中最关键的设计决策，也是 README 专门用一节解释的原因：

marimo 使用 WebSocket 做实时通信。Starlette 的BaseHTTPMiddleware只处理 HTTP 请求，因此在那里设置的scope["user"]在 WebSocket 连接上是不可见的。纯 ASGI 中间件则能同时处理两者。

从源码结构看这一点确实成立：marimo 内部多处依赖scope["user"]作为开发者约定的身份载体。例如内置的 ProxyMiddleware 判断请求是否已认证时，直接读取scope.get("user")并检查is_authenticated属性；marimo/_server/api/auth.py 中的CustomAuthenticationMiddleware甚至会显式保存并还原开发者提前写入的scope["user"]（KEY = "_marimo_prev_user"），以避免 Starlette 的AuthenticationMiddleware覆盖它。也就是说，scope["user"]/scope["meta"]是 marimo 公开约定的 ASGI 接口——如果你的中间件只对 HTTP 生效，notebook 在 WebSocket 握手后重新拉取请求上下文时就会拿不到身份信息。

main.py 逐段解析：一个可运行的鉴权骨架

依赖声明（PEP 723 内联元数据）

main.py 文件头部用# /// script块声明了依赖，这是 uv 的 PEP 723 内联脚本元数据格式，使得uv run --no-project main.py无需项目级pyproject.toml即可自动建环境并安装依赖：

# /// script # requires-python = ">=3.12" # dependencies = [ # "fastapi", # "marimo", # "starlette", # "uvicorn", # "itsdangerous", # "python-multipart", # ] # ///

其中itsdangerous是 StarletteSessionMiddleware做 Cookie 签名所需的，python-multipart则用于表单解析。

AuthMiddleware：覆盖 HTTP 与 WebSocket 的纯 ASGI 中间件

核心实现见 AuthMiddleware，完整逻辑如下：

class AuthMiddleware: # Paths that don't require authentication PUBLIC_PATHS = {"/login"} def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope["type"] not in ("http", "websocket"): await self.app(scope, receive, send) return # SessionMiddleware has already run, so scope["session"] is available. session = scope.get("session", {}) username = session.get("username") if username: # Set user/meta so marimo can read them via mo.app_meta().request scope["user"] = { "is_authenticated": True, "username": username, } scope["meta"] = {"role": "admin"} await self.app(scope, receive, send) return # Not logged in — block unauthenticated access. path = scope.get("path", "") # Allow public paths through without authentication. if path in self.PUBLIC_PATHS: await self.app(scope, receive, send) return # Reject unauthenticated WebSocket connections. if scope["type"] == "websocket": from starlette.websockets import WebSocket ws = WebSocket(scope, receive, send) await ws.close(code=4003) return # Redirect unauthenticated HTTP requests to /login. response = Response( status_code=302, headers={"location": "/login"} ) await response(scope, receive, send)

各分支的处理策略值得注意：

只拦截http与websocket：lifespan等其他 ASGI 事件直接透传；
已登录：从scope["session"]取用户名，写入scope["user"]（含is_authenticated与username两个键，这正是 marimo 侧约定的可序列化结构，下文详述）和scope["meta"]（自定义数据，例如{"role": "admin"}）；
未登录且命中PUBLIC_PATHS（如/login）：放行；
未登录的 WebSocket：直接以状态码4003关闭连接——不能像 HTTP 那样 302 跳转；
未登录的 HTTP：返回 302 重定向到/login。

代码注释中还给出两条实用建议：不要用BaseHTTPMiddleware（原因见上节）；生产环境中可以考虑改用starlette.middleware.authentication.AuthenticationMiddleware，本示例是简化版。

中间件顺序：后添加者最外层

app.add_middleware的调用顺序是理解这段代码的第二个关键点，源码注释明确写道：

# Middleware ordering: In Starlette, the LAST added middleware is the # OUTERMOST (runs first). We need SessionMiddleware to run before # AuthMiddleware so that scope["session"] is populated. So we add # AuthMiddleware first (innermost) and SessionMiddleware last (outermost). app.add_middleware(AuthMiddleware) app.add_middleware( SessionMiddleware, secret_key=os.getenv("SECRET_KEY", "change-me-in-production"), )

Starlette 中最后添加的中间件最先执行（最外层）。AuthMiddleware依赖SessionMiddleware先解析 Cookie 并填充scope["session"]，所以SessionMiddleware必须在最外层：请求先经过它解析会话，再进入AuthMiddleware读取。若顺序写反，scope.get("session")永远是空，鉴权会全部失效。secret_key从环境变量SECRET_KEY读取，用于会话 Cookie 签名，注释提醒生产环境必须替换默认值。

登录 / 登出路由与登录页

示例用一个内联 HTML 字符串作为登录页（LOGIN_PAGE，main.py），并实现了三个路由：

GET /login：渲染登录表单；
POST /login：校验表单中的username/password与模拟用户库users_db = {"admin": "password123"}是否匹配。成功则把用户名写入request.session["username"]并 302 回首页；失败则渲染带红色错误提示的登录页；
GET /logout：request.session.clear()清空会话后重定向到/login。

注释提醒users_db只是模拟数据，生产环境应替换为真实数据库。

挂载 marimo 应用

最后通过marimo.create_asgi_app构建 ASGI 应用并挂载到 FastAPI 根路径：

marimo_app = ( marimo.create_asgi_app(include_code=True) .with_app(path="/", root=notebook_path) .build() ) app.mount("/", marimo_app) if __name__ == "__main__": uvicorn.run(app, host="127.0.0.1", port=8000)

其中notebook_path指向同目录下的 notebook.py；include_code=True表示允许在编辑模式下显示代码。整个 ASGI 应用（包括其内部的 WebSocket 端点）都会被外层 FastAPI 中间件链包裹，因此AuthMiddleware写入的scope["user"]/scope["meta"]能一路透传到 marimo 的会话层。

notebook 侧：用 mo.app_meta().request 读取用户信息

notebook.py 的核心单元格只有几行：

@app.cell def _(mo): req = mo.app_meta().request user = req.user if req else None meta = req.meta if req else None mo.md(f""" ## User info from `mo.app_meta().request` - **user**: `{user}` - **username**: `{user['username'] if isinstance(user, dict) else 'N/A'}` - **meta**: `{meta}` """) return

mo.app_meta()返回AppMeta对象，定义见 marimo/_runtime/runtime.py。其request属性在 marimo/_runtime/app_meta.py 中实现：从运行上下文get_context().request取值；若上下文未初始化（例如以脚本方式直接运行而非在应用中执行），则返回None——这就是 notebook 里需要if req else None防御性判断的原因。文档约定request上通常包含headers、cookies、query_params、path_params、user、url等字段。

源码纵深：scope["user"] / scope["meta"] 如何变成 request.user / request.meta

marimo 在 ASGI 边界处把scope中的身份信息转换为一个可跨进程传递的HTTPRequest对象，其定义见 marimo/_runtime/commands.py：

@dataclass class HTTPRequest(Mapping[str, Any]): """Serializable HTTP request representation. Mimics Starlette/FastAPI Request but is pickle-able and contains only a safe subset of data. Excludes session and auth to prevent exposing sensitive data. """ url: dict[str, Encodable] base_url: dict[str, Encodable] headers: dict[str, str] query_params: dict[str, list[str]] path_params: dict[str, Encodable] cookies: dict[str, str] meta: dict[str, Encodable] # User-defined storage user: Encodable

两个细节值得注意：

session与auth被刻意排除（源码注释写明“它们可能包含应用作者不希望暴露的信息”），所以你在 notebook 里拿不到原始会话对象，只有中间件显式放进scope["user"]/scope["meta"]的数据；
_user_to_dict负责归一化：如果scope["user"]是 Starlette 的BaseUser实例（如SimpleUser），会被转换为{"username", "is_authenticated", "display_name"}字典，因为原始对象会破坏 msgspec 序列化；如果本身是字典（如本示例写法），则原样通过。_meta_to_dict则只强制meta是字典，内部值若不可序列化会在 IPC 编码时报错而非被静默转换。

由此可以推断出对scope["user"]的取值建议：直接用可 JSON 化的字典（如示例）最为稳妥，键is_authenticated和username会被 marimo 内部组件（如代理中间件的认证检查）识别。

运行示例

安装 uv（Pep 723 内联脚本的运行依赖）；
在examples/frameworks/fastapi-auth/目录下执行：
```
uv run --no-project main.py
```
uv 会自动根据文件头部的# /// script块创建临时环境并安装 fastapi、marimo、starlette、uvicorn、itsdangerous、python-multipart；
打开http://localhost:8000/，使用admin/password123登录；
登录成功后，notebook 单元格会通过mo.app_meta().request展示认证用户名（admin）与 meta 数据（{'role': 'admin'}）；访问未登录状态下的根路径会被 302 重定向到登录页，WebSocket 连接则会被以4003关闭。

上生产前的注意事项

结合源码与示例注释，部署前建议至少处理以下事项：

SECRET_KEY必须通过环境变量注入强随机值，否则会话 Cookie 签名可被伪造；
将users_db字典替换为真实用户存储，密码应使用哈希（如bcrypt/argon2）而非明文比较；
若应用更复杂，可评估starlette.middleware.authentication.AuthenticationMiddleware替代手写中间件——marimo 侧的CustomAuthenticationMiddleware（marimo/_server/api/auth.py）已能兼容开发者预置的scope["user"]，但本示例的纯 ASGI 写法对 WebSocket 的控制最直接；
本示例的scope["user"]/scope["meta"]契约只适用于 ASGI 部署路径（mo.app_meta().request依赖运行上下文），以脚本方式python notebook.py直接运行时request为None，代码需自行兜底。