diff --git a/integration-tests/modules/__tests__/order/workflows/create-return.spec.ts b/integration-tests/modules/__tests__/order/workflows/create-return.spec.ts
new file mode 100644
index 0000000000..858f801626
--- /dev/null
+++ b/integration-tests/modules/__tests__/order/workflows/create-return.spec.ts
@@ -0,0 +1,473 @@
+import { ModuleRegistrationName, Modules } from "@medusajs/modules-sdk"
+import {
+  FulfillmentWorkflow,
+  IOrderModuleService,
+  IRegionModuleService,
+  IStockLocationServiceNext,
+  OrderWorkflow,
+  ProductDTO,
+  RegionDTO,
+  ShippingOptionDTO,
+  StockLocationDTO,
+} from "@medusajs/types"
+import { medusaIntegrationTestRunner } from "medusa-test-utils/dist"
+import {
+  createReturnOrderWorkflow,
+  createShippingOptionsWorkflow,
+} from "@medusajs/core-flows"
+import {
+  ContainerRegistrationKeys,
+  remoteQueryObjectFromString,
+  RuleOperator,
+} from "@medusajs/utils"
+
+jest.setTimeout(500000)
+
+const env = { MEDUSA_FF_MEDUSA_V2: true }
+const providerId = "manual_test-provider"
+
+async function prepareDataFixtures({ container }) {
+  const fulfillmentService = container.resolve(
+    ModuleRegistrationName.FULFILLMENT
+  )
+  const salesChannelService = container.resolve(
+    ModuleRegistrationName.SALES_CHANNEL
+  )
+  const stockLocationModule: IStockLocationServiceNext = container.resolve(
+    ModuleRegistrationName.STOCK_LOCATION
+  )
+  const productModule = container.resolve(ModuleRegistrationName.PRODUCT)
+  const inventoryModule = container.resolve(ModuleRegistrationName.INVENTORY)
+
+  const shippingProfile = await fulfillmentService.createShippingProfiles({
+    name: "test",
+    type: "default",
+  })
+
+  const fulfillmentSet = await fulfillmentService.create({
+    name: "Test fulfillment set",
+    type: "manual_test",
+  })
+
+  const serviceZone = await fulfillmentService.createServiceZones({
+    name: "Test service zone",
+    fulfillment_set_id: fulfillmentSet.id,
+    geo_zones: [
+      {
+        type: "country",
+        country_code: "US",
+      },
+    ],
+  })
+
+  const regionService = container.resolve(
+    ModuleRegistrationName.REGION
+  ) as IRegionModuleService
+
+  const [region] = await regionService.create([
+    {
+      name: "Test region",
+      currency_code: "eur",
+      countries: ["fr"],
+    },
+  ])
+
+  const salesChannel = await salesChannelService.create({
+    name: "Webshop",
+  })
+
+  const location: StockLocationDTO = await stockLocationModule.create({
+    name: "Warehouse",
+    address: {
+      address_1: "Test",
+      city: "Test",
+      country_code: "US",
+      postal_code: "12345",
+      phone: "12345",
+    },
+  })
+
+  const [product] = await productModule.create([
+    {
+      title: "Test product",
+      variants: [
+        {
+          title: "Test variant",
+          sku: "test-variant",
+        },
+      ],
+    },
+  ])
+
+  const inventoryItem = await inventoryModule.create({
+    sku: "inv-1234",
+  })
+
+  await inventoryModule.createInventoryLevels([
+    {
+      inventory_item_id: inventoryItem.id,
+      location_id: location.id,
+      stocked_quantity: 2,
+      reserved_quantity: 0,
+    },
+  ])
+
+  const remoteLink = container.resolve(ContainerRegistrationKeys.REMOTE_LINK)
+
+  await remoteLink.create([
+    {
+      [Modules.STOCK_LOCATION]: {
+        stock_location_id: location.id,
+      },
+      [Modules.FULFILLMENT]: {
+        fulfillment_set_id: fulfillmentSet.id,
+      },
+    },
+    {
+      [Modules.SALES_CHANNEL]: {
+        sales_channel_id: salesChannel.id,
+      },
+      [Modules.STOCK_LOCATION]: {
+        stock_location_id: location.id,
+      },
+    },
+    {
+      [Modules.PRODUCT]: {
+        variant_id: product.variants[0].id,
+      },
+      [Modules.INVENTORY]: {
+        inventory_item_id: inventoryItem.id,
+      },
+    },
+  ])
+
+  const shippingOptionData: FulfillmentWorkflow.CreateShippingOptionsWorkflowInput =
+    {
+      name: "Return shipping option",
+      price_type: "flat",
+      service_zone_id: serviceZone.id,
+      shipping_profile_id: shippingProfile.id,
+      provider_id: providerId,
+      type: {
+        code: "manual-type",
+        label: "Manual Type",
+        description: "Manual Type Description",
+      },
+      prices: [
+        {
+          currency_code: "usd",
+          amount: 10,
+        },
+        {
+          region_id: region.id,
+          amount: 100,
+        },
+      ],
+      rules: [
+        {
+          attribute: "is_return",
+          operator: RuleOperator.EQ,
+          value: '"true"',
+        },
+      ],
+    }
+
+  const { result } = await createShippingOptionsWorkflow(container).run({
+    input: [shippingOptionData],
+  })
+
+  const remoteQueryObject = remoteQueryObjectFromString({
+    entryPoint: "shipping_option",
+    variables: {
+      id: result[0].id,
+    },
+    fields: [
+      "id",
+      "name",
+      "price_type",
+      "service_zone_id",
+      "shipping_profile_id",
+      "provider_id",
+      "data",
+      "metadata",
+      "type.*",
+      "created_at",
+      "updated_at",
+      "deleted_at",
+      "shipping_option_type_id",
+      "prices.*",
+    ],
+  })
+
+  const remoteQuery = container.resolve(ContainerRegistrationKeys.REMOTE_QUERY)
+
+  const [createdShippingOption] = await remoteQuery(remoteQueryObject)
+  return {
+    shippingOption: createdShippingOption,
+    region,
+    salesChannel,
+    location,
+    product,
+  }
+}
+
+async function createOrderFixture({ container, product }) {
+  const orderService: IOrderModuleService = container.resolve(
+    ModuleRegistrationName.ORDER
+  )
+  let order = await orderService.create({
+    region_id: "test_region_idclear",
+    email: "foo@bar.com",
+    items: [
+      {
+        title: "Custom Item 2",
+        variant_sku: product.variants[0].sku,
+        variant_title: product.variants[0].title,
+        quantity: 1,
+        unit_price: 50,
+        adjustments: [
+          {
+            code: "VIP_25 ETH",
+            amount: "0.000000000000000005",
+            description: "VIP discount",
+            promotion_id: "prom_123",
+            provider_id: "coupon_kings",
+          },
+        ],
+      } as any,
+    ],
+    transactions: [
+      {
+        amount: 50, // TODO: check calculation, I think it should be 60 wit the shipping but the order total is 50
+        currency_code: "usd",
+      },
+    ],
+    sales_channel_id: "test",
+    shipping_address: {
+      first_name: "Test",
+      last_name: "Test",
+      address_1: "Test",
+      city: "Test",
+      country_code: "US",
+      postal_code: "12345",
+      phone: "12345",
+    },
+    billing_address: {
+      first_name: "Test",
+      last_name: "Test",
+      address_1: "Test",
+      city: "Test",
+      country_code: "US",
+      postal_code: "12345",
+    },
+    shipping_methods: [
+      {
+        name: "Test shipping method",
+        amount: 10,
+        data: {},
+        tax_lines: [
+          {
+            description: "shipping Tax 1",
+            tax_rate_id: "tax_usa_shipping",
+            code: "code",
+            rate: 10,
+          },
+        ],
+        adjustments: [
+          {
+            code: "VIP_10",
+            amount: 1,
+            description: "VIP discount",
+            promotion_id: "prom_123",
+          },
+        ],
+      },
+    ],
+    currency_code: "usd",
+    customer_id: "joe",
+  })
+
+  await orderService.addOrderAction([
+    {
+      action: "FULFILL_ITEM",
+      order_id: order.id,
+      version: order.version,
+      reference: "fullfilment",
+      reference_id: "fulfill_123",
+      details: {
+        reference_id: order.items![0].id,
+        quantity: 1,
+      },
+    },
+    {
+      action: "SHIP_ITEM",
+      order_id: order.id,
+      version: order.version,
+      reference: "fullfilment",
+      reference_id: "fulfill_123",
+      details: {
+        reference_id: order.items![0].id,
+        quantity: 1,
+      },
+    },
+  ])
+
+  await orderService.applyPendingOrderActions(order.id)
+
+  order = await orderService.retrieve(order.id, {
+    relations: ["items"],
+  })
+
+  return order
+}
+
+medusaIntegrationTestRunner({
+  env,
+  testSuite: ({ getContainer }) => {
+    let container
+
+    beforeAll(() => {
+      container = getContainer()
+    })
+
+    describe("Create return order workflow", () => {
+      let shippingOption: ShippingOptionDTO
+      let region: RegionDTO
+      let location: StockLocationDTO
+      let product: ProductDTO
+
+      let orderService: IOrderModuleService
+
+      beforeEach(async () => {
+        const fixtures = await prepareDataFixtures({
+          container,
+        })
+
+        shippingOption = fixtures.shippingOption
+        region = fixtures.region
+        location = fixtures.location
+        product = fixtures.product
+
+        orderService = container.resolve(ModuleRegistrationName.ORDER)
+      })
+
+      it("should create a return order", async () => {
+        const order = await createOrderFixture({ container, product })
+        const createReturnOrderData: OrderWorkflow.CreateOrderReturnWorkflowInput =
+          {
+            order_id: order.id,
+            return_shipping: {
+              option_id: shippingOption.id,
+            },
+            items: [
+              {
+                id: order.items![0].id,
+                quantity: 1,
+              },
+            ],
+          }
+
+        await createReturnOrderWorkflow(container).run({
+          input: createReturnOrderData,
+          throwOnError: false,
+        })
+
+        const remoteQuery = container.resolve(
+          ContainerRegistrationKeys.REMOTE_QUERY
+        )
+        const remoteQueryObject = remoteQueryObjectFromString({
+          entryPoint: "order",
+          variables: {
+            id: order.id,
+          },
+          fields: [
+            "*",
+            "items.*",
+            "shipping_methods.*",
+            "total",
+            "item_total",
+            "fulfillments.*",
+          ],
+        })
+
+        const [returnOrder] = await remoteQuery(remoteQueryObject)
+
+        expect(returnOrder).toEqual(
+          expect.objectContaining({
+            id: expect.any(String),
+            display_id: 1,
+            region_id: "test_region_idclear",
+            customer_id: "joe",
+            version: 2,
+            sales_channel_id: "test", // TODO: What about order with a sales channel but a shipping option link to a stock from another channel?
+            status: "pending",
+            is_draft_order: false,
+            email: "foo@bar.com",
+            currency_code: "usd",
+            shipping_address_id: expect.any(String),
+            billing_address_id: expect.any(String),
+            items: [
+              expect.objectContaining({
+                id: order.items![0].id,
+                title: "Custom Item 2",
+                variant_sku: product.variants[0].sku,
+                variant_title: product.variants[0].title,
+                requires_shipping: true,
+                is_discountable: true,
+                is_tax_inclusive: false,
+                compare_at_unit_price: null,
+                unit_price: 50,
+                quantity: 1,
+                detail: expect.objectContaining({
+                  id: expect.any(String),
+                  order_id: expect.any(String),
+                  version: 2,
+                  item_id: expect.any(String),
+                  quantity: 1,
+                  fulfilled_quantity: 1,
+                  shipped_quantity: 1,
+                  return_requested_quantity: 1,
+                  return_received_quantity: 0,
+                  return_dismissed_quantity: 0,
+                  written_off_quantity: 0,
+                }),
+              }),
+            ],
+            shipping_methods: expect.arrayContaining([
+              expect.objectContaining({
+                id: expect.any(String),
+                name: "Test shipping method",
+                description: null,
+                is_tax_inclusive: false,
+                shipping_option_id: null,
+                amount: 10,
+                order_id: expect.any(String),
+              }),
+              expect.objectContaining({
+                id: expect.any(String),
+                name: shippingOption.name,
+                description: null,
+                is_tax_inclusive: false,
+                shipping_option_id: shippingOption.id,
+                amount: 10,
+                order_id: expect.any(String),
+              }),
+            ]),
+            fulfillments: [
+              expect.objectContaining({
+                id: expect.any(String),
+                location_id: location.id,
+                provider_id: providerId,
+                shipping_option_id: shippingOption.id,
+                // TODO: Validate the address once we are fixed on it
+                /*delivery_address: {
+                  id: "fuladdr_01HY0RTAP0P1EEAFK7BXJ0BKBN",
+                },*/
+              }),
+            ],
+          })
+        )
+      })
+    })
+  },
+})
diff --git a/packages/core/core-flows/src/common/index.ts b/packages/core/core-flows/src/common/index.ts
index e99533ae88..013c43af22 100644
--- a/packages/core/core-flows/src/common/index.ts
+++ b/packages/core/core-flows/src/common/index.ts
@@ -1,2 +1,3 @@
 export * from "./steps/remove-remote-links"
 export * from "./steps/use-remote-query"
+export * from "./steps/create-remote-links"
diff --git a/packages/core/core-flows/src/common/steps/create-remote-links.ts b/packages/core/core-flows/src/common/steps/create-remote-links.ts
new file mode 100644
index 0000000000..08002cc391
--- /dev/null
+++ b/packages/core/core-flows/src/common/steps/create-remote-links.ts
@@ -0,0 +1,29 @@
+import { LinkDefinition, RemoteLink } from "@medusajs/modules-sdk"
+import { createStep, StepResponse } from "@medusajs/workflows-sdk"
+
+import { ContainerRegistrationKeys } from "@medusajs/utils"
+
+type CreateRemoteLinksStepInput = LinkDefinition[]
+
+export const createLinksStepId = "create-links"
+export const createLinkStep = createStep(
+  createLinksStepId,
+  async (data: CreateRemoteLinksStepInput, { container }) => {
+    const link = container.resolve<RemoteLink>(
+      ContainerRegistrationKeys.REMOTE_LINK
+    )
+    await link.create(data)
+
+    return new StepResponse(data, data)
+  },
+  async (createdLinks, { container }) => {
+    if (!createdLinks) {
+      return
+    }
+
+    const link = container.resolve<RemoteLink>(
+      ContainerRegistrationKeys.REMOTE_LINK
+    )
+    await link.dismiss(createdLinks)
+  }
+)
diff --git a/packages/core/core-flows/src/order/steps/create-return.ts b/packages/core/core-flows/src/order/steps/create-return.ts
new file mode 100644
index 0000000000..3d021256af
--- /dev/null
+++ b/packages/core/core-flows/src/order/steps/create-return.ts
@@ -0,0 +1,29 @@
+import { ModuleRegistrationName } from "@medusajs/modules-sdk"
+import { CreateOrderReturnDTO, IOrderModuleService } from "@medusajs/types"
+import { createStep, StepResponse } from "@medusajs/workflows-sdk"
+
+type CreateReturnStepInput = CreateOrderReturnDTO
+
+export const createReturnStepId = "create-return"
+export const createReturnStep = createStep(
+  createReturnStepId,
+  async (data: CreateReturnStepInput, { container }) => {
+    const service = container.resolve<IOrderModuleService>(
+      ModuleRegistrationName.ORDER
+    )
+
+    const created = await service.createReturn(data)
+    return new StepResponse(created, created)
+  },
+  async (createdId, { container }) => {
+    if (!createdId) {
+      return
+    }
+
+    const service = container.resolve<IOrderModuleService>(
+      ModuleRegistrationName.ORDER
+    )
+
+    // TODO: delete return
+  }
+)
diff --git a/packages/core/core-flows/src/order/steps/get-item-tax-lines.ts b/packages/core/core-flows/src/order/steps/get-item-tax-lines.ts
index 50294f0f82..3ecb11bd1e 100644
--- a/packages/core/core-flows/src/order/steps/get-item-tax-lines.ts
+++ b/packages/core/core-flows/src/order/steps/get-item-tax-lines.ts
@@ -6,12 +6,12 @@ import {
   OrderShippingMethodDTO,
   OrderWorkflowDTO,
   ShippingTaxLineDTO,
-  TaxCalculationContext,
   TaxableItemDTO,
   TaxableShippingDTO,
+  TaxCalculationContext,
 } from "@medusajs/types"
 import { MedusaError } from "@medusajs/utils"
-import { StepResponse, createStep } from "@medusajs/workflows-sdk"
+import { createStep, StepResponse } from "@medusajs/workflows-sdk"
 
 interface StepInput {
   order: OrderWorkflowDTO
@@ -104,8 +104,8 @@ export const getOrderItemTaxLinesStep = createStep(
   async (data: StepInput, { container }) => {
     const {
       order,
-      items,
-      shipping_methods: shippingMethods,
+      items = [],
+      shipping_methods: shippingMethods = [],
       force_tax_calculation: forceTaxCalculation = false,
     } = data
     const taxService = container.resolve<ITaxModuleService>(
@@ -123,15 +123,19 @@ export const getOrderItemTaxLinesStep = createStep(
       return new StepResponse(stepResponseData)
     }
 
-    stepResponseData.lineItemTaxLines = (await taxService.getTaxLines(
-      normalizeLineItemsForTax(order, items),
-      taxContext
-    )) as ItemTaxLineDTO[]
+    if (items.length) {
+      stepResponseData.lineItemTaxLines = (await taxService.getTaxLines(
+        normalizeLineItemsForTax(order, items),
+        taxContext
+      )) as ItemTaxLineDTO[]
+    }
 
-    stepResponseData.shippingMethodsTaxLines = (await taxService.getTaxLines(
-      normalizeLineItemsForShipping(order, shippingMethods),
-      taxContext
-    )) as ShippingTaxLineDTO[]
+    if (shippingMethods.length) {
+      stepResponseData.shippingMethodsTaxLines = (await taxService.getTaxLines(
+        normalizeLineItemsForShipping(order, shippingMethods),
+        taxContext
+      )) as ShippingTaxLineDTO[]
+    }
 
     return new StepResponse(stepResponseData)
   }
diff --git a/packages/core/core-flows/src/order/workflows/create-return.ts b/packages/core/core-flows/src/order/workflows/create-return.ts
new file mode 100644
index 0000000000..4cfc4d1431
--- /dev/null
+++ b/packages/core/core-flows/src/order/workflows/create-return.ts
@@ -0,0 +1,339 @@
+import {
+  CreateOrderShippingMethodDTO,
+  FulfillmentWorkflow,
+  OrderDTO,
+  OrderWorkflow,
+  ShippingOptionDTO,
+  WithCalculatedPrice,
+} from "@medusajs/types"
+import {
+  createWorkflow,
+  transform,
+  WorkflowData,
+} from "@medusajs/workflows-sdk"
+import { createLinkStep, useRemoteQueryStep } from "../../common"
+import {
+  arrayDifference,
+  ContainerRegistrationKeys,
+  isDefined,
+  MathBN,
+  MedusaError,
+  Modules,
+} from "@medusajs/utils"
+import { updateOrderTaxLinesStep } from "../steps"
+import { createReturnStep } from "../steps/create-return"
+import { createFulfillmentWorkflow } from "../../fulfillment"
+
+function throwIfOrderIsCancelled({ order }: { order: OrderDTO }) {
+  // TODO: need work, check canceled
+  if (false /*order.canceled_at*/) {
+    throw new MedusaError(
+      MedusaError.Types.INVALID_DATA,
+      `Order with id ${order.id} has been cancelled.`
+    )
+  }
+}
+
+function throwIfItemsDoesNotExistsInOrder({
+  order,
+  inputItems,
+}: {
+  order: Pick<OrderDTO, "id" | "items">
+  inputItems: OrderWorkflow.CreateOrderReturnWorkflowInput["items"]
+}) {
+  const orderItemIds = order.items?.map((i) => i.id) ?? []
+  const inputItemIds = inputItems.map((i) => i.id)
+  const diff = arrayDifference(inputItemIds, orderItemIds)
+
+  if (diff.length) {
+    throw new MedusaError(
+      MedusaError.Types.INVALID_DATA,
+      `Items with ids ${diff.join(", ")} does not exist in order with id ${
+        order.id
+      }.`
+    )
+  }
+}
+
+function validateReturnReasons(
+  {
+    orderId,
+    inputItems,
+  }: {
+    orderId: string
+    inputItems: OrderWorkflow.CreateOrderReturnWorkflowInput["items"]
+  },
+  { container }
+) {
+  const reasonIds = inputItems.map((i) => i.reason_id).filter(Boolean)
+
+  if (!reasonIds.length) {
+    return
+  }
+
+  const remoteQuery = container.resolve(ContainerRegistrationKeys.REMOTE_QUERY)
+
+  const returnReasons = remoteQuery({
+    entry_point: "return_reasons",
+    fields: ["return_reason_children.*"],
+    variables: { id: [inputItems.map((item) => item.reason_id)] },
+  })
+
+  const reasons = returnReasons.map((r) => r.id)
+  const hasInvalidReasons = reasons.filter(
+    // We do not allow for root reason to be applied
+    (reason) => reason.return_reason_children.length > 0
+  )
+  const hasNonExistingReasons = arrayDifference(reasonIds, reasons)
+
+  if (hasNonExistingReasons.length) {
+    throw new MedusaError(
+      MedusaError.Types.INVALID_DATA,
+      `Return reason with id ${hasNonExistingReasons.join(
+        ", "
+      )} does not exists.`
+    )
+  }
+
+  if (hasInvalidReasons.length()) {
+    throw new MedusaError(
+      MedusaError.Types.INVALID_DATA,
+      `Cannot apply return reason with id ${hasInvalidReasons.join(
+        ", "
+      )} to order with id ${orderId}. Return reason has nested reasons.`
+    )
+  }
+}
+
+function prepareShippingMethodData({
+  orderId,
+  inputShippingOption,
+  returnShippingOption,
+}: {
+  orderId: string
+  inputShippingOption: OrderWorkflow.CreateOrderReturnWorkflowInput["return_shipping"]
+  returnShippingOption: ShippingOptionDTO & WithCalculatedPrice
+}) {
+  const obj: CreateOrderShippingMethodDTO = {
+    name: returnShippingOption.name,
+    order_id: orderId,
+    shipping_option_id: returnShippingOption.id,
+    amount: 0,
+    data: {},
+    // Computed later in the flow
+    tax_lines: [],
+    adjustments: [],
+  }
+
+  if (isDefined(inputShippingOption.price) && inputShippingOption.price >= 0) {
+    obj.amount = inputShippingOption.price
+  } else {
+    if (returnShippingOption.price_type === "calculated") {
+      // TODO: retrieve calculated price and assign to amount
+    } else {
+      obj.amount = returnShippingOption.calculated_price.calculated_amount
+    }
+  }
+
+  return obj
+}
+
+function validateCustomRefundAmount({
+  order,
+  refundAmount,
+}: {
+  order: Pick<OrderDTO, "item_total">
+  refundAmount?: number
+}) {
+  // validate that the refund prop input is less than order.item_total (item total)
+  // TODO: Probably this amount should be retrieved from the payments linked to the order
+  if (refundAmount && MathBN.gt(refundAmount, order.item_total)) {
+    throw new MedusaError(
+      MedusaError.Types.INVALID_DATA,
+      `Refund amount cannot be greater than order total.`
+    )
+  }
+}
+
+function prepareFulfillmentData({
+  order,
+  input,
+  returnShippingOption,
+}: {
+  order: OrderDTO
+  input: OrderWorkflow.CreateOrderReturnWorkflowInput
+  returnShippingOption: {
+    id: string
+    provider_id: string
+    service_zone: { fulfillment_set: { location?: { id: string } } }
+  }
+}) {
+  const inputItems = input.items
+  const orderItemsMap = new Map<string, Required<OrderDTO>["items"][0]>(
+    order.items!.map((i) => [i.id, i])
+  )
+  const fulfillmentItems = inputItems.map((i) => {
+    const orderItem = orderItemsMap.get(i.id)!
+    return {
+      line_item_id: i.id,
+      quantity: i.quantity,
+      return_quantity: i.quantity,
+      title: orderItem.variant_title ?? orderItem.title,
+      sku: orderItem.variant_sku || "",
+      barcode: orderItem.variant_barcode || "",
+    } as FulfillmentWorkflow.CreateFulfillmentItemWorkflowDTO
+  })
+
+  let locationId: string | undefined = input.location_id
+  if (!locationId) {
+    locationId = returnShippingOption.service_zone.fulfillment_set.location?.id
+  }
+
+  if (!locationId) {
+    throw new MedusaError(
+      MedusaError.Types.INVALID_DATA,
+      `Cannot create return without stock location, either provide a location or you should link the shipping option ${returnShippingOption.id} to a stock location.`
+    )
+  }
+
+  return {
+    input: {
+      location_id: locationId,
+      provider_id: returnShippingOption.provider_id,
+      shipping_option_id: input.return_shipping.option_id,
+      items: fulfillmentItems,
+      labels: [] as FulfillmentWorkflow.CreateFulfillmentLabelWorkflowDTO[],
+      delivery_address: order.shipping_address ?? ({} as any), // TODO: should it be the stock location address?
+      order: {} as FulfillmentWorkflow.CreateFulfillmentOrderWorkflowDTO, // TODO see what todo here, is that even necessary?
+    },
+  }
+}
+
+function prepareReturnShippingOptionQueryVariables({
+  order,
+  input,
+}: {
+  order: {
+    currency_code: string
+    region_id?: string
+  }
+  input: {
+    return_shipping: OrderWorkflow.CreateOrderReturnWorkflowInput["return_shipping"]
+  }
+}) {
+  const variables = {
+    id: input.return_shipping.option_id,
+    calculated_price: {
+      context: {
+        currency_code: order.currency_code,
+      },
+    },
+  }
+
+  if (order.region_id) {
+    variables.calculated_price.context["region_id"] = order.region_id
+  }
+
+  return variables
+}
+
+export const createReturnOrderWorkflowId = "create-return-order"
+export const createReturnOrderWorkflow = createWorkflow(
+  createReturnOrderWorkflowId,
+  (
+    input: WorkflowData<OrderWorkflow.CreateOrderReturnWorkflowInput>
+  ): WorkflowData<void> => {
+    const order: OrderDTO = useRemoteQueryStep({
+      entry_point: "orders",
+      fields: [
+        "id",
+        "region_id",
+        "currency_code",
+        "total",
+        "item_total",
+        "items.*",
+      ],
+      variables: { id: input.order_id },
+      list: false,
+      throw_if_key_not_found: true,
+    })
+
+    transform({ order }, throwIfOrderIsCancelled)
+    transform(
+      { order, inputItems: input.items },
+      throwIfItemsDoesNotExistsInOrder
+    )
+    transform(
+      { orderId: input.order_id, inputItems: input.items },
+      validateReturnReasons
+    )
+    transform(
+      { order, refundAmount: input.refund_amount },
+      validateCustomRefundAmount
+    )
+
+    const returnShippingOptionsVariables = transform(
+      { input, order },
+      prepareReturnShippingOptionQueryVariables
+    )
+
+    const returnShippingOption = useRemoteQueryStep({
+      entry_point: "shipping_options",
+      fields: [
+        "id",
+        "price_type",
+        "name",
+        "provider_id",
+        "calculated_price.calculated_amount",
+        "service_zone.fulfillment_set.location.id",
+      ],
+      variables: returnShippingOptionsVariables,
+      list: false,
+      throw_if_key_not_found: true,
+    }).config({ name: "return-shipping-option" })
+
+    const shippingMethodData = transform(
+      {
+        orderId: input.order_id,
+        inputShippingOption: input.return_shipping,
+        returnShippingOption,
+      },
+      prepareShippingMethodData
+    )
+
+    createReturnStep({
+      order_id: input.order_id,
+      items: input.items,
+      shipping_method: shippingMethodData,
+      created_by: input.created_by,
+    })
+
+    updateOrderTaxLinesStep({
+      order_id: input.order_id,
+      shipping_methods: [shippingMethodData as any], // The types does not seems correct in that step and expect too many things compared to the actual needs
+    })
+
+    const fulfillmentData = transform(
+      { order, input, returnShippingOption },
+      prepareFulfillmentData
+    )
+
+    const fulfillment = createFulfillmentWorkflow.runAsStep(fulfillmentData)
+
+    // TODO call the createReturn from the fulfillment provider
+
+    const link = transform(
+      { order_id: input.order_id, fulfillment },
+      (data) => {
+        return [
+          {
+            [Modules.ORDER]: { order_id: data.order_id },
+            [Modules.FULFILLMENT]: { fulfillment_id: data.fulfillment.id },
+          },
+        ]
+      }
+    )
+
+    createLinkStep(link)
+  }
+)
diff --git a/packages/core/core-flows/src/order/workflows/index.ts b/packages/core/core-flows/src/order/workflows/index.ts
index 40bedfb9fb..cd814fee4c 100644
--- a/packages/core/core-flows/src/order/workflows/index.ts
+++ b/packages/core/core-flows/src/order/workflows/index.ts
@@ -1,2 +1,3 @@
 export * from "./create-orders"
 export * from "./update-tax-lines"
+export * from "./create-return"
diff --git a/packages/core/modules-sdk/src/remote-link.ts b/packages/core/modules-sdk/src/remote-link.ts
index 21b25d7b7f..2eb5209660 100644
--- a/packages/core/modules-sdk/src/remote-link.ts
+++ b/packages/core/modules-sdk/src/remote-link.ts
@@ -14,7 +14,7 @@ export type DeleteEntityInput = {
 }
 export type RestoreEntityInput = DeleteEntityInput
 
-type LinkDefinition = {
+export type LinkDefinition = {
   [moduleName: string]: {
     [fieldName: string]: string
   }
diff --git a/packages/core/types/src/common/index.ts b/packages/core/types/src/common/index.ts
index d3a280a643..2717f4ecc6 100644
--- a/packages/core/types/src/common/index.ts
+++ b/packages/core/types/src/common/index.ts
@@ -3,3 +3,4 @@ export * from "./rule"
 export * from "./batch"
 export * from "./config-module"
 export * from "./medusa-container"
+export * from "./with-calculated"
diff --git a/packages/core/types/src/common/with-calculated.ts b/packages/core/types/src/common/with-calculated.ts
new file mode 100644
index 0000000000..374fc116a1
--- /dev/null
+++ b/packages/core/types/src/common/with-calculated.ts
@@ -0,0 +1,3 @@
+export interface WithCalculatedPrice {
+  calculated_price: { calculated_amount: number }
+}
diff --git a/packages/core/types/src/order/common.ts b/packages/core/types/src/order/common.ts
index 9342d3517d..61b4685c38 100644
--- a/packages/core/types/src/order/common.ts
+++ b/packages/core/types/src/order/common.ts
@@ -2,6 +2,19 @@ import { BaseFilterable } from "../dal"
 import { OperatorMap } from "../dal/utils"
 import { BigNumberRawValue, BigNumberValue } from "../totals"
 
+export type ChangeActionType =
+  | "CANCEL"
+  | "CANCEL_RETURN"
+  | "FULFILL_ITEM"
+  | "ITEM_ADD"
+  | "ITEM_REMOVE"
+  | "RECEIVE_DAMAGED_RETURN_ITEM"
+  | "RECEIVE_RETURN_ITEM"
+  | "RETURN_ITEM"
+  | "SHIPPING_ADD"
+  | "SHIP_ITEM"
+  | "WRITE_OFF_ITEM"
+
 export type OrderSummaryDTO = {
   total: BigNumberValue
   subtotal: BigNumberValue
@@ -1187,7 +1200,7 @@ export interface OrderChangeActionDTO {
   /**
    * The action of the order change action
    */
-  action: string
+  action: ChangeActionType
   /**
    * The details of the order change action
    */
diff --git a/packages/core/types/src/order/mutations.ts b/packages/core/types/src/order/mutations.ts
index ba594a8db4..dcbc5327fb 100644
--- a/packages/core/types/src/order/mutations.ts
+++ b/packages/core/types/src/order/mutations.ts
@@ -1,5 +1,6 @@
 import { BigNumberInput } from "../totals"
 import {
+  ChangeActionType,
   OrderItemDTO,
   OrderLineItemDTO,
   OrderReturnReasonDTO,
@@ -300,7 +301,7 @@ export interface CreateOrderChangeActionDTO {
   version?: number
   reference?: string
   reference_id?: string
-  action: string
+  action: ChangeActionType
   internal_note?: string
   amount?: BigNumberInput
   details?: Record<string, unknown>
diff --git a/packages/core/types/src/workflow/index.ts b/packages/core/types/src/workflow/index.ts
index 4c44c78b6c..9b592a1093 100644
--- a/packages/core/types/src/workflow/index.ts
+++ b/packages/core/types/src/workflow/index.ts
@@ -9,3 +9,4 @@ export * as ProductCategoryWorkflow from "./product-category"
 export * as RegionWorkflow from "./region"
 export * as ReservationWorkflow from "./reservation"
 export * as UserWorkflow from "./user"
+export * as OrderWorkflow from "./order"
diff --git a/packages/core/types/src/workflow/order/create-return-order.ts b/packages/core/types/src/workflow/order/create-return-order.ts
new file mode 100644
index 0000000000..8ce1a3b01e
--- /dev/null
+++ b/packages/core/types/src/workflow/order/create-return-order.ts
@@ -0,0 +1,26 @@
+import { BigNumberInput } from "../../totals"
+
+interface CreateOrderReturnItem {
+  id: string
+  quantity: BigNumberInput
+  internal_note?: string
+  reason_id?: string
+  metadata?: Record<string, any>
+}
+
+export interface CreateOrderReturnWorkflowInput {
+  order_id: string
+  created_by?: string // The id of the authenticated user
+  items: CreateOrderReturnItem[]
+  return_shipping: {
+    option_id: string
+    price?: number
+  }
+  note?: string
+  receive_now?: boolean
+  refund_amount?: number
+  /**
+   * Default fallback to the shipping option location id
+   */
+  location_id?: string
+}
diff --git a/packages/core/types/src/workflow/order/index.ts b/packages/core/types/src/workflow/order/index.ts
new file mode 100644
index 0000000000..29b175eae9
--- /dev/null
+++ b/packages/core/types/src/workflow/order/index.ts
@@ -0,0 +1 @@
+export * from "./create-return-order"
diff --git a/packages/core/utils/src/link/links.ts b/packages/core/utils/src/link/links.ts
index 745da8aa52..1e4d3cec66 100644
--- a/packages/core/utils/src/link/links.ts
+++ b/packages/core/utils/src/link/links.ts
@@ -80,4 +80,10 @@ export const LINKS = {
     Modules.PAYMENT,
     "payment_collection_id"
   ),
+  OrderFulfillment: composeLinkName(
+    Modules.ORDER,
+    "order_id",
+    Modules.FULFILLMENT,
+    "fulfillment_id"
+  ),
 }
diff --git a/packages/core/utils/src/modules-sdk/decorators/module.ts b/packages/core/utils/src/modules-sdk/decorators/module.ts
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/packages/core/workflows-sdk/src/utils/composer/helpers/resolve-value.ts b/packages/core/workflows-sdk/src/utils/composer/helpers/resolve-value.ts
index b8189fb8ea..599713a6fd 100644
--- a/packages/core/workflows-sdk/src/utils/composer/helpers/resolve-value.ts
+++ b/packages/core/workflows-sdk/src/utils/composer/helpers/resolve-value.ts
@@ -57,7 +57,7 @@ export async function resolveValue(input, transactionContext) {
       )
 
       if (typeof parentRef[key] === "object") {
-        await unwrapInput(parentRef[key], parentRef[key])
+        parentRef[key] = await unwrapInput(parentRef[key], parentRef[key])
       }
     }
 
diff --git a/packages/core/workflows-sdk/src/utils/composer/type.ts b/packages/core/workflows-sdk/src/utils/composer/type.ts
index e7f2691042..31fa467ce8 100644
--- a/packages/core/workflows-sdk/src/utils/composer/type.ts
+++ b/packages/core/workflows-sdk/src/utils/composer/type.ts
@@ -201,7 +201,7 @@ export type ReturnWorkflow<
     runAsStep: ({
       input,
     }: {
-      input: TData
+      input: TData | WorkflowData<TData>
     }) => ReturnType<StepFunction<TData, TResult>>
     run: <TDataOverride = undefined, TResultOverride = undefined>(
       ...args: Parameters<
diff --git a/packages/modules/fulfillment/src/services/fulfillment-module-service.ts b/packages/modules/fulfillment/src/services/fulfillment-module-service.ts
index 27bcdba30d..0aa40c53c9 100644
--- a/packages/modules/fulfillment/src/services/fulfillment-module-service.ts
+++ b/packages/modules/fulfillment/src/services/fulfillment-module-service.ts
@@ -14,17 +14,17 @@ import {
   UpdateServiceZoneDTO,
 } from "@medusajs/types"
 import {
+  arrayDifference,
   EmitEvents,
   FulfillmentUtils,
+  getSetDifference,
   InjectManager,
   InjectTransactionManager,
+  isString,
   MedusaContext,
   MedusaError,
   Modules,
   ModulesSdkUtils,
-  arrayDifference,
-  getSetDifference,
-  isString,
   promiseAll,
 } from "@medusajs/utils"
 import {
diff --git a/packages/modules/link-modules/src/definitions/fulfillment-set-location.ts b/packages/modules/link-modules/src/definitions/fulfillment-set-location.ts
index cc39659544..579d01afdb 100644
--- a/packages/modules/link-modules/src/definitions/fulfillment-set-location.ts
+++ b/packages/modules/link-modules/src/definitions/fulfillment-set-location.ts
@@ -49,6 +49,9 @@ export const LocationFulfillmentSet: ModuleJoinerConfig = {
     },
     {
       serviceName: Modules.FULFILLMENT,
+      fieldAlias: {
+        location: "locations_link.location",
+      },
       relationship: {
         serviceName: LINKS.LocationFulfillmentSet,
         primaryKey: "fulfillment_set_id",
diff --git a/packages/modules/link-modules/src/definitions/index.ts b/packages/modules/link-modules/src/definitions/index.ts
index 288b03e548..37498fc482 100644
--- a/packages/modules/link-modules/src/definitions/index.ts
+++ b/packages/modules/link-modules/src/definitions/index.ts
@@ -12,3 +12,4 @@ export * from "./readonly"
 export * from "./region-payment-provider"
 export * from "./sales-channel-location"
 export * from "./shipping-option-price-set"
+export * from "./order-fulfillment"
diff --git a/packages/modules/link-modules/src/definitions/order-fulfillment.ts b/packages/modules/link-modules/src/definitions/order-fulfillment.ts
new file mode 100644
index 0000000000..f134c3370c
--- /dev/null
+++ b/packages/modules/link-modules/src/definitions/order-fulfillment.ts
@@ -0,0 +1,63 @@
+import { Modules } from "@medusajs/modules-sdk"
+import { ModuleJoinerConfig } from "@medusajs/types"
+import { LINKS } from "@medusajs/utils"
+
+export const OrderFulfillment: ModuleJoinerConfig = {
+  serviceName: LINKS.OrderFulfillment,
+  isLink: true,
+  databaseConfig: {
+    tableName: "order_fulfillment",
+    idPrefix: "orderful",
+  },
+  alias: [
+    {
+      name: ["order_fulfillment", "order_fulfillments"],
+      args: {
+        entity: "LinkOrderFulfillment",
+      },
+    },
+  ],
+  primaryKeys: ["id", "order_id", "fulfillment_id"],
+  relationships: [
+    {
+      serviceName: Modules.ORDER,
+      primaryKey: "id",
+      foreignKey: "order_id",
+      alias: "order",
+    },
+    {
+      serviceName: Modules.FULFILLMENT,
+      primaryKey: "id",
+      foreignKey: "fulfillment_id",
+      alias: "fulfillments",
+      args: {
+        // TODO: We are not suppose to know the module implementation here, wait for later to think about inferring it
+        methodSuffix: "Fulfillments",
+      },
+    },
+  ],
+  extends: [
+    {
+      serviceName: Modules.ORDER,
+      fieldAlias: {
+        fulfillments: "fulfillment_link.fulfillments",
+      },
+      relationship: {
+        serviceName: LINKS.OrderFulfillment,
+        primaryKey: "order_id",
+        foreignKey: "id",
+        alias: "fulfillment_link",
+        isList: true,
+      },
+    },
+    {
+      serviceName: Modules.FULFILLMENT,
+      relationship: {
+        serviceName: LINKS.OrderFulfillment,
+        primaryKey: "fulfillment_id",
+        foreignKey: "id",
+        alias: "order_link",
+      },
+    },
+  ],
+}
diff --git a/packages/modules/order/src/services/order-module-service.ts b/packages/modules/order/src/services/order-module-service.ts
index 41fa14ad26..70f2fe89ab 100644
--- a/packages/modules/order/src/services/order-module-service.ts
+++ b/packages/modules/order/src/services/order-module-service.ts
@@ -2214,7 +2214,7 @@ export default class OrderModuleService<
     if (!isString(data.shipping_method)) {
       const methods = await this.createShippingMethods(
         data.order_id,
-        data.shipping_method as any,
+        [{ order_id: data.order_id, ...data.shipping_method }],
         sharedContext
       )
       shippingMethodId = methods[0].id