Cloudflare D1 存储

Cloudflare D1 存储实现使用 Cloudflare D1 提供 serverless SQL 数据库方案，支持关系操作和事务一致性。

不支持 Observability

Cloudflare D1 存储不支持 observability 域。来自 MastraStorageExporter 的 Trace 无法持久化到 D1，并且仅将 D1 作为存储 Provider 时，Studio 的 observability 功能无法使用。若要启用 observability，请使用组合存储将 observability 数据路由到 ClickHouse 等受支持的 Provider。

行大小限制

Cloudflare D1 强制实施 1 MiB 的最大行大小。存储带有图片等 base64 编码附件的消息时可能超出此限制。请参阅处理大型附件，了解包括上传附件到外部存储在内的解决方法。

安装

npm

npm install @mastra/cloudflare-d1@latest

pnpm

pnpm add @mastra/cloudflare-d1@latest

Yarn

yarn add @mastra/cloudflare-d1@latest

Bun

bun add @mastra/cloudflare-d1@latest

使用方法

与 Mastra CloudflareDeployer 配合使用

在 Cloudflare 上将 D1Store 与 Mastra 配合使用的标准方式是使用 CloudflareDeployer。从 cloudflare:workers 导入 env，并在 new Mastra({...}) 内联初始化 D1Store。

src/mastra/index.ts

import { env } from 'cloudflare:workers'
import { D1Store } from '@mastra/cloudflare-d1'
import { Mastra } from '@mastra/core'
import { CloudflareDeployer } from '@mastra/deployer-cloudflare'

export const mastra = new Mastra({
  storage: new D1Store({ binding: env.DB }),
  deployer: new CloudflareDeployer({
    name: 'my-worker',
    d1_databases: [
      {
        binding: 'DB',
        database_name: 'your-database-name',
        database_id: 'your-database-id',
      },
    ],
  }),
})

备注

使用 import { env } from 'cloudflare:workers' 时，必须在 new Mastra({...}) 内联初始化 D1Store，而不能提取到模块级变量。或者，在 env 可用后于 fetch handler 内初始化 D1Store。详情请参阅 CloudflareDeployer 参考。

在没有 HTTP routes 的 Cloudflare Worker 中使用

如果希望在 Worker 中直接调用 Mastra（例如运行 Agent 或触发 Workflow）且不提供 HTTP routes，则无需 CloudflareDeployer。从 worker 的 env 参数访问 D1 binding，并以编程方式调用 Mastra。

import { D1Store } from '@mastra/cloudflare-d1'
import { Mastra } from '@mastra/core'

type Env = {
  DB: D1Database
}

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext) {
    const mastra = new Mastra({
      storage: new D1Store({ binding: env.DB }),
    })

    const agent = mastra.getAgent('my-agent')
    const result = await agent.generate('Hello')

    return Response.json({ text: result.text })
  },
}

使用 REST API

对于非 Workers 环境（Node.js、serverless functions 等），请使用 REST API 方式：

import { D1Store } from '@mastra/cloudflare-d1'

const storage = new D1Store({
  accountId: process.env.CLOUDFLARE_ACCOUNT_ID!, // Cloudflare Account ID
  databaseId: process.env.CLOUDFLARE_D1_DATABASE_ID!, // D1 Database ID
  apiToken: process.env.CLOUDFLARE_API_TOKEN!, // Cloudflare API Token
  tablePrefix: 'dev_', // Optional: isolate tables per environment
})

Wrangler 配置

将 D1 数据库 binding 添加到 wrangler.toml：

[[d1_databases]]
binding = "DB"
database_name = "your-database-name"
database_id = "your-database-id"

或在 wrangler.jsonc 中：

{
  "d1_databases": [
    {
      "binding": "DB",
      "database_name": "your-database-name",
      "database_id": "your-database-id",
    },
  ],
}

参数

binding?:

D1Database

Cloudflare D1 Workers binding（用于 Workers runtime）

accountId?:

string

Cloudflare Account ID（用于 REST API）

databaseId?:

string

Cloudflare D1 Database ID（用于 REST API）

apiToken?:

string

Cloudflare API Token（用于 REST API）

tablePrefix?:

string

所有表名的可选前缀（有助于隔离环境）

补充说明

Schema 管理

该存储实现会自动处理 schema 创建和更新。它会创建以下表：

• threads：存储对话线程
• messages：存储单条消息
• metadata：存储线程和消息的附加元数据

初始化

将 storage 传递给 Mastra class 时，会在任何存储操作之前自动调用 init()：

import { Mastra } from '@mastra/core'
import { D1Store } from '@mastra/cloudflare-d1'

type Env = {
  DB: D1Database
}

// In a Cloudflare Worker
export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext) {
    const storage = new D1Store({
      binding: env.DB,
    })

    const mastra = new Mastra({
      storage, // init() is called automatically
    })

    // Your handler logic here
    return new Response('Success')
  },
}

如果不通过 Mastra 而直接使用 storage，必须显式调用 init() 来创建表：

import { D1Store } from '@mastra/cloudflare-d1'

type Env = {
  DB: D1Database
}

// In a Cloudflare Worker
export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext) {
    const storage = new D1Store({
      id: 'd1-storage',
      binding: env.DB,
    })

    // Required when using storage directly
    await storage.init()

    // Access domain-specific stores via getStore()
    const memoryStore = await storage.getStore('memory')
    const thread = await memoryStore?.getThreadById({ threadId: '...' })

    return new Response('Success')
  },
}

注意

如果未调用 init()，不会创建表，存储操作将静默失败或抛出错误。

事务和一致性

Cloudflare D1 为单行操作提供事务保证。多个操作可作为单个全有或全无的工作单元执行。

表创建和迁移

初始化 storage 时会自动创建表（并可使用 tablePrefix 选项按环境隔离），但高级 schema 变更需要手动迁移和谨慎规划。例如，添加列或更改数据类型和索引时，需要避免数据丢失。