Auto-activate for msgspec imports, msgspec.Struct definitions, msgspec.json/msgpack usage. High-performance Python serialization and validation library: Struct definitions, Meta constraints, tagged unions, enc_hook/dec_hook, convert(). Produces msgspec Structs, serialization codecs, validation schemas, and type-safe conversion patterns. Use when: defining data models with msgspec, serializing/deserializing JSON or MessagePack, validating data with Meta constraints, building discriminated unions, or converting dicts/objects with msgspec.convert(). Not for Pydantic models, dataclasses, or attrs -- msgspec has its own patterns.
From flownpx claudepluginhub cofin/flow --plugin flowThis skill uses the workspace's default tool permissions.
references/constraints.mdreferences/tagged-unions.mdSearches, retrieves, and installs Agent Skills from prompts.chat registry using MCP tools like search_skills and get_skill. Activates for finding skills, browsing catalogs, or extending Claude.
Searches prompts.chat for AI prompt templates by keyword or category, retrieves by ID with variable handling, and improves prompts via AI. Use for discovering or enhancing prompts.
Guides agentic engineering workflows: eval-first loops, 15-min task decomposition, model routing (Haiku/Sonnet/Opus), AI code reviews, and cost tracking.
msgspec is a high-performance Python library for serialization, deserialization, and validation. Structs are ~5x more memory-efficient than regular classes and serialize faster than Pydantic or dataclasses.
T | None (not Optional[T])from __future__ import annotations -- breaks msgspec runtime type resolutionkw_only=True for Structs with more than 2 fieldsimport msgspec
# Basic struct
class User(msgspec.Struct):
id: int
name: str
email: str | None = None
# Performance options
class Event(msgspec.Struct, frozen=True, gc=False):
"""frozen=True: immutable + hashable. gc=False: skip GC for short-lived objects."""
event_type: str
payload: dict[str, object]
# Keyword-only (recommended for >2 fields)
class Config(msgspec.Struct, kw_only=True):
host: str
port: int = 5432
ssl: bool = False
# Array-like encoding (tuple encoding, more compact)
class Point(msgspec.Struct, array_like=True):
x: float
y: float
# Rename fields for serialization
class ApiResponse(msgspec.Struct, rename="camel"):
user_id: int # serialized as "userId"
created_at: str # serialized as "createdAt"
# Reject unknown fields at API boundaries
class StrictInput(msgspec.Struct, forbid_unknown_fields=True):
name: str
value: int
from typing import Annotated
import msgspec
from msgspec import Meta
class Product(msgspec.Struct):
name: Annotated[str, Meta(min_length=1, max_length=100)]
price: Annotated[float, Meta(gt=0)]
quantity: Annotated[int, Meta(ge=0, le=10_000)]
sku: Annotated[str, Meta(pattern=r"^[A-Z]{2}-\d{4}$")]
weight_kg: Annotated[float, Meta(multiple_of=0.001)]
# Reusable constraint aliases
PositiveInt = Annotated[int, Meta(gt=0)]
NonEmptyStr = Annotated[str, Meta(min_length=1)]
Percentage = Annotated[float, Meta(ge=0.0, le=100.0)]
class Order(msgspec.Struct):
id: PositiveInt
label: NonEmptyStr
discount: Percentage = 0.0
import msgspec
# JSON -- singleton encoder/decoder (cache these!)
encoder = msgspec.json.Encoder()
decoder = msgspec.json.Decoder(User)
data = encoder.encode(user) # bytes
user = decoder.decode(b'{"id":1,"name":"Alice"}')
# Functional API (convenience, slightly slower)
data = msgspec.json.encode(user)
user = msgspec.json.decode(b'...', type=User)
# MessagePack (binary, more compact)
data = msgspec.msgpack.encode(user)
user = msgspec.msgpack.decode(data, type=User)
# Custom hooks for non-native types (datetime, UUID, Decimal)
from datetime import datetime
import uuid
def enc_hook(obj: object) -> object:
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, uuid.UUID):
return str(obj)
raise TypeError(f"Unsupported type: {type(obj)}")
def dec_hook(type: type, obj: object) -> object:
if type is datetime:
return datetime.fromisoformat(obj)
if type is uuid.UUID:
return uuid.UUID(obj)
raise TypeError(f"Unsupported type: {type}")
encoder = msgspec.json.Encoder(enc_hook=enc_hook)
decoder = msgspec.json.Decoder(MyStruct, dec_hook=dec_hook)
import msgspec
raw = {"id": "42", "name": "Alice"} # id is a string
# Strict mode (default): raises on type mismatch
user = msgspec.convert(raw, User) # ValidationError: id must be int
# Lax mode: coerces compatible types
user = msgspec.convert(raw, User, strict=False) # id coerced to 42
# str_keys: dict keys are strings (useful for JSON-loaded dicts)
data = {"1": "Alice", "2": "Bob"}
result = msgspec.convert(data, dict[int, str], str_keys=True)
# Convert with dec_hook for custom types
user = msgspec.convert(raw, UserWithUUID, dec_hook=dec_hook)
# Convert dataclass/dict/object to Struct
from dataclasses import dataclass
@dataclass
class LegacyUser:
id: int
name: str
legacy = LegacyUser(id=1, name="Alice")
user = msgspec.convert(msgspec.structs.asdict(legacy), User)
# Or directly:
user = msgspec.convert(legacy, User)
import msgspec
# Runtime struct from field definitions
fields = [
("id", int),
("name", str),
("score", Annotated[float, Meta(ge=0.0)]),
]
DynamicModel = msgspec.defstruct("DynamicModel", fields, kw_only=True)
# With defaults
fields_with_defaults = [
("id", int),
("active", bool, True), # (name, type, default)
]
FlexModel = msgspec.defstruct("FlexModel", fields_with_defaults)
import msgspec
from typing import Literal
# Default tag field is "type", tag value is the class name
class Dog(msgspec.Struct, tag=True):
name: str
breed: str
class Cat(msgspec.Struct, tag=True):
name: str
indoor: bool
Animal = Dog | Cat
# Deserialize: inspects "type" field to pick correct class
animal = msgspec.json.decode(b'{"type":"Dog","name":"Rex","breed":"Lab"}', type=Animal)
# Custom tag values
class CreateEvent(msgspec.Struct, tag="create"):
resource: str
class DeleteEvent(msgspec.Struct, tag="delete"):
resource: str
soft: bool = True
Event = CreateEvent | DeleteEvent
# Custom tag field name
class V1Request(msgspec.Struct, tag="v1", tag_field="version"):
payload: str
class V2Request(msgspec.Struct, tag="v2", tag_field="version"):
payload: str
metadata: dict[str, str] = {}
Request = V1Request | V2Request
<workflow>
Create msgspec Structs for all data shapes. Use kw_only=True for Structs with more than 2 fields. Use frozen=True for immutable value objects. Use forbid_unknown_fields=True for API-boundary input validation.
Annotate fields with Annotated[Type, Meta(...)] for numeric ranges, string lengths, and regex patterns. Define reusable constraint aliases at module level to avoid repetition.
Use msgspec.json for JSON APIs and msgspec.msgpack for binary protocols or internal messaging. Instantiate Encoder/Decoder once at module level as singletons. Add enc_hook/dec_hook for custom types (datetime, UUID, Decimal, Enum).
Use tagged unions (tag=True or tag="value") for discriminated unions. Define a union type alias (Event = CreateEvent | DeleteEvent) and decode against it. Use tag_field to customize the discriminator field name.
Test round-trip encode/decode. Confirm ValidationError is raised for constraint violations. Verify tag dispatch selects the correct Struct type for all union variants.
kw_only=True for Structs with >2 fields -- prevents positional argument confusion and makes instantiation self-documenting.forbid_unknown_fields=True at API boundaries -- rejects payloads with unexpected keys, preventing silent data loss.Meta constraints over manual validation -- zero runtime overhead; constraints are checked during decode, not after.gc=False for short-lived, non-circular objects -- eliminates GC overhead for hot-path objects like request/response shapes.isinstance chains.from __future__ import annotations -- deferred annotation evaluation breaks msgspec's runtime type introspection.strict=False only at trust boundaries -- lax coercion is useful for converting legacy dicts but can mask type errors in internal code.Before delivering msgspec code, verify:
from __future__ import annotations anywhere in the moduleforbid_unknown_fields=TrueMeta (not manual if checks)enc_hook/dec_hook handle all non-native types used in Structskw_only=True on Structs with more than 2 fieldsTask: Define an event system with tagged unions, constraints, and JSON serialization.
from __future__ import annotations # DO NOT DO THIS -- breaks msgspec
# events.py
from typing import Annotated, Literal
from datetime import datetime
import uuid
import msgspec
from msgspec import Meta
# --- Constraint aliases ---
NonEmptyStr = Annotated[str, Meta(min_length=1, max_length=255)]
PositiveInt = Annotated[int, Meta(gt=0)]
# --- Event variants (tagged union) ---
class UserCreatedEvent(msgspec.Struct, tag="user.created", tag_field="event_type", kw_only=True, gc=False):
event_id: uuid.UUID
user_id: PositiveInt
email: NonEmptyStr
occurred_at: datetime
class UserDeletedEvent(msgspec.Struct, tag="user.deleted", tag_field="event_type", kw_only=True, gc=False):
event_id: uuid.UUID
user_id: PositiveInt
occurred_at: datetime
reason: str | None = None
UserEvent = UserCreatedEvent | UserDeletedEvent
# --- Custom hooks for datetime and UUID ---
def enc_hook(obj: object) -> object:
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, uuid.UUID):
return str(obj)
raise TypeError(f"Unsupported type: {type(obj)}")
def dec_hook(type: type, obj: object) -> object:
if type is datetime:
return datetime.fromisoformat(obj)
if type is uuid.UUID:
return uuid.UUID(obj)
raise TypeError(f"Unsupported type: {type}")
# --- Singleton codec ---
_encoder = msgspec.json.Encoder(enc_hook=enc_hook)
_decoder = msgspec.json.Decoder(UserEvent, dec_hook=dec_hook)
def encode_event(event: UserEvent) -> bytes:
return _encoder.encode(event)
def decode_event(data: bytes) -> UserEvent:
return _decoder.decode(data)
# --- Usage ---
event = UserCreatedEvent(
event_id=uuid.uuid4(),
user_id=42,
email="alice@example.com",
occurred_at=datetime.utcnow(),
)
payload = encode_event(event)
# b'{"event_type":"user.created","event_id":"...","user_id":42,"email":"alice@example.com","occurred_at":"..."}'
recovered = decode_event(payload)
assert isinstance(recovered, UserCreatedEvent)
</example>
For detailed guides and reference tables, refer to the following documents in references/: