Backbone

Backbone is the only thing you implement. It has two required methods — capabilities() and generate() — plus two optional lifecycle hooks.

from motionmcp import Backbone, ModelSpec, GenerateRequest, MotionResult

class MyBackbone(Backbone):
    def setup(self) -> None: ...                     # optional
    def teardown(self) -> None: ...                  # optional

    def capabilities(self) -> ModelSpec: ...
    async def generate(self, req: GenerateRequest) -> MotionResult: ...

capabilities()

Returns a ModelSpec describing the model. The SDK serialises it into the per-model entry under /capabilities.models[].

ModelSpec is a Pydantic model with sensible defaults — only id, fps, and canonical_skeleton are required.

from motionmcp import ModelSpec, Skeleton, Joint

ModelSpec(
    id="my-model-v1",
    fps=30.0,
    canonical_skeleton=Skeleton(joints=[
        Joint(name="Hips", parent=None,
              rest_translation=(0.0, 0.95, 0.0),
              rest_rotation=(0.0, 0.0, 0.0, 1.0)),
    ]),

    # All fields below are optional with the defaults shown:
    supports_retargeting=False,
    supports_async=False,
    supported_constraints=[],          # empty = the model accepts no constraints
    supported_guidance_types=["nocfg", "regular", "separated"],
    predicted_contact_joints=[],
    native_clip_seconds=10.0,
    chunking="none",                   # or "stitched"
    recommended_max_duration_seconds=12.0,
    # limits=Limits(max_duration_seconds=30.0, max_num_samples=16, …)
)

See Capabilities reference → for what every field controls on the wire.

Capabilities is called once per request

The SDK calls capabilities() on every /capabilities and /generate request to look up the spec. Keep it cheap — return a cached object if your spec depends on slow lookups.

generate(request)

Runs the model on a validated GenerateRequest and returns a MotionResult. May be async def or sync — the SDK detects the form and awaits when needed.

async def generate(self, req: GenerateRequest) -> MotionResult:
    # req.skeleton, req.segments, req.constraints — parsed Pydantic models
    # req.total_frames                              — int helper
    # req.fps(spec.fps)                             — effective fps
    # req.options.num_samples                       — None-safe access via .options

    rotations    = run_my_model(req)
    translations = run_my_root(req)

    return MotionResult(
        rotations=rotations,
        root_translations=translations,
    )

The shape of MotionResult is documented in Results →. The SDK encodes it to glTF for you.

Sync vs async

Both work. Async is preferred when generation does I/O (queueing on a GPU worker, calling out to another service). Sync is fine for a self-contained model:

class FastBackbone(Backbone):
    def generate(self, req: GenerateRequest) -> MotionResult:
        ...

The SDK uses inspect.iscoroutinefunction to pick the right path.

Raising errors

Raise motionmcp.ProtocolError (or one of the typed convenience constructors) for protocol-level failures. The SDK converts them to the standard error envelope with the right HTTP status:

from motionmcp.errors import unknown_joint, ProtocolError

if user_joint not in self.skeleton_names:
    raise unknown_joint(user_joint, sorted(self.skeleton_names))

if not self.weights_loaded:
    raise ProtocolError("model_unavailable", "weights still loading")

Anything that isn't a ProtocolError becomes 500 internal_error.

See Errors → for the full list.

Motion references (1.2)

A motion_reference segment is a prompt given as a motion. To take one, list it and cap the clip's length:

spec = ModelSpec(..., supported_segments=["text", "unconditioned", "motion_reference"])
spec.limits.max_reference_frames = 300

The SDK validates a reference with the text segment's rules (it has a duration_frames and a seed, mixes with other segments, goes with loop, constraints and guidance as text does) plus its payload: joints on its own skeleton (or the request skeleton when it has none), the root joint among them, finite numbers, the clip no longer than max_reference_frames frames as sent. It reaches you as parsed, not resampled or retargeted. A clip within max_reference_frames MUST be accepted at any fps: if resampling it to your model's rate gives more frames than your window, resample it to fit (or trim it), don't refuse it. Treat it as a text segment whose caption embedding comes from the clip instead of the text encoder, and run everything else — lengths, seeds, guidance defaults, transitions, loop, batching — through your text path unchanged:

for seg in req.segments:
    if seg.type == "motion_reference":
        # seg.rotations (T_ref × J × 4), seg.root_positions (T_ref × 3),
        # seg.joint_names, seg.fps; seg.skeleton (Skeleton | None): the
        # clip's own rig, to retarget from when set
        caption = embed_clip(seg)        # clip -> the text encoder's kind of vector
    elif seg.type == "text":
        caption = encode_text(seg.prompt)
    ...                                   # the same path for both

/capabilities advertises supports_motion_reference_mixed: true for such a model and false for any other (a reference always mixes like text). If your model can't take a non-identity rest_rotation (on the request skeleton or a segment's), raise 400 invalid_skeleton for it rather than ignoring it. What clients send — rotations from rest with identity rest_rotation, the root's world position, any rig's own skeleton — is in Building reference segments →.

Video references (1.2)

A video_reference segment is a prompt given as a video of a person. To take one, list it and cap the video:

spec = ModelSpec(..., supported_segments=["text", "unconditioned", "video_reference"])
spec.limits.max_video_bytes = 50_000_000     # decoded size of the video file
spec.limits.max_video_seconds = 20.0         # seconds of video to read

The SDK validates it with the text segment's rules plus its payload: one source (an https video.url, or base64 video.data with a media_type), 0 ≤ start_s < end_s, the inline data within max_video_bytes (413 payload_too_large) and a trimmed video within max_video_seconds (400 invalid_options). It never decodes the video; the rest is yours, once you have it: fetch the URL, check an untrimmed video's length, find the most prominent person (the largest, most visible one) to follow, and raise invalid_request when you can't.

Fetching a URL safely. A video.url comes from the client, so fetching it makes your server an SSRF target (its loopback, its cloud metadata endpoint, its private network). A backbone that fetches one MUST:

• fetch https only, with the URL rule the schema applies (a host, no user:pw@, no whitespace or control characters, a valid port);
• resolve the host and refuse it if any address is loopback, private, link-local (which covers 169.254.169.254), carrier-grade NAT, multicast, reserved or unspecified, then connect to the address it checked (so a second DNS answer can't swap it);
• follow redirects by hand, a few at most, checking every hop the same way;
• stream the body and stop past limits.max_video_bytes (413 payload_too_large), and bound the whole fetch with a timeout.

motionmcp.fetch_video_url(url, max_bytes=None, timeout=30.0) does all of this with the standard library, raising ProtocolErrors you can let propagate (invalid_request, payload_too_large, timeout). IPv6 addresses that reach an IPv4 one (mapped, NAT64, 6to4, Teredo) are judged by that address, and timeout is one hard deadline over connect, TLS, headers and body. max_bytes=None (your model advertises no max_video_bytes) means motionmcp.fetch.DEFAULT_MAX_VIDEO_BYTES, 100 MiB. It is synchronous: from an async def generate, run it with asyncio.to_thread so it doesn't block the event loop.

import asyncio

from motionmcp import ProtocolError, fetch_video_url

async def generate(self, req):
    cap = self.capabilities().limits.max_video_bytes      # None -> the 100 MiB default
    for seg in req.segments:
        if seg.type == "video_reference":
            raw = (seg.video.decoded() if seg.video.data else
                   await asyncio.to_thread(fetch_video_url, seg.video.url, max_bytes=cap))
            # seg.video.media_type (inline), seg.start_s / seg.end_s (seconds
            # into the video, None = its start / end), seg.fps (a hint, or None)
            track = track_most_prominent_person(raw, seg.start_s, seg.end_s)
            if track is None:
                raise ProtocolError("invalid_request", "no person to follow in the video",
                                    details={"segment": req.segments.index(seg)})
            caption = embed_track(track)      # video -> the text encoder's kind of vector
        ...                                   # then the text path, as for a motion reference

req.video_references lists them in order (req.video_reference is the first). seg.video.num_bytes is the inline file's size, worked out from the base64's length (the SDK checks the alphabet but doesn't decode it); seg.video.decoded() decodes it. Keep limits.max_request_bytes above max_video_bytes × 4/3 if you take inline video: the body carries it as base64, and the SDK server refuses a body over max_request_bytes before parsing it.

generate_batch(requests)

Optional. The SDK serves batch bodies for every backbone (and advertises supports_batch); without this method it runs generate() on each item in turn. Override it when the model can do better — one GPU pass over several characters:

class MyBackbone(Backbone):
    def generate_batch(self, requests):
        # one entry per request, in order: a MotionResult, or the exception
        # that item raised (a ProtocolError gives a clean error envelope)
        return [self.generate(r) for r in requests]

Items arrive already validated, grouped by model. Sync or async, like generate().

setup() and teardown()

Optional hooks called once at server startup and shutdown. Use them for model loading, GPU allocation, warm caches.

class MyBackbone(Backbone):
    def setup(self) -> None:
        self.model = torch.load("weights.pt").to("cuda").eval()

    def teardown(self) -> None:
        del self.model
        torch.cuda.empty_cache()

Default implementations are no-ops; override only when needed.

Putting it together

A real backbone with a loaded model:

import numpy as np
import torch

from motionmcp import (
    Backbone, ModelSpec, GenerateRequest, MotionResult,
    Skeleton, serve,
)
from motionmcp.errors import unknown_joint

class SomaBackbone(Backbone):
    def __init__(self, weights_path: str) -> None:
        self.weights_path = weights_path
        self.model = None

    def setup(self) -> None:
        self.model = torch.load(self.weights_path).to("cuda").eval()
        self._spec = ModelSpec(
            id="soma-v1",
            fps=30.0,
            canonical_skeleton=Skeleton.model_validate(
                self.model.canonical_skeleton_dict
            ),
            supports_retargeting=True,
            supported_constraints=["root_path", "pose_keyframe"],
        )

    def capabilities(self) -> ModelSpec:
        return self._spec

    async def generate(self, req: GenerateRequest) -> MotionResult:
        with torch.no_grad():
            out = self.model.run(req)            # your I/O here
        return MotionResult(
            rotations=out.rotations.cpu().numpy(),
            root_translations=out.root_translations.cpu().numpy(),
        )

if __name__ == "__main__":
    serve(SomaBackbone("weights.pt"))