msgspec is the fastest Python serialization library. pip install msgspec. Define struct: import msgspec; class User(msgspec.Struct): name: str; age: int. JSON encode: msgspec.json.encode(user) → bytes. Decode: msgspec.json.decode(b'{"name":"Alice","age":30}', type=User). List: msgspec.json.decode(data, type=list[User]). Dict: msgspec.json.decode(data, type=dict[str, User]). Optional: class User(msgspec.Struct): nickname: str | None = None. Default: class User(msgspec.Struct): role: str = "user". Rename field: from msgspec import field; class User(msgspec.Struct): first_name: str = field(name="firstName"). Frozen: class Point(msgspec.Struct, frozen=True): x: float; y: float. GC off: class Msg(msgspec.Struct, gc=False): — faster for leaf objects. MessagePack: msgspec.msgpack.encode(user), msgspec.msgpack.decode(data, type=User). Decoder object: dec = msgspec.json.Decoder(User), dec.decode(data) — reuse for performance. Encoder: enc = msgspec.json.Encoder(), enc.encode(user). Validate only: msgspec.convert(raw_dict, User) — convert a plain dict to a Struct. Union: class Event(msgspec.Struct, tag=True): ... — tagged unions for discriminated decoding. tag_field="type". class UserEvent(Event, tag="user"). msgspec.json.decode(data, type=UserEvent|OrderEvent). Custom type: def enc_hook(obj): if isinstance(obj, Decimal): return str(obj). msgspec.json.encode(obj, enc_hook=enc_hook). def dec_hook(type, obj): if type is Decimal: return Decimal(obj). Struct fields: msgspec.structs.fields(User). msgspec.structs.asdict(user). msgspec.structs.astuple(user). msgspec.structs.replace(user, name="Bob") — creates new Struct. Json schema: msgspec.json.schema(User). msgspec.inspect.type_info(User). Benchmark: 2-10× faster than Pydantic v2, 10-20× faster than marshmallow for encode/decode of large payloads. Claude Code generates msgspec Struct definitions, JSON decoders, MessagePack pipelines, and tagged union event schemas.
CLAUDE.md for msgspec
## msgspec Stack
- Version: msgspec >= 0.18 | pip install msgspec
- Struct: class M(msgspec.Struct): field: type = default_val
- JSON: msgspec.json.encode(obj) → bytes | msgspec.json.decode(bytes, type=T)
- Msgpack: msgspec.msgpack.encode/decode — binary, faster than JSON for bulk
- Decoder: msgspec.json.Decoder(T) — reuse for high-throughput hot paths
- Convert: msgspec.convert(dict_or_obj, T) — coerce plain dicts to Struct
- Tagged union: class E(Struct, tag=True): ... — auto-discriminated by "type" fieldmsgspec Serialization Pipeline
# app/messages.py — msgspec Struct definitions and encoder/decoder setup
from __future__ import annotations
import time
from datetime import datetime
from decimal import Decimal
from typing import Annotated, Literal, Optional, Union
from uuid import UUID, uuid4
import msgspec
from msgspec import Struct, convert, field, json, msgpack
from msgspec.structs import asdict, astuple, fields, replace
# ─────────────────────────────────────────────────────────────────────────────
# Value types and enums
# ─────────────────────────────────────────────────────────────────────────────
# msgspec supports plain Python enums
from enum import Enum
class UserRole(str, Enum):
USER = "user"
MODERATOR = "moderator"
ADMIN = "admin"
class OrderStatus(str, Enum):
PENDING = "pending"
PAID = "paid"
SHIPPED = "shipped"
DELIVERED = "delivered"
CANCELLED = "cancelled"
# ─────────────────────────────────────────────────────────────────────────────
# Basic Structs
# ─────────────────────────────────────────────────────────────────────────────
class Address(Struct, frozen=True):
"""Frozen value object — hashable, can be used as dict key."""
street: str
city: str
state: str
postal_code: str
country: str = "US"
class User(Struct):
"""Mutable user Struct."""
id: str
email: str
first_name: str = field(name="firstName") # JSON key is "firstName"
last_name: str = field(name="lastName")
role: UserRole = UserRole.USER
is_active: bool = True
address: Optional[Address] = None
created_at: datetime = field(default_factory=lambda: datetime.utcnow())
class UserCreate(Struct):
"""Input schema for creating a user — password excluded from responses."""
email: str
first_name: str = field(name="firstName")
last_name: str = field(name="lastName")
password: str # only in input, not in User response
role: UserRole = UserRole.USER
class UserUpdate(Struct, omit_defaults=True):
"""Partial update — only present fields are serialized. Absent = unchanged."""
first_name: Optional[str] = field(None, name="firstName")
last_name: Optional[str] = field(None, name="lastName")
role: Optional[UserRole] = None
# ─────────────────────────────────────────────────────────────────────────────
# Order Structs
# ─────────────────────────────────────────────────────────────────────────────
class OrderLine(Struct, gc=False):
"""gc=False: no garbage collector overhead — safe because no circular refs."""
product_id: str
sku: str
quantity: int
unit_price: float
@property
def subtotal(self) -> float:
return self.quantity * self.unit_price
class Order(Struct):
id: str
user_id: str
lines: list[OrderLine]
status: OrderStatus = OrderStatus.PENDING
notes: Optional[str] = None
created_at: datetime = field(default_factory=lambda: datetime.utcnow())
@property
def total(self) -> float:
return sum(line.subtotal for line in self.lines)
class CreateOrderRequest(Struct):
user_id: str
lines: list[OrderLine]
notes: Optional[str] = None
# ─────────────────────────────────────────────────────────────────────────────
# Tagged unions — event system
# ─────────────────────────────────────────────────────────────────────────────
# When tag=True, msgspec adds "type" field to JSON and uses it for discrimination.
class BaseEvent(Struct, tag=True, tag_field="type"):
"""Base event — subclasses get automatic "type" field for union decoding."""
event_id: str = field(default_factory=lambda: str(uuid4()))
occurred_at: datetime = field(default_factory=lambda: datetime.utcnow())
class UserCreatedEvent(BaseEvent, tag="user_created"):
user_id: str
email: str
class OrderPlacedEvent(BaseEvent, tag="order_placed"):
order_id: str
user_id: str
total: float
class PaymentReceivedEvent(BaseEvent, tag="payment_received"):
order_id: str
amount: float
currency: str
# Union type — msgspec uses "type" field to pick the right Struct
DomainEvent = Union[UserCreatedEvent, OrderPlacedEvent, PaymentReceivedEvent]
# Decoder for the union — reuse for performance
_EVENT_DECODER = json.Decoder(DomainEvent)
def decode_event(data: bytes) -> DomainEvent:
return _EVENT_DECODER.decode(data)
def encode_event(event: DomainEvent) -> bytes:
return json.encode(event)
# ─────────────────────────────────────────────────────────────────────────────
# Custom enc_hook / dec_hook — extend supported types
# ─────────────────────────────────────────────────────────────────────────────
def enc_hook(obj):
"""Handle types not natively supported by msgspec."""
if isinstance(obj, Decimal):
return str(obj)
if isinstance(obj, UUID):
return str(obj)
raise TypeError(f"Unsupported type: {type(obj)}")
def dec_hook(expected_type, obj):
"""Coerce raw JSON values to expected types."""
if expected_type is Decimal:
return Decimal(obj)
if expected_type is UUID:
return UUID(obj)
raise TypeError(f"Unsupported type: {expected_type}")
# Encoders/decoders with custom hooks — create once, reuse
_ENC = json.Encoder(enc_hook=enc_hook)
_DEC_USER = json.Decoder(User, dec_hook=dec_hook)
_DEC_ORDER = json.Decoder(Order, dec_hook=dec_hook)
# ─────────────────────────────────────────────────────────────────────────────
# Conversion from plain dicts (e.g., database rows, parsed HTTP bodies)
# ─────────────────────────────────────────────────────────────────────────────
def user_from_dict(data: dict) -> User:
"""
Convert a plain dict (from database row, request body, etc.) to a User Struct.
msgspec.convert() performs type coercion (e.g., str→enum, str→datetime).
"""
return convert(data, User)
def order_from_dict(data: dict) -> Order:
return convert(data, Order)
# ─────────────────────────────────────────────────────────────────────────────
# MessagePack for binary protocols
# ─────────────────────────────────────────────────────────────────────────────
_MSGPACK_ENC = msgpack.Encoder(enc_hook=enc_hook)
_MSGPACK_DEC = msgpack.Decoder(DomainEvent, dec_hook=dec_hook)
def encode_event_binary(event: DomainEvent) -> bytes:
"""Encode to MessagePack — ~30% smaller than JSON for typical events."""
return _MSGPACK_ENC.encode(event)
def decode_event_binary(data: bytes) -> DomainEvent:
return _MSGPACK_DEC.decode(data)
# ─────────────────────────────────────────────────────────────────────────────
# JSON Schema generation — for OpenAPI docs
# ─────────────────────────────────────────────────────────────────────────────
def get_schema(type_: type) -> dict:
"""Generate JSON Schema from any msgspec type."""
return json.schema(type_)
# ─────────────────────────────────────────────────────────────────────────────
# Benchmark demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
user = User(
id=str(uuid4()),
email="[email protected]",
first_name="Alice",
last_name="Smith",
role=UserRole.ADMIN,
)
# Encode / decode
encoded = json.encode(user)
print(f"JSON bytes: {len(encoded)}")
print(f"Encoded: {encoded[:80].decode()}")
decoded = json.decode(encoded, type=User)
print(f"Decoded: {decoded.email} role={decoded.role}")
# Tagged union event
event = OrderPlacedEvent(order_id="ord-123", user_id=user.id, total=49.99)
event_bytes = encode_event(event)
print(f"\nEvent JSON: {event_bytes[:80].decode()}")
decoded_event = decode_event(event_bytes)
print(f"Event type: {type(decoded_event).__name__}")
# MessagePack
mp_bytes = encode_event_binary(event)
print(f"\nMsgPack bytes: {len(mp_bytes)} vs JSON: {len(event_bytes)}")
# replace() — functional update (like attrs.evolve)
updated = replace(user, role=UserRole.MODERATOR)
print(f"\nOriginal role: {user.role.value}")
print(f"Updated role: {updated.role.value}")
# Struct to dict
data = asdict(user)
print(f"\nasdict keys: {list(data.keys())}")For the Pydantic v2 alternative — Pydantic v2 added a Rust core and is now competitive for many workloads, but msgspec is still 2–5× faster for encode/decode of arrays of Structs because Struct fields are C-level slots with no Python overhead, msgspec.json.Decoder(list[User]) is a compiled decoder that parses the entire array without Python-level loops, and gc=False on leaf Structs skips garbage collector registration for hot-path message types. For the orjson alternative — orjson is a fast JSON encoder for plain Python objects (dicts, lists, dataclasses) but does no schema validation: orjson.loads(data) returns a plain dict that still needs manual type casting, while msgspec.json.decode(data, type=User) validates every field type and raises DecodeError with the failing field path if the payload does not match the Struct definition. The Claude Skills 360 bundle includes msgspec skill sets covering Struct field definitions and defaults, json.encode/decode and Decoder reuse, msgpack binary encoding, field(name=…) for JSON key aliasing, frozen and gc=False optimizations, tagged union discriminated decoding, enc_hook/dec_hook for custom types, convert for dict coercion, replace for functional updates, and json.schema for OpenAPI generation. Start with the free tier to try high-performance serialization code generation.