Using Adapters

Adapters in Motia provide a pluggable infrastructure layer that enables horizontal scaling and custom implementations. They abstract the underlying infrastructure components (state storage, event handling, streams, and cron locking) so you can swap implementations without changing your application code.

Adapter Types

Motia supports four main adapter types:

Streams: Real-time data streams for live updates
State: Persistent key-value storage shared across steps
Events: Event-based communication between steps
Cron: Distributed cron job locking to prevent duplicate executions

Configuration

Adapters are configured in your motia.config.ts file. If no adapters are specified, Motia uses default implementations suitable for single-instance deployments.

Default Adapters (No Configuration)

motia.config.ts

import { config } from '@motiadev/core'
 
export default config({
  // Uses defaults:
  // - FileStateAdapter for state
  // - InMemoryQueueEventAdapter for events
  // - InMemoryCronAdapter for cron
  // - FileStreamAdapter for streams
})

Custom Adapters

motia.config.ts

import { config } from '@motiadev/core'
import { RedisStateAdapter } from '@motiadev/adapter-redis-state'
import { RedisStreamAdapter } from '@motiadev/adapter-redis-streams'
import { RabbitMQEventAdapter } from '@motiadev/adapter-rabbitmq-events'
import { RedisCronAdapter } from '@motiadev/adapter-redis-cron'
 
export default config({
  adapters: {
    state: new RedisStateAdapter({
      host: process.env.REDIS_HOST,
      port: 6379,
    }),
    streams: new RedisStreamAdapter({
      host: process.env.REDIS_HOST,
      port: 6379,
    }),
    events: new RabbitMQEventAdapter({
      url: process.env.RABBITMQ_URL,
    }),
    cron: new RedisCronAdapter({
      host: process.env.REDIS_HOST,
      port: 6379,
    }),
  },
})

Usage in Code

Stream Adapter

Streams are accessed via context.streams in your step handlers. Each stream is defined by a .stream.ts file and accessed by its name.

Defining a Stream:

steps/messages.stream.ts

import { z } from 'zod'
import type { StreamConfig } from 'motia'
 
export const messageSchema = z.object({
  content: z.string(),
  userId: z.string(),
  timestamp: z.number(),
})
 
export type Message = z.infer<typeof messageSchema>
 
export const config: StreamConfig = {
  name: 'messages',
  schema: messageSchema,
  baseConfig: { storageType: 'default' },
}

Using a Stream:

steps/chat.step.ts

import type { Handlers } from 'motia'
 
export const handler: Handlers['Chat'] = async (req, { streams }) => {
  // Set a stream item
  await streams.messages.set('chat-123', 'msg-1', {
    content: 'Hello!',
    userId: 'user-1',
    timestamp: Date.now(),
  })
 
  // Get a stream item
  const message = await streams.messages.get('chat-123', 'msg-1')
 
  // Get all items in a group
  const allMessages = await streams.messages.getGroup('chat-123')
 
  // Delete a stream item
  await streams.messages.delete('chat-123', 'msg-1')
 
  return { status: 200, body: { messages: allMessages } }
}

State Adapter

State is accessed via context.state and provides persistent storage scoped to a trace ID.

steps/processor.step.ts

import type { Handlers } from 'motia'
 
export const handler: Handlers['Processor'] = async (req, { state, traceId }) => {
  // Set a value
  await state.set(traceId, 'userCount', 42)
 
  // Get a value
  const count = await state.get<number>(traceId, 'userCount')
 
  // Get all keys for a trace
  const keys = await state.keys(traceId)
 
  // Delete a value
  await state.delete(traceId, 'userCount')
 
  // Clear all state for a trace
  await state.clear(traceId)
 
  return { status: 200, body: { count } }
}

Event Adapter

Events are used to trigger other steps. Use context.emit() to send events, and configure steps with subscribes to listen for events.

Emitting Events:

steps/order.step.ts

import type { Handlers } from 'motia'
 
export const handler: Handlers['Order'] = async (req, { emit }) => {
  await emit({
    topic: 'order.created',
    data: { orderId: '123', amount: 99.99 },
  })
 
  return { status: 200, body: { success: true } }
}

Subscribing to Events:

steps/notify.step.ts

import type { EventConfig, Handlers } from 'motia'
import { z } from 'zod'
 
export const config: EventConfig = {
  name: 'Notify',
  type: 'event',
  subscribes: ['order.created'],
  input: z.object({
    orderId: z.string(),
    amount: z.number(),
  }),
}
 
export const handler: Handlers['Notify'] = async (input, { logger }) => {
  logger.info('Order received', { orderId: input.orderId })
  // Send notification...
}

Cron Adapter

Cron adapters are automatically used by Motia when cron steps run. They provide distributed locking to ensure only one instance executes a cron job, even when running multiple Motia instances.

Defining a Cron Step:

steps/daily-report.step.ts

import type { CronConfig, Handlers } from 'motia'
 
export const config: CronConfig = {
  name: 'DailyReport',
  type: 'cron',
  cron: '0 9 * * *', // Run daily at 9 AM
  flows: ['reports'],
}
 
export const handler: Handlers['DailyReport'] = async ({ emit, logger }) => {
  logger.info('Running daily report')
  await emit({
    topic: 'report.generated',
    data: { date: new Date().toISOString() },
  })
}

The cron adapter automatically:

Acquires a lock before execution
Releases the lock after completion
Prevents duplicate executions across multiple instances

No additional code is needed to use the cron adapter - it works automatically when you configure a cron step.

Available Implementations

Official Packages

@motiadev/adapter-redis-state - Redis-based state storage
@motiadev/adapter-redis-streams - Redis streams implementation
@motiadev/adapter-redis-cron - Redis distributed locking for cron
@motiadev/adapter-rabbitmq-events - RabbitMQ event handling

Default Implementations

These are included in @motiadev/core and work without external dependencies:

FileStateAdapter - File-based state (default)
MemoryStateAdapter - In-memory state (testing)
FileStreamAdapter - File-based streams (default)
MemoryStreamAdapter - In-memory streams (testing)
InMemoryQueueEventAdapter - In-memory event queue (default)
InMemoryCronAdapter - Single-instance cron locking (default)

When to Use Custom Adapters

Use default adapters when:

Running a single Motia instance
Local development
Testing

Use distributed adapters (Redis, RabbitMQ) when:

Running multiple Motia instances
Need horizontal scaling
Production deployments
High availability requirements