Instruction file imported from Inesh-Reddy/motia-framework (.cursor/rules/motia/api-steps.mdc). Copyright stays with the author.

API Steps Guide

API Steps expose HTTP endpoints that can trigger workflows and emit events.

Creating API Steps

Steps need to be created in the steps folder, it can be in subfolders.

• Steps in TS and JS should end with .step.ts and .step.js respectively.
• Steps in Python should end with _step.py.

Definition

Defining an API Step is done by two elements. Configuration and Handler.

Schema Definition

• TypeScript/JavaScript: Motia uses Zod schemas for automatic validation of request/response data
• Python: Motia uses JSON Schema format. You can optionally use Pydantic models to generate JSON Schemas and handle manual validation in your handlers

Configuration

TypeScript/JavaScript: You need to export a config constant via export const config that is a ApiRouteConfig type.

Python: You need to define a config dictionary with the same properties as the TypeScript ApiRouteConfig.

export type Emit = string | { 
  /**
   * The topic name to emit to.
   */
  topic: string;

  /**
   * Optional label for the emission, could be used for documentation or UI.
   */
  label?: string;

  /**
   * This is purely for documentation purposes, 
   * it doesn't affect the execution of the step.
   * 
   * In Workbench, it will render differently based on this value.
   */
  conditional?: boolean;
}

export interface QueryParam {
  /**
   * The name of the query parameter
   */
  name: string
  /**
   * The description of the query parameter
   */
  description: string
}

export interface ApiRouteConfig {
  /**
   * Should always be api
   */
  type: 'api'

  /**
   * A unique name for this API step, used internally and for linking handlers.
   */
  name: string

  /**
   * Optional human-readable description.
   */
  description?: string

  /**
   * The URL path for this API endpoint (e.g., '/users/:id').
   */
  path: string

  /**
   * The HTTP method for this route.
   * POST, GET, PUT, DELETE, PATCH, OPTIONS, HEAD
   */
  method: ApiRouteMethod

  /**
   * Topics this API step can emit events to.
   * Important note: All emits in the handler need to be listed here.
   */
  emits: Emit[]

  /**
   * Optional: Topics that are virtually emitted, perhaps for documentation or lineage, 
   * but not strictly required for execution.
   * 
   * In Motia Workbench, they will show up as gray connections to other steps.
   */
  virtualEmits?: Emit[]

  /**
   * Optional: Virtually subscribed topics.
   * 
   * Used by API steps when we want to chain different HTTP requests
   * that could happen sequentially
   */
  virtualSubscribes?: string[]
  /**
   * Flows are used to group multiple steps to be visible in diagrams in Workbench
   */
  flows?: string[]

  /**
   * List of middlewares that will be executed BEFORE the handler is called
   */
  middleware?: ApiMiddleware<any, any, any>[]

  /**
   * Defined with Zod library, can be a ZodObject OR a ZodArray.
   * 
   * Note: This is not validated automatically, you need to validate it in the handler.
   */
  bodySchema?: ZodInput

  /**
   * Defined with Zod library, can be a ZodObject OR a ZodArray
   * 
   * The key (number) is the HTTP status code this endpoint can return and
   * for each HTTP Status Code, you need to define a Zod schema that defines the response body
   */
  responseSchema?: Record<number, ZodInput>

  /**
   * Mostly for documentation purposes, it will show up in Endpoints section in Workbench
   */
  queryParams?: QueryParam[]

  /**
   * Files to include in the step bundle.
   * Needs to be relative to the step file.
   */
  includeFiles?: string[]
}

Handler

The handler is a function that is exported via export const handler that is a ApiRouteHandler type.

Type Definition from Motia

export interface ApiRequest<TBody = unknown> {
  /**
   * Key-value pairs of path parameters (e.g., from '/users/:id').
   */
  pathParams: Record<string, string>
  /**
   * Key-value pairs of query string parameters. Values can be string or array of strings.
   */
  queryParams: Record<string, string | string[]>
  /**
   * The parsed request body (typically an object if JSON, but can vary).
   */
  body: TBody
  /**
   * Key-value pairs of request headers. Values can be string or array of strings.
   */
  headers: Record<string, string | string[]>
}

export type ApiRouteHandler<
  /** 
   * The type defined by config['bodySchema']
   */
  TRequestBody = unknown,
  /**
   * The type defined by config['responseSchema']
   */
  TResponseBody extends ApiResponse<number, unknown> = ApiResponse<number, unknown>,
  /**
   * The type defined by config['emits'] which is dynamic depending 
   * on the topic handlers (Event Steps)
   */
  TEmitData = never,
> = (req: ApiRequest<TRequestBody>, ctx: FlowContext<TEmitData>) => Promise<TResponseBody>

Handler definition

TypeScript/JavaScript:

export const handler: Handlers['CreateResource'] = async (req, { emit, logger, state, streams }) => {
  // Implementation
}

Python:

async def handler(req, context):
    # req: dictionary containing pathParams, queryParams, body, headers
    # context: object containing emit, logger, state, streams, trace_id
    pass

Examples

TypeScript Example

import { ApiRouteConfig, Handlers } from 'motia';
import { z } from 'zod';

const bodySchema = z.object({
  title: z.string().min(1, "Title cannot be empty"),
  description: z.string().optional(),
  category: z.string().min(1, "Category is required"),
  metadata: z.record(z.any()).optional()
})

export const config: ApiRouteConfig = {
  type: 'api',
  name: 'CreateResource',
  path: '/resources',
  method: 'POST',
  emits: ['send-email'],
  flows: ['resource-management'],
  bodySchema,
  responseSchema: {
    201: z.object({
      id: z.string(),
      title: z.string(),
      category: z.string()
    }),
    400: z.object({ error: z.string() })
  }
};

export const handler: Handlers['CreateResource'] = async (req, { emit, logger }) => {
  try {
    const { title, description, category, metadata } = bodySchema.parse(req.body);
    
    // Use the logger for structured logging. It's good practice to log key events or data.
    logger.info('Attempting to create resource', { title, category });
    
    /**
     * Create files to manage service calls.
     * 
     * Let's try to use Domain Driven Design to create files to manage service calls.
     * Steps are the entry points, they're the Controllers on the DDD architecture.
     */
    const result = await service.createResource(resourceData);

    /**
     * This is how we emit events to trigger Event Steps.
     * Only use emits if the task can take a while to complete.
     * 
     * Examples of long tasks are:
     * - LLM Calls
     * - Processing big files, like images, videos, audio, etc.
     * - Sending emails
     * 
     * Other applicable examples are tasks that are likely to fail, examples:
     * - Webhook call to external systems
     * 
     * API Calls that are okay to fail gracefully can be done without emits.
     */
    await emit({
      /**
       * 'topic' must be one of the topics listed in config['emits'].
       * do not emit to topics that are not defined in Steps
       */
      topic: 'send-email',
      /**
       * 'data' is the payload of the event message. 
       * make sure the data used is compliant with the Event Step input schema
       */
      data: {
        /**
         * The data to send to the Event Step.
         */
        resource: result,
        /**
         * The user to send the email to.
         */
        user
      }
    });
    
    logger.info('Resource created successfully', { resourceId, title, category });

    // Return a response object for the HTTP request.
    return {
      status: 201, // CREATED (specified in config['responseSchema'])
      /**
       * 'body' is the JSON response body sent back to the client.
       */
      body: { 
        id: result.id,
        title: result.title,
        category: result.category,
        status: 'active'
      }
    };
  } catch (error) {
    /**
     * For one single step project, it is fine to 
     * handle ZodErrors here, on multiple steps projects,
     * it is highly recommended to handle them as a middleware
     * (defined in config['middleware'])
     */
    if (error instanceof ZodError) {
      logger.error('Resource creation failed', { error: error.message });
      return {
        status: 400,
        body: { error: 'Validation failed' }
      };
    }

    logger.error('Resource creation failed', { error: error.message });
    return {
      status: 500,
      body: { error: 'Creation failed' }
    };
  }
};

Python Example

from pydantic import BaseModel, Field
from typing import Optional, Dict, Any

class ResourceData(BaseModel):
    title: str = Field(..., min_length=1, description="Title cannot be empty")
    description: Optional[str] = None
    category: str = Field(..., min_length=1, description="Category is required")
    metadata: Optional[Dict[str, Any]] = None

class ResourceResponse(BaseModel):
    id: str
    title: str
    category: str
    status: str

class ErrorResponse(BaseModel):
    error: str

config = {
    "type": "api",
    "name": "CreateResource",
    "path": "/resources",
    "method": "POST",
    "emits": ["send-email"],
    "flows": ["resource-management"],
    "bodySchema": ResourceData.model_json_schema(),
    "responseSchema": {
        201: ResourceResponse.model_json_schema(),
        400: ErrorResponse.model_json_schema()
    }
}

async def handler(req, context):
    try:
        body = req.get("body", {})
        
        # Optional: Validate input manually using Pydantic (Motia doesn't do this automatically)
        resource_data = ResourceData(**body)
        
        context.logger.info("Attempting to create resource", {
            "title": resource_data.title, 
            "category": resource_data.category
        })
        
        # Process the resource creation
        result = await service.create_resource({
            "title": resource_data.title,
            "description": resource_data.description,
            "category": resource_data.category,
            "metadata": resource_data.metadata
        })

        # Emit event to trigger other steps
        await context.emit({
            "topic": "send-email",
            "data": {
                "resource": result,
                "user_id": "example-user"
            }
        })
        
        context.logger.info("Resource created successfully", {
            "resource_id": result.get("id"),
            "title": result.get("title"),
            "category": result.get("category")
        })

        return {
            "status": 201,
            "body": {
                "id": result.get("id"),
                "title": result.get("title"),
                "category": result.get("category"),
                "status": "active"
            }
        }
        
    except ValidationError as e:
        context.logger.error("Resource creation failed - Pydantic validation error", {"error": str(e)})
        return {
            "status": 400,
            "body": {"error": "Validation failed"}
        }
    except Exception as e:
        context.logger.error("Resource creation failed", {"error": str(e)})
        return {
            "status": 500,
            "body": {"error": "Creation failed"}
        }