diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionType.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionType.java
index eedbd2688..0143a76e6 100644
--- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionType.java
+++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionType.java
@@ -17,13 +17,16 @@ package org.springframework.cloud.function.context;
 
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.util.Optional;
 import java.util.function.Function;
 
-import org.springframework.cloud.function.context.catalog.FunctionInspector;
+import org.reactivestreams.Publisher;
+
 import org.springframework.core.ResolvableType;
 import org.springframework.messaging.Message;
 
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 /**
  * @author Dave Syer
@@ -65,36 +68,112 @@ public class FunctionType {
 				|| Message.class.isAssignableFrom(outputType);
 	}
 
-	public static FunctionType compose(FunctionType input, FunctionType output) {
-		ResolvableType inputGeneric;
-		ResolvableType inputType = ResolvableType.forClass(input.getInputType());
-		if (input.isMessage()) {
-			inputType = ResolvableType.forClassWithGenerics(Message.class, inputType);
+	public boolean isWrapper() {
+		return isWrapper(getInputWrapper()) || isWrapper(getOutputWrapper());
+	}
+
+	public static boolean isWrapper(Type type) {
+		return Publisher.class.equals(type) || Flux.class.equals(type)
+				|| Mono.class.equals(type) || Optional.class.equals(type);
+	}
+
+	public static FunctionType from(Class<?> input) {
+		return new FunctionType(ResolvableType
+				.forClassWithGenerics(Function.class, input, Object.class).getType());
+	}
+
+	public FunctionType to(Class<?> output) {
+		ResolvableType inputGeneric = input(this);
+		ResolvableType outputGeneric = output(output);
+		return new FunctionType(ResolvableType
+				.forClassWithGenerics(Function.class, inputGeneric, outputGeneric)
+				.getType());
+	}
+
+	public FunctionType message() {
+		if (isMessage()) {
+			return this;
 		}
-		ResolvableType outputGeneric;
-		ResolvableType outputType = ResolvableType.forClass(output.getOutputType());
-		if (output.isMessage()) {
-			outputType = ResolvableType.forClassWithGenerics(Message.class, outputType);
-		}
-		if (FunctionInspector.isWrapper(input.getInputWrapper())) {
-			inputGeneric = ResolvableType.forClassWithGenerics(input.getInputWrapper(),
-					inputType);
-		}
-		else {
-			inputGeneric = inputType;
-		}
-		if (FunctionInspector.isWrapper(output.getInputWrapper())) {
-			outputGeneric = ResolvableType.forClassWithGenerics(output.getInputWrapper(),
-					outputType);
-		}
-		else {
-			outputGeneric = outputType;
+		ResolvableType inputGeneric = message(getInputType());
+		ResolvableType outputGeneric = message(getOutputType());
+		if (isWrapper(getInputWrapper())) {
+			inputGeneric = ResolvableType.forClassWithGenerics(getInputWrapper(),
+					inputGeneric);
+			outputGeneric = ResolvableType.forClassWithGenerics(getInputWrapper(),
+					outputGeneric);
 		}
 		return new FunctionType(ResolvableType
 				.forClassWithGenerics(Function.class, inputGeneric, outputGeneric)
 				.getType());
 	}
 
+	public FunctionType wrap(Class<?> wrapper) {
+		if (wrapper.isAssignableFrom(getInputWrapper())) {
+			return this;
+		}
+		return new FunctionType(ResolvableType.forClassWithGenerics(Function.class,
+				wrap(wrapper, getInputType()), wrap(wrapper, getOutputType())).getType());
+	}
+
+	public static FunctionType compose(FunctionType input, FunctionType output) {
+		ResolvableType inputGeneric = input(input);
+		ResolvableType outputGeneric = output(output);
+		return new FunctionType(ResolvableType
+				.forClassWithGenerics(Function.class, inputGeneric, outputGeneric)
+				.getType());
+	}
+
+	private ResolvableType wrap(Class<?> wrapper, Class<?> type) {
+		return isMessage() ? wrap(wrapper, message(type))
+				: ResolvableType.forClassWithGenerics(wrapper, type);
+	}
+
+	private ResolvableType wrap(Class<?> wrapper, ResolvableType type) {
+		return ResolvableType.forClassWithGenerics(wrapper, type);
+	}
+
+	private ResolvableType message(Class<?> type) {
+		return ResolvableType.forClassWithGenerics(Message.class, type);
+	}
+
+	private static ResolvableType input(FunctionType type) {
+		return type.input(type.getInputType());
+	}
+
+	private static ResolvableType output(FunctionType type) {
+		return type.output(type.getOutputType());
+	}
+
+	private ResolvableType output(Class<?> type) {
+		ResolvableType generic;
+		ResolvableType raw = ResolvableType.forClass(type);
+		if (isMessage()) {
+			raw = ResolvableType.forClassWithGenerics(Message.class, raw);
+		}
+		if (FunctionType.isWrapper(getOutputWrapper())) {
+			generic = ResolvableType.forClassWithGenerics(getOutputWrapper(), raw);
+		}
+		else {
+			generic = raw;
+		}
+		return generic;
+	}
+
+	private ResolvableType input(Class<?> type) {
+		ResolvableType generic;
+		ResolvableType raw = ResolvableType.forClass(type);
+		if (isMessage()) {
+			raw = ResolvableType.forClassWithGenerics(Message.class, raw);
+		}
+		if (FunctionType.isWrapper(getInputWrapper())) {
+			generic = ResolvableType.forClassWithGenerics(getInputWrapper(), raw);
+		}
+		else {
+			generic = raw;
+		}
+		return generic;
+	}
+
 	private Class<?> findType(ParamType paramType) {
 		int index = paramType.isOutput() ? 1 : 0;
 		Type type = this.type;
@@ -146,7 +225,7 @@ public class FunctionType {
 			Type typeArgumentAtIndex = parameterizedType.getActualTypeArguments()[index];
 			if (typeArgumentAtIndex instanceof ParameterizedType
 					&& !paramType.isWrapper()) {
-				if (FunctionInspector.isWrapper(
+				if (FunctionType.isWrapper(
 						((ParameterizedType) typeArgumentAtIndex).getRawType())) {
 					param = ((ParameterizedType) typeArgumentAtIndex)
 							.getActualTypeArguments()[0];
diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionInspector.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionInspector.java
index 8107a3128..34f3fc298 100644
--- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionInspector.java
+++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionInspector.java
@@ -16,14 +16,6 @@
 
 package org.springframework.cloud.function.context.catalog;
 
-import java.lang.reflect.Type;
-import java.util.Optional;
-
-import org.reactivestreams.Publisher;
-
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
 /**
  * @author Dave Syer
  *
@@ -44,10 +36,4 @@ public interface FunctionInspector {
 
 	String getName(Object function);
 
-	// Maybe make this a default method?
-	static boolean isWrapper(Type type) {
-		return Publisher.class.equals(type) || Flux.class.equals(type) || Mono.class.equals(type)
-				|| Optional.class.equals(type);
-	}
-
 }
diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java
index 4b7db0856..3f1400b4b 100644
--- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java
+++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java
@@ -541,8 +541,7 @@ public class ContextFunctionCatalogAutoConfiguration {
 		}
 
 		private boolean hasFluxTypes(Object function) {
-			return FunctionInspector.isWrapper(findType(function).getInputWrapper())
-					|| FunctionInspector.isWrapper(findType(function).getOutputWrapper());
+			return findType(function).isWrapper();
 		}
 
 		private FunctionType findType(String name, AbstractBeanDefinition definition) {
diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/FunctionTypeTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/FunctionTypeTests.java
index b91436694..9c9b37d68 100644
--- a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/FunctionTypeTests.java
+++ b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/FunctionTypeTests.java
@@ -99,6 +99,16 @@ public class FunctionTypeTests {
 		assertThat(function.isMessage()).isEqualTo(false);
 	}
 
+	@Test
+	public void plainFunctionFromApi() {
+		FunctionType function = FunctionType.from(Integer.class).to(String.class);
+		assertThat(function.getInputType()).isEqualTo(Integer.class);
+		assertThat(function.getOutputType()).isEqualTo(String.class);
+		assertThat(function.getInputWrapper()).isEqualTo(Integer.class);
+		assertThat(function.getOutputWrapper()).isEqualTo(String.class);
+		assertThat(function.isMessage()).isEqualTo(false);
+	}
+
 	@Test
 	public void fluxMessageFunctionFromType() {
 		Type type = ResolvableType
@@ -118,6 +128,31 @@ public class FunctionTypeTests {
 		assertThat(function.isMessage()).isEqualTo(true);
 	}
 
+	@Test
+	public void fluxMessageFunctionFromApi() {
+		FunctionType function = FunctionType.from(Foo.class).to(Bar.class).message()
+				.wrap(Flux.class);
+		assertThat(function.getInputType()).isEqualTo(Foo.class);
+		assertThat(function.getOutputType()).isEqualTo(Bar.class);
+		assertThat(function.getInputWrapper()).isEqualTo(Flux.class);
+		assertThat(function.getOutputWrapper()).isEqualTo(Flux.class);
+		assertThat(function.isMessage()).isEqualTo(true);
+	}
+
+	@Test
+	public void idempotentMessage() {
+		FunctionType function = FunctionType.from(Foo.class).to(Bar.class).message()
+				.wrap(Flux.class);
+		assertThat(function).isSameAs(function.message());
+	}
+
+	@Test
+	public void idempotentWrapper() {
+		FunctionType function = FunctionType.from(Foo.class).to(Bar.class).message()
+				.wrap(Flux.class);
+		assertThat(function).isSameAs(function.wrap(Flux.class));
+	}
+
 	private static class IntegerToString implements Function<Integer, String> {
 		@Override
 		public String apply(Integer t) {