企业官网建设流程全解析

Mastra Elysia 服务器适配器（@mastra/elysia）实战指南：在 Elysia 应用中挂载 Agent、Workflow 与流式 API

【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra

本篇指南围绕 Mastra 开源仓库中server-adapters/elysia包的变更记录与其配套源码，系统讲解@mastra/elysia适配器的设计初衷、安装接入方式、核心请求处理链路、SSE 流式输出、认证鉴权与 OpenAPI 集成方案。读完本文，你将掌握如何在一个已有的 Elysia HTTP 应用中嵌入 Mastra 的 Agent、Workflow、Tool、Memory 与流式接口，并理解该适配器在底层如何处理路由、请求体解析、参数映射与异常响应。

一、什么是@mastra/elysia：为什么需要它

Mastra 是一个 TypeScript 编写的 AI 应用框架，核心能力包括 Agent、Workflow、Tool、Memory 等。要让这些能力以 HTTP API 的形式对外暴露，Mastra 提供了@mastra/server这一与框架无关的服务层，以及一组面向具体 Web 框架的"服务器适配器"（Express、Fastify、Hono、NestJS、Elysia 等）。

根据 server-adapters/elysia/README.md 的定位说明：

@mastra/elysiamounts Mastra's agent, workflow, tool, memory, and streaming APIs on an Elysia application. Use it when Elysia is already your HTTP server and you want Mastra endpoints in the same process.

即：当你的项目已经以 Elysia 作为 HTTP 服务框架时，无需另起一个独立服务，只需通过@mastra/elysia把 Mastra 的 Agent、Workflow、Tool、Memory 以及流式（streaming）API 全部挂载到现有的 Elysia 应用实例上，让两者共享同一个进程与端口。这在以下场景尤为合适：

• 已有 Elysia 业务 API，需要新增 AI 能力而不引入第二个服务进程；
• 需要将 Mastra 端点与既有中间件（CORS、日志、鉴权）在同一管道中统一处理；
• 希望复用 Elysia 生态（如@elysiajs/cors、@elysiajs/openapi）对 Mastra 路由进行增强。

该适配器在 CHANGELOG 的 0.1.0 版本（对应 PR #22274）中首次引入，其发布说明写道："Added an Elysia server adapter. Use the new @mastra/elysia package to run a Mastra server inside an Elysia app."

二、安装与快速开始

2.1 安装依赖

根据 server-adapters/elysia/package.json，该包以@mastra/core（>=1.50.0-0 <2.0.0-0）与elysia（^1.4.25）为 peer 依赖，运行环境要求 Node.js>=22.13.0：

npm install @mastra/elysia # 确保同时安装 peer 依赖 npm install elysia@^1.4.25 npm install @mastra/core

运行时依赖仅两个：@mastra/server（提供底层的 MastraServer 抽象与路由定义）与fetch-to-node（用于将 Fetch 语义的请求转换为 Node 的req/res，以支撑 MCP 传输）。

2.2 最小接入示例

CHANGELOG 0.1.0 版本给出了官方的最小示例，这也是该适配器的标准接线方式：

import { Elysia } from 'elysia'; import { MastraServer } from '@mastra/elysia'; import { mastra } from './mastra'; const app = new Elysia(); const server = new MastraServer({ app, mastra }); await server.init(); app.listen(4111);

关键点：

1. MastraServer构造函数接收{ app, mastra }，其中app是 Elysia 应用实例，mastra是你通过new Mastra()组装好的实例（包含 agents、workflows、storage 等）；
2. 必须先await server.init()再app.listen(...)——init()内部会完成所有 Mastra 路由（Agent 执行、Workflow 执行、记忆、流式等）向 Elysia 应用的注册；
3. 端口由 Elysia 的listen()决定，Mastra 端点与你的业务端点共享同一端口。

从构造函数源码（packages/server/src/server/server-adapter/index.ts）可以看到，MastraServer基类还支持以下可选配置项：

2.3 完整可运行示例

仓库中的 server-adapters/elysia/examples/index.ts 提供了一个真实可运行的完整示例：它构建了一个带weatherTool的天气 Agent、一个基于 Workflow 的行程规划流程，并演示了 CORS、OpenAPI 与 Swagger UI 的整合。其核心接线部分：

import { cors } from '@elysiajs/cors'; import { openapi } from '@elysiajs/openapi'; import { Mastra } from '@mastra/core'; import Elysia from 'elysia'; import { MastraServer, getMastraOpenAPIDoc } from '../src/index'; const app = new Elysia(); app.use(cors({ origin: '*' })); // 先创建并初始化 Mastra server const srv = new MastraServer({ mastra, openapiPath: '/openapi.json', app }); await srv.init(); // 从已初始化的 server 提取 OpenAPI 文档 const mastraOpenAPI = getMastraOpenAPIDoc(srv, { title: 'Mastra API with Weather Agent', version: '1.0.0', }); app.use( openapi({ provider: 'swagger-ui', path: '/swagger-ui', specPath: '/openapi.json', documentation: { info: mastraOpenAPI.info, paths: mastraOpenAPI.paths, components: mastraOpenAPI.components, }, }), ); app.listen(3001);

启动后即可访问/openapi.json与/swagger-ui，Mastra 生成的所有端点都会出现在 Swagger UI 中。

三、核心实现剖析：MastraServer 如何融入 Elysia

@mastra/elysia的入口文件 server-adapters/elysia/src/index.ts 中定义了MastraServer类，它继承自@mastra/server/server-adapter导出的抽象基类，通过实现若干抽象方法完成与 Elysia 的对接。

3.1 路由注册：方法链式挂载

registerRoute()方法将每条ServerRoute（方法、路径、响应类型、处理器）按 HTTP 方法映射为 Elysia 的链式调用：

const method = route.method.toLowerCase() as 'get' | 'post' | 'put' | 'delete' | 'patch' | 'all'; appmethod;

为了让 Elysia 能接收任意Elysia实例，源码还定义了最小接口ElysiaApp，只要求use、derive、get/post/put/delete/patch/all、onAfterHandle、onError等能力，从而避免泛型严格匹配带来的类型摩擦。

3.2 路由参数归一化：解决 Elysia 路由冲突

这是该适配器一个非常关键的实现细节。Elysia 的路由器要求同一路径段位的参数名一致，否则可能冲突。Mastra 内部不同路由在同一段位可能使用不同参数名（例如/stored/agents/:storedAgentId与/agents/:agentId）。为此源码实现了：

• normalizeRouteParams()：将:xxx形式的参数统一改写为:p0、:p1等位置化名称，同时记录原始参数名顺序；
• remapParams()：在请求进入时，再把 Elysia 解析出的p0/p1映射回原始的agentId等参数名，保证上层路由处理器拿到的参数名不变。

这一层"写入时归一化、读取时还原"的设计，保证了两套路由体系（Elysia 的路由表与 Mastra 的ServerRoute）可以共存而不冲突。

3.3 请求上下文注入：createContextMiddleware

通过app.derive(createContextMiddleware())，适配器为每个请求注入统一的 Elysia 上下文，包含：

{ requestContext, // 合并后的 RequestContext（来自 body 或 query） mastra, // Mastra 实例 registeredTools, // 已注册工具 taskStore, // 任务存储 abortSignal, // 请求取消信号（透传 Elysia 的 ctx.request.signal） customRouteAuthConfig, }

requestContext的解析策略（对应源码createContextMiddleware()）：

• POST/PUT/PATCH：若Content-Type为application/json，尝试从请求体中的requestContext字段提取；
• GET：尝试从查询参数requestContext中读取，优先按 JSON 解析，失败则回退为base64(JSON)解析。

随后通过mergeRequestContext与applyRequestMetadataToContext将用户信息、请求元数据（如 Header）合并进上下文，供后续的鉴权、RBAC 与 FGA 检查使用。

3.4 请求体与参数解析：兼容 Elysia 的预解析

Elysia 在进入路由处理器前通常已经解析了ctx.body、ctx.query、ctx.params。适配器对此做了双轨处理：

• JSON body：优先使用ctx.body（Elysia 预解析结果）；若未解析到且声明了 JSON，则回退到request.clone().text()手动解析；
• multipart/form-data：调用parseFormData()同时兼容原生FormData与 Elysia 的预解析对象格式，将File转换为Buffer，并尝试把字符串值按 JSON 解析（如options字段）；
• 查询参数：通过normalizeQueryParams()展平数组与嵌套对象（来自@mastra/server/server-adapter的公共工具）。

任何解析失败都会产生bodyParseError，最终由 handler 统一返回结构化的 400 响应（{ error: 'Invalid request body', issues: [...] }）。

3.5 Zod 校验与错误响应

适配器在查询参数、请求体、路径参数三层分别调用基类的parseQueryParams/parseBody/parsePathParams进行 Zod schema 校验与类型强制转换（如z.coerce.number()）。若抛出 Zod 错误，通过resolveValidationError()生成带正确 HTTP 状态码的校验错误响应；其余错误则统一包装为 JSON 响应。此外，基类导出的getCustomHTTPExceptionResponse被用来保留业务侧显式抛出的自定义 HTTP 异常响应。

四、流式响应：SSE 与增量序列化

Agent 与 Workflow 的流式输出是 Mastra 服务的核心能力之一。MastraServer.stream()实现了从上游ReadableStream到 HTTP 响应的完整转码管道：

响应头策略（按streamFormat区分）：

• sse格式：Content-Type: text/event-stream、Cache-Control: no-cache、Connection: keep-alive、X-Accel-Buffering: no；
• stream格式：Content-Type: text/plain，数据块以 ASCII 记录分隔符\x1E分隔。

关键行为（均可在 server-adapters/elysia/src/index.ts 的stream()方法中对应找到）：

1. 连接即冲刷：当路由声明sseFlushOnConnect: true时，连接建立后立即发送: connected\n\n注释行，用于穿透反向代理或触发客户端"已连接"事件（测试见 elysia-adapter.test.ts 的 "SSE stream handshake" 用例）；
2. SSE 注释透传：以:开头的字符串按原样写入，不包裹data:前缀（对应测试 "should pass SSE comment chunks through without data wrapping"）；
3. 敏感数据脱敏：默认开启streamOptions.redact = true，通过redactStreamChunk()移除系统提示词、工具定义、API Key 等敏感字段，再下发客户端（测试 "should redact sensitive data from stream chunks by default" 验证SECRET_SYSTEM_PROMPT、secret_tool均不出现在任何 chunk 中；设置为{ redact: false }则原样透传）；
4. 不可序列化 chunk 容错：对BigInt等JSON.stringify无法处理的值，serializeStreamChunk()返回{ ok: false }时跳过该 chunk 并记录日志，而不是中断整个流——这是针对 GitHub issue #17821 的回归修复（测试见 "Stream Chunk Serialization" 一节："a chunk that JSON.stringify can't handle used to throw inside the stream loop and silently close the HTTP stream"）；
5. 结束标记：SSE 流结束时追加data: [DONE]\n\n。

对于datastream-response、mcp-http、mcp-sse等特殊响应类型，sendResponse()还会通过fetch-to-node的toReqRes/toFetchResponse把 Fetch 语义请求转给 MCP 服务端，并用createSafeReadableStream()包装上游流——即使上游中途报错，也能保留已发送的 chunk 并正常关闭，避免客户端挂起。

五、认证、授权与自定义路由

5.1 认证中间件

server-adapters/elysia/src/auth-middleware.ts 导出了createAuthMiddleware，用于在 Mastra 之外给 Elysia 路由加一层认证。它支持两种凭证来源：

• Authorization: Bearer <token>请求头；
• 查询参数?apiKey=...（当缺少 Authorization 头时回退）。

中间件内部调用@mastra/server/auth的coreAuthMiddleware，并结合ctx.customRouteAuthConfig标记当前METHOD:path需要认证；认证失败时返回结构化的 JSON 错误响应。

import { createAuthMiddleware } from '@mastra/elysia'; const app = new Elysia(); app.derive(createAuthMiddleware({ mastra, requiresAuth: true }));

5.2 路由级鉴权、RBAC 与 FGA

registerRoute()的 handler 在真正执行业务逻辑前依次完成三道检查（对应源码注释顺序）：

1. 路由级认证：checkRouteAuth()基于requiresAuth配置执行认证；
2. RBAC 权限：若mastra.getServer()?.auth已配置，通过动态加载@mastra/core/auth/ee的hasPermission，以requestContext.get('userPermissions')校验requiresPermission；若版本过低，会打印升级提示[@mastra/elysia] Auth features require @mastra/core >= 1.6.0；
3. FGA（关系型访问控制）：checkRouteFGA()用 URL 参数、查询参数与请求体的合并结果，按路由声明的fga规则做细粒度授权。

这三层检查同样应用于registerCustomApiRoutes()注册的自定义 API 路由。仓库中 rbac-permissions.test.ts 与 auth-middleware.test.ts 分别覆盖了这两类场景。

5.3 自定义 API 路由与错误处理

• 通过registerApiRoute（来自@mastra/core/server）注册的自定义路由会被统一注册到 Elysia，并支持鉴权、FGA 与 OpenAPI 元数据（测试见 "Custom API Routes" 一节）；若自定义路由路径以 serverprefix开头，init()会直接拒绝（/must not start with "\/mastra"/），避免与内置路由冲突，框架内部路由除外；
• registerAuthMiddleware()注册了全局onError钩子，把 Elysia 自身的请求体解析/校验错误（400 段状态码）统一转换为结构化的 JSON 响应，而不是 Elysia 默认的纯文本Bad Request；
• registerHttpLoggingMiddleware()在启用httpLoggingConfig时，通过derive+onAfterHandle记录METHOD path status duration，并支持includeQueryParams、includeHeaders与redactHeaders（敏感头替换为[REDACTED]）。

六、OpenAPI 集成：让 Mastra 路由进入 Swagger UI

这是 0.1.0 版本随适配器一并引入的能力。CHANGELOG 记录："Added theconvertCustomRoutesToOpenAPIPathsexport to@mastra/server/server-adapterso server adapters can include custom API routes in generated OpenAPI documents."

server-adapters/elysia/src/helper.ts 提供两个配套函数：

• getMastraOpenAPIDoc(server, options)：从已初始化（init()之后）的MastraServer提取 OpenAPI 3.1.0 文档。内部流程为：基于SERVER_ROUTES调用generateOpenAPIDocument()→ 若有自定义路由则用convertCustomRoutesToOpenAPIPaths()合并进paths→ 若配置了prefix（且非/）则统一改写所有路径前缀（并防双斜杠）→ 用WeakMap缓存结果避免重复生成；options.clearCache: true可强制重新生成；
• clearMastraOpenAPICache(server?)：在路由动态新增后手动失效缓存（WeakMap 无法全量清空，只能按实例删除）。

两者的搭配用法已在 2.3 节的示例中展示：把getMastraOpenAPIDoc返回的info、paths、components喂给@elysiajs/openapi插件，即可在/swagger-ui获得完整的 Mastra API 文档，包括所有 Zod schema 对应的类型定义。对应的集成测试见 elysia-adapter.test.ts 的 "OpenAPI Spec" 一节（验证openapi: '3.1.0'、前缀为/api时的servers覆盖、自定义路由的逐路径servers覆盖等）。

七、版本演进梳理（来自 CHANGELOG）

从 server-adapters/elysia/CHANGELOG.md 可以清晰看到该适配器的演进脉络：

其中 0.1.5 的修复说明与本包源码中registerRoute()的体积检查逻辑一一对应：当content-length存在且超过route.maxBodySize ?? bodyLimitOptions.maxSize时直接返回 413；当请求体已被 Elysia 预解析（ctx.body !== undefined）、无Content-Length且序列化后字节数超限时，同样触发 413 兜底，并可调用bodyLimitOptions.onError定制错误响应。

依赖关系上，该包始终与@mastra/core、@mastra/server保持同频发布（如 0.1.7-alpha.3 依赖@mastra/core@1.67.0-alpha.3与@mastra/server@1.67.0-alpha.3），升级适配器时需同步关注这两个核心包的版本兼容性。

八、测试覆盖与质量保障

server-adapters/elysia拥有相当完整的测试矩阵，全部位于 server-adapters/elysia/src/tests目录：

• elysia-adapter.test.ts：通过共享的createRouteAdapterTestSuite跑通通用路由适配器测试套件，并额外覆盖 SSE 握手、流脱敏、不可序列化 chunk、AbortSignal 透传、multipart、OpenAPI、自定义路由等场景；
• auth-middleware.test.ts：认证中间件的 Bearer Token / apiKey 场景；
• rbac-permissions.test.ts：RBAC 与 FGA 授权链路；
• malformed-json.test.ts、validation-error-hook.test.ts：畸形 JSON 与 Zod 校验错误的响应一致性；
• datastream-error-handling.test.ts：数据流中途出错的容错行为；
• http-logging.test.ts：HTTP 访问日志（含脱敏头）；
• mcp-routes.test.ts、mcp-transport.test.ts：MCP HTTP / SSE 传输的端到端验证；
• server-app-access.test.ts：对 Elysia app 实例访问方式的约束。

测试通过app.fetch(new Request(...))直接在进程内模拟 HTTP 请求，无需真实监听端口即可验证行为（部分用例使用startElysiaServer辅助函数启动真实服务）。

九、小结

@mastra/elysia是一个设计紧凑、与框架深度集成的服务器适配器：它把 Mastra 的 Agent、Workflow、Tool、Memory 与流式能力以标准 HTTP 路由的形式挂载到 Elysia 应用上，同时解决了 Elysia 路由参数命名冲突、请求体预解析兼容、Zod 校验错误结构化、SSE 流式输出、敏感数据脱敏、RBAC/FGA 鉴权以及 OpenAPI 文档生成等一系列实际问题。

若你的项目已经使用 Elysia，接入方式非常直接：new MastraServer({ app, mastra })→await server.init()→app.listen(port)。结合 examples/index.ts 中的完整示例与 index.ts 的源码，你可以快速把 AI 能力注入既有服务，并通过 Swagger UI 即刻获得一份完整的 API 文档。

【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra