realtime-streaming

Real-time Streaming

Building event driven applications often requires some real-time streaming capabilities.

• Like integration with LLMs, they should be implemented in an asynchronous way and updates should come in real-time.
• Chat applications, real-time collaboration, etc.
• Long living processes like data processing, video processing, etc.

Motia has a built-in real-time streaming system that allows you to easily implement real-time streaming capabilities in your application.

It's called Streams.

Stream Configuration

Creating a Stream means defining a data schema that will be stored and served to the clients who are subscribing.

StreamConfig Type Definition

export type ZodInput = ZodObject<any> | ZodArray<any>
export type StepSchemaInput = ZodInput | JsonSchema

export type StreamSubscription = { groupId: string; id?: string }

export interface StreamConfig {
  /**
   * The stream name, used on the client side to subscribe.
   */
  name: string

  /**
   * Schema for stream data. Accepts either:
   * - Zod schema (ZodObject or ZodArray)
   * - JSON Schema object
   */
  schema: StepSchemaInput

  /**
   * Storage configuration for the stream.
   * Use 'default' for built-in storage or 'custom' with a factory function.
   */
  baseConfig: { storageType: 'default' } | { storageType: 'custom'; factory: () => MotiaStream<any> }

  /**
   * Optional: Access control callback to authorize subscriptions.
   * Return true to allow access, false to deny.
   */
  canAccess?: (subscription: StreamSubscription, authContext: any) => boolean | Promise<boolean>
}

TypeScript Example

// steps/streams/chat-messages.stream.ts
import { StreamConfig } from 'motia'
import { z } from 'zod'

export const chatMessageSchema = z.object({
  id: z.string(),
  userId: z.string(),
  message: z.string(),
  timestamp: z.string()
})

export type ChatMessage = z.infer<typeof chatMessageSchema>

export const config: StreamConfig = {
  name: 'chatMessage',
  schema: chatMessageSchema,
  baseConfig: { storageType: 'default' },
}