diff --git a/modules/core/src/main/scala/vulcan/AvroException.scala b/modules/core/src/main/scala/vulcan/AvroException.scala
index 2af07627..f966c78c 100644
--- a/modules/core/src/main/scala/vulcan/AvroException.scala
+++ b/modules/core/src/main/scala/vulcan/AvroException.scala
@@ -381,12 +381,14 @@ object AvroException {
     )
 
   /**
-    * Exception raised when adding a logical type to a
-    * schema with a type which has no logical types.
+    * Exception raised when adding a logical type to a schema
+    * that already has a logical type, or that has a type
+    * for which the logical type is invalid.
     */
   final class UnsupportedLogicalType private[vulcan] (logicalType: LogicalType, schema: Schema)
     extends AvroException(
-      s"Unable to add logical type ${logicalType.name} to ${SchemaNames.describe(schema)}"
+      s"Unable to add logical type ${logicalType.name} to ${SchemaNames.describe(schema)}" +
+        schema.props("logicalType").flatMap(_.asString).fold("")(name => s" with logical type $name")
     )
 
   /**
diff --git a/modules/core/src/main/scala/vulcan/Codec.scala b/modules/core/src/main/scala/vulcan/Codec.scala
index cba18254..fa305ab5 100644
--- a/modules/core/src/main/scala/vulcan/Codec.scala
+++ b/modules/core/src/main/scala/vulcan/Codec.scala
@@ -162,6 +162,17 @@ sealed abstract class Codec[A] {
   final def exmap[B](f: A => Either[AvroException, B])(g: B => Either[AvroException, A]): Codec[B] =
     new Codec.Exmap(this, f, g)
 
+  /**
+    * Returns a new [[Codec]] with the logical type added to its schema.
+    *
+    * The schema is an error if the logical type is invalid for the type
+    * of the schema, or if the schema already has a logical type.
+    *
+    * Decoding requires writer schemas to have the same logical type.
+    */
+  final def withLogicalType(logicalType: LogicalType): Codec[A] =
+    new Codec.LogicalTypeCodec(this, logicalType)
+
   /** Attempts to encode the specified value. */
   private[vulcan] def encode(a: A): Either[AvroException, AvroValue]
 
@@ -208,9 +219,6 @@ sealed abstract class Codec[A] {
   private[vulcan] final def withTypeName(typeName: String): Codec[A] =
     new Codec.WithTypeName(this, typeName)
 
-  private[vulcan] final def withLogicalType(logicalType: LogicalType): Codec[A] =
-    new Codec.LogicalTypeCodec(this, logicalType)
-
   /** Encodes the value as JSON for defaults in schemas, where unions are without the type. */
   private[vulcan] final def encodeForSchema(a: A): Either[AvroException, Json] =
     for {
@@ -1699,16 +1707,23 @@ object Codec {
     loop()
   }
 
-  private def addLogicalType(schema: Schema, logicalType: LogicalType): Either[AvroException, Schema] =
-    schema match {
-      case int: Schema.Int => Right(int.withLogicalType(logicalType))
-      case long: Schema.Long => Right(long.withLogicalType(logicalType))
-      case bytes: Schema.Bytes => Right(bytes.withLogicalType(logicalType))
-      case string: Schema.String => Right(string.withLogicalType(logicalType))
-      case fixed: Schema.Fixed => Right(fixed.withLogicalType(logicalType))
-      case _ =>
-        Left(new UnsupportedLogicalType(logicalType, schema))
-    }
+  /** The schema with the logical type, unless it has a logical type or the logical type is invalid. */
+  private def addLogicalType(schema: Schema, logicalType: LogicalType): Either[AvroException, Schema] = {
+    val added: Option[Schema] =
+      schema match {
+        case _ if schema.props.contains("logicalType") => None
+        case int: Schema.Int => Some(int.withLogicalType(logicalType))
+        case long: Schema.Long => Some(long.withLogicalType(logicalType))
+        case bytes: Schema.Bytes => Some(bytes.withLogicalType(logicalType))
+        case string: Schema.String => Some(string.withLogicalType(logicalType))
+        case fixed: Schema.Fixed => Some(fixed.withLogicalType(logicalType))
+        case _ => None
+      }
+
+    added
+      .filter(_.logicalType.contains(logicalType))
+      .toRight(new UnsupportedLogicalType(logicalType, schema))
+  }
 
   /** Decodes the decimal using the scale, and checks that it does not exceed the precision. */
   private def decodeDecimal(
diff --git a/tests/shared/src/test/scala/vulcan/CodecSpec.scala b/tests/shared/src/test/scala/vulcan/CodecSpec.scala
index a1d82537..5559e380 100644
--- a/tests/shared/src/test/scala/vulcan/CodecSpec.scala
+++ b/tests/shared/src/test/scala/vulcan/CodecSpec.scala
@@ -3271,4 +3271,79 @@ object CodecSpec extends BaseSuite {
       )
     }
   }
+
+  locally {
+    case class Nanos(value: Long)
+
+    object Nanos {
+      implicit val nanosCodec: Codec[Nanos] =
+        Codec.long.imap(Nanos(_))(_.value).withLogicalType(LogicalType.timestampNanos)
+    }
+
+    pureTest("Codec withLogicalType schema should add the logical type") {
+      expectSchema[Nanos] {
+        """{"type":"long","logicalType":"timestamp-nanos"}"""
+      }
+    }
+
+    pureTest("Codec withLogicalType decode should decode values written with the logical type") {
+      expectDecode[Nanos](
+        Codec[Long].encode(1L),
+        Right(Schema.Long.withLogicalType(LogicalType.timestampNanos)),
+        Nanos(1L)
+      )
+    }
+
+    pureTest("Codec withLogicalType decode should error if the writer schema has no logical type") {
+      expectDecodeError[Nanos](
+        Codec[Long].encode(1L),
+        Right(Schema.Long),
+        "Got unexpected missing logical type"
+      )
+    }
+  }
+
+  locally {
+    case class Id(value: ByteVector)
+
+    object Id {
+      implicit val idCodec: Codec[Id] =
+        Codec
+          .fixed[Id]("Id", "com.example", 16, _.value, bytes => Right(Id(bytes)))
+          .withLogicalType(LogicalType.uuid)
+    }
+
+    pureTest("Codec withLogicalType schema should add the logical type to fixed") {
+      expectSchema[Id] {
+        """{"type":"fixed","name":"Id","namespace":"com.example","size":16,"logicalType":"uuid"}"""
+      }
+    }
+
+    pureTest("Codec withLogicalType should roundtrip fixed with the logical type") {
+      expectRoundtrip(Id(ByteVector.fill(16L)(1)))
+    }
+  }
+
+  pureTest("Codec withLogicalType schema should error if the logical type is invalid for the type") {
+    implicit val codec: Codec[String] =
+      Codec.string.withLogicalType(LogicalType.timestampMillis)
+
+    expectSchemaError[String]("Unable to add logical type timestamp-millis to string")
+  }
+
+  pureTest("Codec withLogicalType schema should error if the schema already has a logical type") {
+    implicit val codec: Codec[Instant] =
+      Codec.instant.withLogicalType(LogicalType.timestampMicros)
+
+    expectSchemaError[Instant](
+      "Unable to add logical type timestamp-micros to long with logical type timestamp-millis"
+    )
+  }
+
+  pureTest("Codec withLogicalType schema should error if the type has no logical types") {
+    implicit val codec: Codec[Boolean] =
+      Codec.boolean.withLogicalType(LogicalType.uuid)
+
+    expectSchemaError[Boolean]("Unable to add logical type uuid to boolean")
+  }
 }