diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FormatVersion.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FormatVersion.java
new file mode 100644
index 0000000..8f01a33
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FormatVersion.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.statefun.flink.core.jsonmodule;
+
+enum FormatVersion {
+  v1_0("1.0");
+
+  private String versionStr;
+
+  FormatVersion(String versionStr) {
+    this.versionStr = versionStr;
+  }
+
+  @Override
+  public String toString() {
+    return versionStr;
+  }
+
+  static FormatVersion fromString(String versionStr) {
+    switch (versionStr) {
+      case "1.0":
+        return v1_0;
+      default:
+        throw new IllegalArgumentException("Unrecognized format version: " + versionStr);
+    }
+  }
+}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModuleFactory.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModuleFactory.java
new file mode 100644
index 0000000..3616bcd
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModuleFactory.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.statefun.flink.core.jsonmodule;
+
+import java.net.URL;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.statefun.flink.common.json.Selectors;
+import org.apache.flink.statefun.sdk.spi.StatefulFunctionModule;
+
+final class JsonModuleFactory {
+
+  private JsonModuleFactory() {}
+
+  static StatefulFunctionModule create(JsonNode root, URL moduleUrl) {
+    final FormatVersion formatVersion = formatVersion(root);
+    final JsonNode spec = root.at(Pointers.MODULE_SPEC);
+
+    switch (formatVersion) {
+      case v1_0:
+        return new JsonModule(spec, moduleUrl);
+      default:
+        throw new IllegalArgumentException("Unrecognized format version: " + formatVersion);
+    }
+  }
+
+  private static FormatVersion formatVersion(JsonNode root) {
+    String versionStr = Selectors.textAt(root, Pointers.FORMAT_VERSION);
+    return FormatVersion.fromString(versionStr);
+  }
+}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/Pointers.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/Pointers.java
index cc7caec..5c5b973 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/Pointers.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/Pointers.java
@@ -23,6 +23,8 @@
 
   private Pointers() {}
 
+  public static final JsonPointer FORMAT_VERSION = JsonPointer.compile("/version");
+
   // -------------------------------------------------------------------------------------
   // top level spec definition
   // -------------------------------------------------------------------------------------