diff --git a/src/modules/Elsa.Workflows.Core/Builders/WorkflowBuilder.cs b/src/modules/Elsa.Workflows.Core/Builders/WorkflowBuilder.cs
index 878de48b8..00398ce8c 100644
--- a/src/modules/Elsa.Workflows.Core/Builders/WorkflowBuilder.cs
+++ b/src/modules/Elsa.Workflows.Core/Builders/WorkflowBuilder.cs
@@ -114,14 +114,7 @@ public class WorkflowBuilder(IActivityVisitor activityVisitor, IIdentityGraphSer
     /// <inheritdoc />
     public InputDefinition WithInput<T>(string name, string? description = default)
     {
-        return WithInput(inputDefinition =>
-        {
-            inputDefinition.Name = name;
-            inputDefinition.Type = typeof(T);
-
-            if (description != null)
-                inputDefinition.Description = description;
-        });
+        return WithInput(name, typeof(T), description);
     }
 
     /// <inheritdoc />
@@ -164,6 +157,46 @@ public class WorkflowBuilder(IActivityVisitor activityVisitor, IIdentityGraphSer
         return this;
     }
 
+    public OutputDefinition WithOutput<T>(string name, string? description = default)
+    {
+        return WithOutput(name, typeof(T), description);
+    }
+
+    public OutputDefinition WithOutput(string name, Type type, string? description = default)
+    {
+        return WithOutput(outputDefinition =>
+        {
+            outputDefinition.Name = name;
+            outputDefinition.Type = type;
+
+            if (description != null)
+                outputDefinition.Description = description;
+        });
+    }
+
+    public OutputDefinition WithOutput(string name, Type type, Action<OutputDefinition>? setup = default)
+    {
+        return WithOutput(outputDefinition =>
+        {
+            outputDefinition.Name = name;
+            outputDefinition.Type = type;
+            setup?.Invoke(outputDefinition);
+        });
+    }
+
+    public OutputDefinition WithOutput(Action<OutputDefinition> setup)
+    {
+        var outputDefinition = new OutputDefinition();
+        setup(outputDefinition);
+        return WithOutput(outputDefinition);
+    }
+
+    public OutputDefinition WithOutput(OutputDefinition outputDefinition)
+    {
+        Outputs.Add(outputDefinition);
+        return outputDefinition;
+    }
+
     /// <inheritdoc />
     public IWorkflowBuilder WithCustomProperty(string name, object value)
     {
diff --git a/src/modules/Elsa.Workflows.Core/Contracts/IWorkflowBuilder.cs b/src/modules/Elsa.Workflows.Core/Contracts/IWorkflowBuilder.cs
index 73dc2fd92..20207f590 100644
--- a/src/modules/Elsa.Workflows.Core/Contracts/IWorkflowBuilder.cs
+++ b/src/modules/Elsa.Workflows.Core/Contracts/IWorkflowBuilder.cs
@@ -145,6 +145,31 @@ public interface IWorkflowBuilder
     /// A fluent method for adding an input to <see cref="Inputs"/>.
     /// </summary>
     IWorkflowBuilder WithInput(InputDefinition inputDefinition);
+    
+    /// <summary>
+    /// A fluent method for adding an output to <see cref="Outputs"/>.
+    /// </summary>
+    OutputDefinition WithOutput<T>(string name, string? description = default);
+    
+    /// <summary>
+    /// A fluent method for adding an output to <see cref="Outputs"/>.
+    /// </summary>
+    OutputDefinition WithOutput(string name, Type type, string? description = default);
+    
+    /// <summary>
+    /// A fluent method for adding an output to <see cref="Outputs"/>.
+    /// </summary>
+    OutputDefinition WithOutput(string name, Type type, Action<OutputDefinition>? setup = default);
+    
+    /// <summary>
+    /// A fluent method for adding an output to <see cref="Outputs"/>.
+    /// </summary>
+    OutputDefinition WithOutput(Action<OutputDefinition> setup);
+    
+    /// <summary>
+    /// A fluent method for adding an output to <see cref="Outputs"/>.
+    /// </summary>
+    OutputDefinition WithOutput(OutputDefinition outputDefinition);
 
     /// <summary>
     /// A fluent method for adding a property to <see cref="CustomProperties"/>.
diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs b/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs
new file mode 100644
index 000000000..263cc1e64
--- /dev/null
+++ b/src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs
@@ -0,0 +1,91 @@
+using System.Runtime.CompilerServices;
+using Elsa.Common.Models;
+using Elsa.Extensions;
+using Elsa.Workflows.Attributes;
+using Elsa.Workflows.Contracts;
+using Elsa.Workflows.Management;
+using Elsa.Workflows.Models;
+using Elsa.Workflows.Options;
+using Elsa.Workflows.UIHints;
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.Runtime.Activities;
+
+/// <summary>
+/// Creates a new workflow instance of the specified workflow and dispatches it for execution.
+/// </summary>
+[Activity("Elsa", "Composition", "Create a new workflow instance of the specified workflow and execute it.", Kind = ActivityKind.Task)]
+[UsedImplicitly]
+public class ExecuteWorkflow : Activity<ExecuteWorkflowResult>
+{
+    /// <inheritdoc />
+    public ExecuteWorkflow([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
+    {
+    }
+
+    /// <summary>
+    /// The definition ID of the workflow to execute. 
+    /// </summary>
+    [Input(
+        DisplayName = "Workflow Definition",
+        Description = "The definition ID of the workflow to execute.",
+        UIHint = InputUIHints.WorkflowDefinitionPicker
+    )]
+    public Input<string> WorkflowDefinitionId { get; set; } = default!;
+
+    /// <summary>
+    /// The correlation ID to associate the workflow with. 
+    /// </summary>
+    [Input(
+        DisplayName = "Correlation ID",
+        Description = "The correlation ID to associate the workflow with."
+    )]
+    public Input<string?> CorrelationId { get; set; } = default!;
+
+    /// <summary>
+    /// The input to send to the workflow.
+    /// </summary>
+    [Input(Description = "The input to send to the workflow.")]
+    public Input<IDictionary<string, object>?> Input { get; set; } = default!;
+
+    /// <inheritdoc />
+    protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
+    {
+        var result = await ExecuteWorkflowAsync(context);
+        context.SetResult(result);
+        await context.CompleteActivityAsync();
+    }
+
+    private async ValueTask<ExecuteWorkflowResult> ExecuteWorkflowAsync(ActivityExecutionContext context)
+    {
+        var workflowDefinitionId = WorkflowDefinitionId.Get(context);
+        var input = Input.GetOrDefault(context) ?? new Dictionary<string, object>();
+        var correlationId = CorrelationId.GetOrDefault(context);
+        var workflowInvoker = context.GetRequiredService<IWorkflowRunner>();
+        var identityGenerator = context.GetRequiredService<IIdentityGenerator>();
+        var workflowDefinitionService = context.GetRequiredService<IWorkflowDefinitionService>();
+        var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published, context.CancellationToken);
+
+        if (workflowGraph == null)
+            throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found.");
+
+        var options = new RunWorkflowOptions
+        {
+            ParentWorkflowInstanceId = context.WorkflowExecutionContext.Id,
+            Input = input,
+            CorrelationId = correlationId,
+            WorkflowInstanceId = identityGenerator.GenerateId()
+        };
+
+        var workflowResult = await workflowInvoker.RunAsync(workflowGraph, options, context.CancellationToken);
+        var info = new ExecuteWorkflowResult
+        {
+            WorkflowInstanceId = options.WorkflowInstanceId,
+            Status = workflowResult.WorkflowState.Status,
+            SubStatus = workflowResult.WorkflowState.SubStatus,
+            Output = workflowResult.WorkflowState.Output
+        };
+
+        return info;
+    }
+}
\ No newline at end of file
diff --git a/src/modules/Elsa.Workflows.Runtime/Models/ExecuteWorkflowResult.cs b/src/modules/Elsa.Workflows.Runtime/Models/ExecuteWorkflowResult.cs
new file mode 100644
index 000000000..b2bcd36e7
--- /dev/null
+++ b/src/modules/Elsa.Workflows.Runtime/Models/ExecuteWorkflowResult.cs
@@ -0,0 +1,14 @@
+namespace Elsa.Workflows.Runtime;
+
+/// <summary>
+/// Represents the result of executing a workflow.
+/// </summary>
+public class ExecuteWorkflowResult
+{
+    public string WorkflowDefinitionVersionId { get; set; } = default!;
+    public string WorkflowInstanceId { get; set; } = default!;
+    public string? CorrelationId { get; set; }
+    public WorkflowStatus Status { get; set; }
+    public WorkflowSubStatus SubStatus { get; set; }
+    public IDictionary<string, object>? Output { get; set; }
+}
\ No newline at end of file
diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/ExecuteWorkflows/ExecuteWorkflowsTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/ExecuteWorkflows/ExecuteWorkflowsTests.cs
new file mode 100644
index 000000000..4ad6f0242
--- /dev/null
+++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/ExecuteWorkflows/ExecuteWorkflowsTests.cs
@@ -0,0 +1,21 @@
+using Elsa.Workflows.ComponentTests.Scenarios.ExecuteWorkflows.Workflows;
+using Elsa.Workflows.Contracts;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.ExecuteWorkflows;
+
+public class ExecuteWorkflowsTests : AppComponentTest
+{
+    private readonly IWorkflowRunner _workflowRunner;
+
+    public ExecuteWorkflowsTests(App app) : base(app)
+    {
+        _workflowRunner = Scope.ServiceProvider.GetRequiredService<IWorkflowRunner>();
+    }
+
+    [Fact]
+    public async Task ExecuteWorkflow_ShouldExecuteWorkflow()
+    {
+        await _workflowRunner.RunAsync<MainWorkflow>();
+    }
+}
\ No newline at end of file
diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/ExecuteWorkflows/Workflows/MainWorkflow.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/ExecuteWorkflows/Workflows/MainWorkflow.cs
new file mode 100644
index 000000000..f9b08a3a6
--- /dev/null
+++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/ExecuteWorkflows/Workflows/MainWorkflow.cs
@@ -0,0 +1,34 @@
+using System.Text.Json;
+using Elsa.Workflows.Activities;
+using Elsa.Workflows.Contracts;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Activities;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.ExecuteWorkflows.Workflows;
+
+public class MainWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = Guid.NewGuid().ToString();
+
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+        var workflowResult = builder.WithVariable<ExecuteWorkflowResult>();
+        builder.Root = new Sequence
+        {
+            Activities =
+            {
+                new ExecuteWorkflow
+                {
+                    WorkflowDefinitionId = new(SubroutineWorkflow.DefinitionId),
+                    Input = new(new Dictionary<string, object>
+                    {
+                        ["Value"] = 21
+                    }),
+                    Result = new(workflowResult)
+                },
+                new WriteLine(context => $"Subroutine output: {JsonSerializer.Serialize(workflowResult.Get(context))}")
+            }
+        };
+    }
+}
\ No newline at end of file
diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/ExecuteWorkflows/Workflows/SubroutineWorkflow.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/ExecuteWorkflows/Workflows/SubroutineWorkflow.cs
new file mode 100644
index 000000000..0adcb58dc
--- /dev/null
+++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/ExecuteWorkflows/Workflows/SubroutineWorkflow.cs
@@ -0,0 +1,32 @@
+using Elsa.Extensions;
+using Elsa.Workflows.Activities;
+using Elsa.Workflows.Contracts;
+using Elsa.Workflows.Management.Activities.SetOutput;
+using Hangfire.Annotations;
+
+namespace Elsa.Workflows.ComponentTests.Scenarios.ExecuteWorkflows.Workflows;
+
+[UsedImplicitly]
+public class SubroutineWorkflow : WorkflowBase
+{
+    public static readonly string DefinitionId = Guid.NewGuid().ToString(); 
+    protected override void Build(IWorkflowBuilder builder)
+    {
+        builder.WithDefinitionId(DefinitionId);
+        var valueInput = builder.WithInput<double>("Value");
+        var output = builder.WithOutput<double>("Output");
+        
+        builder.Root = new Sequence
+        {
+            Activities =
+            {
+                new WriteLine(context => $"Running subroutine on value {context.GetInput<double>(valueInput)}..."),
+                new SetOutput
+                {
+                    OutputName = new(output.Name),
+                    OutputValue = new(context => context.GetInput<double>(valueInput) * 2)
+                }
+            }
+        };
+    }
+}
\ No newline at end of file