# MastraStorage
[EN] Source: https://mastra.ai/docs/server-db/storage

`MastraStorage` provides a unified interface for managing:

- **Suspended Workflows**: the serialized state of suspended workflows (so they can be resumed later)
- **Memory**: threads and messages per `resourceId` in your application
- **Traces**: OpenTelemetry traces from all components of Mastra
- **Eval Datasets**: scores and scoring reasons from eval runs

<br />

<br />

<StorageOverviewImage />

Mastra provides different storage providers, but you can treat them as interchangeable. Eg, you could use libsql in development but postgres in production, and your code will work the same both ways.

## Configuration

Mastra can be configured with a default storage option:

```typescript copy
import { Mastra } from "@mastra/core/mastra";
import { LibSQLStore } from "@mastra/libsql";

const mastra = new Mastra({
  storage: new LibSQLStore({
    url: "file:./mastra.db",
  }),
});
```

If you do not specify any `storage` configuration, Mastra will not persist data across application restarts or deployments. For any
deployment beyond local testing you should provide your own storage
configuration either on `Mastra` or directly within `new Memory()`.

## Data Schema

<Tabs>
  <TabItem value="messages" label="Messages">
Stores conversation messages and their metadata. Each message belongs to a thread and contains the actual content along with metadata about the sender role and message type.

<br />
<SchemaTable
  columns={[
    {
      name: "id",
      type: "uuidv4",
      description:
        "Unique identifier for the message (format: `xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx`)",
      constraints: [{ type: "primaryKey" }, { type: "nullable", value: false }],
    },
    {
      name: "thread_id",
      type: "uuidv4",
      description: "Parent thread reference",
      constraints: [
        { type: "foreignKey", value: "threads.id" },
        { type: "nullable", value: false },
      ],
    },
    {
      name: "resourceId",
      type: "uuidv4",
      description: "ID of the resource that owns this message",
      constraints: [{ type: "nullable", value: true }],
    },
    {
      name: "content",
      type: "text",
      description:
        "JSON of the message content in V2 format. Example: `{ format: 2, parts: [...] }`",
      constraints: [{ type: "nullable", value: false }],
    },
    {
      name: "role",
      type: "text",
      description: "Enum of `user | assistant`",
      constraints: [{ type: "nullable", value: false }],
    },
    {
      name: "createdAt",
      type: "timestamp",
      description: "Used for thread message ordering",
      constraints: [{ type: "nullable", value: false }],
    },
  ]}
/>

The message `content` column contains a JSON object conforming to the `MastraMessageContentV2` type, which is designed to align closely with the AI SDK `UIMessage` message shape.

<SchemaTable
  columns={[
    {
      name: "format",
      type: "integer",
      description: "Message format version (currently 2)",
      constraints: [{ type: "nullable", value: false }],
    },
    {
      name: "parts",
      type: "array (JSON)",
      description:
        "Array of message parts (text, tool-invocation, file, reasoning, etc.). The structure of items in this array varies by `type`.",
      constraints: [{ type: "nullable", value: false }],
    },
    {
      name: "experimental_attachments",
      type: "array (JSON)",
      description: "Optional array of file attachments",
      constraints: [{ type: "nullable", value: true }],
    },
    {
      name: "content",
      type: "text",
      description: "Optional main text content of the message",
      constraints: [{ type: "nullable", value: true }],
    },
    {
      name: "toolInvocations",
      type: "array (JSON)",
      description: "Optional array summarizing tool calls and results",
      constraints: [{ type: "nullable", value: true }],
    },
    {
      name: "reasoning",
      type: "object (JSON)",
      description:
        "Optional information about the reasoning process behind the assistant's response",
      constraints: [{ type: "nullable", value: true }],
    },
    {
      name: "annotations",
      type: "object (JSON)",
      description: "Optional additional metadata or annotations",
      constraints: [{ type: "nullable", value: true }],
    },
  ]}
/>

</TabItem>

  <TabItem value="threads" label="Threads">
Groups related messages together and associates them with a resource. Contains metadata about the conversation.

<br />
<SchemaTable
  columns={[
    {
      name: "id",
      type: "uuidv4",
      description:
        "Unique identifier for the thread (format: `xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx`)",
      constraints: [{ type: "primaryKey" }, { type: "nullable", value: false }],
    },
    {
      name: "resourceId",
      type: "text",
      description:
        "Primary identifier of the external resource this thread is associated with. Used to group and retrieve related threads.",
      constraints: [{ type: "nullable", value: false }],
    },
    {
      name: "title",
      type: "text",
      description: "Title of the conversation thread",
      constraints: [{ type: "nullable", value: false }],
    },
    {
      name: "metadata",
      type: "text",
      description: "Custom thread metadata as stringified JSON. Example:",
      example: {
        category: "support",
        priority: 1,
      },
    },
    {
      name: "createdAt",
      type: "timestamp",
      constraints: [{ type: "nullable", value: false }],
    },
    {
      name: "updatedAt",
      type: "timestamp",
      description: "Used for thread ordering history",
      constraints: [{ type: "nullable", value: false }],
    },
  ]}
/>

</TabItem>
  <TabItem value="resources" label="Resources">
Stores user-specific data for resource-scoped working memory. Each resource represents a user or entity, allowing working memory to persist across all conversation threads for that user.

<br />
<SchemaTable
  columns={[
    {
      name: "id",
      type: "text",
      description:
        "Resource identifier (user or entity ID) - same as resourceId used in threads and agent calls",
      constraints: [{ type: "primaryKey" }, { type: "nullable", value: false }],
    },
    {
      name: "workingMemory",
      type: "text",
      description:
        "Persistent working memory data as Markdown text. Contains user profile, preferences, and contextual information that persists across conversation threads.",
      constraints: [{ type: "nullable", value: true }],
    },
    {
      name: "metadata",
      type: "jsonb",
      description: "Additional resource metadata as JSON. Example:",
      example: {
        preferences: { language: "en", timezone: "UTC" },
        tags: ["premium", "beta-user"],
      },
      constraints: [{ type: "nullable", value: true }],
    },
    {
      name: "createdAt",
      type: "timestamp",
      description: "When the resource record was first created",
      constraints: [{ type: "nullable", value: false }],
    },
    {
      name: "updatedAt",
      type: "timestamp",
      description: "When the working memory was last updated",
      constraints: [{ type: "nullable", value: false }],
    },
  ]}
/>

**Note**: This table is only created and used by storage adapters that support resource-scoped working memory (LibSQL, PostgreSQL, Upstash). Other storage adapters will provide helpful error messages if resource-scoped memory is attempted.

</TabItem>
  <TabItem value="workflows" label="Workflows">
When `suspend` is called on a workflow, its state is saved in the following format. When `resume` is called, that state is rehydrated.

<br />
<SchemaTable
  columns={[
    {
      name: "workflow_name",
      type: "text",
      description: "Name of the workflow",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "run_id",
      type: "uuidv4",
      description: "Unique identifier for the workflow execution. Used to track state across suspend/resume cycles (format: `xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx`)",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "snapshot",
      type: "text",
      description: "Serialized workflow state as JSON. Example:",
      example: {
        value: { currentState: 'running' },
        context: {
          stepResults: {},
          attempts: {},
          triggerData: {}
        },
        activePaths: [],
        runId: '550e8400-e29b-41d4-a716-446655440000',
        timestamp: 1648176000000
      },
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "createdAt",
      type: "timestamp",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "updatedAt",
      type: "timestamp",
      description: "Last modification time, used to track state changes during workflow execution",
      constraints: [{ type: "nullable", value: false }]
    }
  ]}
/>
  </TabItem>
  <TabItem value="evals" label="Evals">
Stores eval results from running metrics against agent outputs.

<br />
<SchemaTable
  columns={[
    {
      name: "input",
      type: "text",
      description: "Input provided to the agent",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "output",
      type: "text",
      description: "Output generated by the agent",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "result",
      type: "jsonb",
      description: "Eval result data that includes score and details. Example:",
      example: {
        score: 0.95,
        details: {
          reason: "Response accurately reflects source material",
          citations: ["page 1", "page 3"]
        }
      },
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "agent_name",
      type: "text",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "metric_name",
      type: "text",
      description: "e.g Faithfulness, Hallucination, etc.",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "instructions",
      type: "text",
      description: "System prompt or instructions for the agent",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "test_info",
      type: "jsonb",
      description: "Additional test metadata and configuration",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "global_run_id",
      type: "uuidv4",
      description: "Groups related evaluation runs (e.g. all unit tests in a CI run)",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "run_id",
      type: "uuidv4",
      description: "Unique identifier for the run being evaluated (format: `xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx`)",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "created_at",
      type: "timestamp",
      constraints: [{ type: "nullable", value: false }]
    }
  ]}
/>
  </TabItem>
  <TabItem value="traces" label="Traces">
Captures OpenTelemetry traces for monitoring and debugging.

<br />
<SchemaTable
  columns={[
    {
      name: "id",
      type: "text",
      description: "Unique trace identifier",
      constraints: [
        { type: "nullable", value: false },
        { type: "primaryKey" }
      ]
    },
    {
      name: "parentSpanId",
      type: "text",
      description: "ID of the parent span. Null if span is top level",
    },
    {
      name: "name",
      type: "text",
      description: "Hierarchical operation name (e.g. `workflow.myWorkflow.execute`, `http.request`, `database.query`)",
      constraints: [{ type: "nullable", value: false }],
    },
    {
      name: "traceId",
      type: "text",
      description: "Root trace identifier that groups related spans",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "scope",
      type: "text",
      description: "Library/package/service that created the span (e.g. `@mastra/core`, `express`, `pg`)",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "kind",
      type: "integer",
      description: "`INTERNAL` (0, within process), `CLIENT` (1, outgoing calls), `SERVER` (2, incoming calls), `PRODUCER` (3, async job creation), `CONSUMER` (4, async job processing)",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "attributes",
      type: "jsonb",
      description: "User defined key-value pairs that contain span metadata",
    },
    {
      name: "status",
      type: "jsonb",
      description: "JSON object with `code` (UNSET=0, ERROR=1, OK=2) and optional `message`. Example:",
      example: {
        code: 1,
        message: "HTTP request failed with status 500"
      }
    },
    {
      name: "events",
      type: "jsonb",
      description: "Time-stamped events that occurred during the span",
    },
    {
      name: "links",
      type: "jsonb",
      description: "Links to other related spans",
      },
    {
      name: "other",
      type: "text",
      description: "Additional OpenTelemetry span fields as stringified JSON. Example:",
      example: {
        droppedAttributesCount: 2,
        droppedEventsCount: 1,
        instrumentationLibrary: "@opentelemetry/instrumentation-http"
      }
    },
    {
      name: "startTime",
      type: "bigint",
      description: "Nanoseconds since Unix epoch when span started",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "endTime",
      type: "bigint",
      description: "Nanoseconds since Unix epoch when span ended",
      constraints: [{ type: "nullable", value: false }]
    },
    {
      name: "createdAt",
      type: "timestamp",
      constraints: [{ type: "nullable", value: false }]
    }
  ]}
/>
  </TabItem>
</Tabs>

### Querying Messages

Messages are stored in a V2 format internally, which is roughly equivalent to the AI SDK's `UIMessage` format. When querying messages using `getMessages`, you can specify the desired output format, defaulting to `v1` for backwards compatibility:

```typescript copy
// Get messages in the default V1 format (roughly equivalent to AI SDK's CoreMessage format)
const messagesV1 = await mastra
  .getStorage()
  .getMessages({ threadId: "your-thread-id" });

// Get messages in the V2 format (roughly equivalent to AI SDK's UIMessage format)
const messagesV2 = await mastra
  .getStorage()
  .getMessages({ threadId: "your-thread-id", format: "v2" });
```

You can also retrieve messages using an array of message IDs. Note that unlike `getMessages`, this defaults to the V2 format:

```typescript copy
const messagesV1 = await mastra
  .getStorage()
  .getMessagesById({ messageIds: messageIdArr, format: "v1" });

const messagesV2 = await mastra
  .getStorage()
  .getMessagesById({ messageIds: messageIdArr });
```

## Storage Providers

Mastra supports the following providers:

- For local development, check out [LibSQL Storage](/reference/storage/libsql)
- For production, check out [PostgreSQL Storage](/reference/storage/postgresql)
- For serverless deployments, check out [Upstash Storage](/reference/storage/upstash)
- For document-based storage, check out [MongoDB Storage](/reference/storage/mongodb)


---
title: "Streaming Events | Streaming"
description: "Learn about the different types of streaming events in Mastra, including text deltas, tool calls, step events, and how to handle them in your applications."
---