diff --git a/CHANGELOG.md b/CHANGELOG.md
index c0acb6b..741b3fb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,8 @@
 
 ### Added
 
+- **Input preparation: `upload_file` and `prepare_inputs` (hosted upload capability).** The Python counterpart of `@pipelex/sdk`'s `uploadFile` / `prepareInputs`, in parity. `client.upload_file(source, *, filename=None, content_type=None)` uploads one local asset — a filesystem path (`str`/`Path`) or raw `bytes` — and returns an `UploadRecord` (`uri`, `content_type`, `size`, `filename`) assembled client-side. `client.prepare_inputs(*, files, pipe_ref=None, inputs)` resolves the target pipe's declared signature from the explicit inputs template, interprets the caller's compact `inputs` top-down against it (the file signal is the canonical Image/Document content shape — a `{"url": …}` dict — mirroring the runtime's `input_normalizer`), uploads the file-bearing values, and returns `PreparedInputs`: a copy-on-write rewrite of `inputs` with each asset reference replaced by canonical content carrying `pipelex-storage://` in `url`, plus one `UploadRecord` per prepared asset. HTTP(S) URLs and existing `pipelex-storage://` URIs pass through unchanged; data URLs and local/byte sources are uploaded; the same source referenced twice uploads once (dedup by source identity); all failures are raised before any run is created. Failures are typed per category: `InvalidLocalSourceError`, `RejectedAssetError`, `UnsupportedUploadCapabilityError`, `UploadAuthenticationError`, `UploadTransportError` (all extend `InputPreparationError`). `prepare_inputs` takes the method closure as inline `files`; catalog `method_id` resolution and opt-in `http(s)` ingest are deferred and additive. See [`docs/input-preparation.md`](./docs/input-preparation.md).
+- **`build_inputs` route (`POST /v1/build/inputs`).** Closes the `/v1/build/*` parity gap the Python SDK had — `client.build_inputs(BuildInputsRequest)` projects a pipe's declared inputs as a fill-in template, returning a 200 verdict discriminated on `is_valid` (`BuildInputsValidReport` | `CrateInvalidReport`); a no-verdict condition throws `ApiResponseError`. It is the signature source `prepare_inputs` reads (with `explicit=True`). Models live in `pipelex_sdk/build_models.py`.
 - **Typed run usage: `RunResults.tokens_usages` + `RunResults.usage_assembly_error`.** The per-call usage records a run produces — token counts by category, the server-computed `cost` in USD, model name and id, the pipe that made the call, job-kind fields and timing, for LLM and img-gen/extract/search calls alike — are now first-class typed fields instead of riding `model_extra`. Records validate into a new `TokensUsageRecord` model (`pipelex_sdk/runs.py`) mirroring the wire contract specified in the MTHDS protocol spec. Both paths populate the pair: the hosted durable path reads it off `GET /v1/runs/{id}/results` (which unpacks the runner's `tokens_usages.json` artifact), and the blocking fallback lifts the same pair out of the execute response's extension-open `pipe_output` — so `result.tokens_usages` reads the same regardless of which path ran.
 
   Note that the rate table (`unit_costs`) no longer crosses the wire: a record now carries the computed `cost` for the call instead, which is `None` when the model has no rate table at all (own-GPU, mock, dry run) and `0` when a rate table priced it at zero. There is no run-level aggregate — sum the records.
diff --git a/docs/architecture.md b/docs/architecture.md
index 22b8525..5f8b652 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -138,7 +138,7 @@ The wire models are snake_case Pydantic v2. Response models are extension-open (
 - **Pipelex API keys** — `list_pipelex_api_keys()`; `create_pipelex_api_key(label)` and `rotate_pipelex_api_key(id)` return the plaintext `api_key` **once**; `revoke_pipelex_api_key(id)`. Creation surfaces a **409 `pipelex_api_key_limit_reached`** when the per-account limit is hit. Rotation sends no body.
 - **Gateway (LLM inference) key** — `create_gateway_api_key(promo_code)` **always sends a JSON body** (even with `promo_code=None` → `{"promo_code": null}`); the server 422s an empty body. `get_gateway_api_key()` → status (`gateway_api_key` is `None` until provisioned).
 - **Onboarding** — `submit_onboarding(OnboardingSubmission)` (`POST /v1/onboarding/submit`, empty 2xx body); absent optional fields are dropped.
-- **Storage** — `resolve_storage_url(uri)` → presigned URL; `upload(UploadInput)` → the stored file handle.
+- **Storage** — `resolve_storage_url(uri)` → presigned URL; `upload(UploadInput)` → the stored file handle. The higher-level `upload_file` / `prepare_inputs` preparation surface built on top of `upload` is now available — see [input-preparation.md](./input-preparation.md).
 - **Run records** — `list_runs(method_id)` → `list[PipelineRun]` (the catalog-style list, distinct from the lifecycle status/result routes); `update_run(run_id, UpdateRunInput)` (admin/manual status patch, empty 2xx body).
 
 ## Health probe
diff --git a/docs/input-preparation.md b/docs/input-preparation.md
new file mode 100644
index 0000000..6d65eed
--- /dev/null
+++ b/docs/input-preparation.md
@@ -0,0 +1,101 @@
+# Input preparation (`upload_file` / `prepare_inputs`)
+
+> **Status: implemented** (`pipelex_sdk/upload.py`, `pipelex_sdk/prepare_inputs.py`, `pipelex_sdk/build_models.py`). This document records the contract (design source: `wip/upload/README.md` in the workspace, tracked in `TODOS.md`). `upload_file` and `prepare_inputs` are the Python counterpart of `@pipelex/sdk`'s `uploadFile` / `prepareInputs`, built on the raw `upload()` wire call. This work also added the `build_inputs` route (the signature source), which the Python SDK previously lacked.
+>
+> **Current scope.** `prepare_inputs` takes the method closure as inline `files` (the signature source). Two pieces are deliberately deferred and additive (they do not change this contract): resolving a closure from a catalog `method_id`, and the opt-in ingest of `http(s)` URLs into storage — for now an `http(s)` URL at a file position always passes through unchanged. Kept in parity with `@pipelex/sdk`.
+
+## Why this exists
+
+A hosted run cannot see the caller's filesystem. Turning caller-local assets into run-ready inputs is therefore the SDK's job, not the runner's — the SDK process is the only component that can read the local file or hold the bytes. Today that work is re-implemented by every consumer (read file → base64 → `POST /v1/upload` → rewrite the input to the returned URI); `fenix-pipelex` is an early real-world example. `prepare_inputs` makes it one reusable, explicit operation.
+
+Preparation is **explicit and separate from running.** `execute` / `start` never silently upload local files. The payoff: file-access errors happen *before* a run exists, prepared inputs are inspectable and reusable across a model sweep or retries without re-uploading, and `start` keeps a deterministic JSON-input contract.
+
+## Parity note
+
+This is the Python side of one cross-language contract. The behavior matrix, pass-through rules, `Dynamic` handling, dedup, and failure categories are identical to the JS SDK; only the accepted source types differ per language. The two SDKs must agree semantically. See the JS counterpart's `docs/input-preparation.md` for the mirror.
+
+The Python SDK previously had **no `/v1/build/*` coverage** — this change added the `build_inputs` counterpart `prepare_inputs` needs to resolve the declared signature (the JS SDK already exposes `buildInputs`).
+
+## The two operations
+
+### `upload_file` — single-asset convenience
+
+Uploads one asset and returns its upload record. It is the language-native convenience over the raw `upload()` wire call (base64 JSON body), assembling the record client-side.
+
+- **Accepted sources:** `str` and `pathlib.Path` filesystem paths, and raw `bytes`. `upload_file` treats every string as a **filesystem path** — a URL handed directly to it is read as a local file and fails.
+- The URL / storage-URI **pass-through** (leaving **HTTP(S) URLs** and existing **`pipelex-storage://` URIs** untouched) is a `prepare_inputs`-level behavior (see pass-through rules below), not a feature of `upload_file` itself: `prepare_inputs` classifies each string (URL vs local path) against the declared signature *before* it reaches `upload_file`.
+- Open file objects and streams are **deferred** — they can be added later without removing anything.
+
+The returned **upload record** guarantees, beyond the source identity:
+
+| Field | Guarantee |
+| --- | --- |
+| `uri` | The `pipelex-storage://` reference for the uploaded asset. |
+| content type (MIME) | Known client-side at upload time. |
+| size (bytes) | Known client-side at upload time. |
+| filename | Already in the wire model. |
+
+The MIME type and size are known client-side, so the record is assembled without extending the `/v1/upload` response. There is deliberately **no checksum field**: within-preparation dedup keys on source identity (not hashing), and cross-preparation dedup is a hosted storage-policy concern (Phase 5).
+
+### `prepare_inputs` — signature-driven input preparation
+
+```
+prepare_inputs(method_ref, pipe, inputs) → PreparedInputs
+```
+
+Takes the **method reference** (bundle files or catalog `method_id`) plus the target **pipe**, resolves the pipe's declared input signature, interprets the caller's compact `inputs` top-down against that signature, uploads the file-bearing values, and returns `PreparedInputs`:
+
+- `inputs` — a **copy** of the caller's inputs with each asset reference replaced by the canonical content shape carrying `pipelex-storage://` in its `url` field (see "Rewritten-input shape" below). Copy-on-write: the caller's original object is never mutated.
+- `uploads` — one upload record per prepared asset (the `upload_file` record shape), exposing `uri` so callers can log which source became which reference without reverse-engineering the rewritten object.
+
+The prepared `inputs` are passed to the existing run lifecycle unchanged.
+
+## Signature-driven asset identification
+
+The SDK **must not** guess that every string resembling a path is an asset — that would make ordinary text inputs environment-dependent and could upload unintended files. Interpretation comes from the method's **declared signature**, never from a value's shape alone. This mirrors the runtime's own top-down interpretation (`pipelex/pipelex/core/memory/input_shaper.py`, `InputShaper`) combined with the file-reference resolution of `pipelex/pipelex/pipeline/input_normalizer.py`, so local and hosted execution read the same compact inputs the same way.
+
+The declared signature is resolved via the explicit inputs template (`build_inputs` with `explicit=True`), which carries concept identity, canonical content shape, and multiplicity per input.
+
+Interpretation per declared input:
+
+- A bare string, `Path`, or `bytes` value at an **Image/Document-declared** input is a **file reference**: local paths, data URLs, and bytes are uploaded and rewritten to `pipelex-storage://` URIs; HTTP(S) URLs and existing `pipelex-storage://` URIs pass through unchanged.
+- The **identical** bare string at a **Text-declared** input is text and is never touched.
+- **Canonical image/document content structures** are recognized by their URL-bearing fields wherever they appear, including nested in structured objects and lists — exactly as the runtime normalizer walks them. The refining case matters: a concept refining `Image`/`PDF` is classified by the **canonical content shape**, not by the concept ref alone.
+- Inputs declared **`Dynamic`** are not path-interpreted (the signature genuinely cannot guide them); they accept canonical content structures or already-prepared references only.
+- A repeated reference to the **same source** within one preparation is uploaded once and rewritten consistently (within-preparation dedup by source identity).
+
+### Pass-through rules
+
+| Source at a file-bearing input | Action |
+| --- | --- |
+| Local path (`str`/`Path`) / data URL / `bytes` | Upload → rewrite to `pipelex-storage://` |
+| Existing `pipelex-storage://` URI | Already prepared — pass through unchanged |
+| HTTP(S) URL | Pass through unchanged, **unless** the caller explicitly asks to ingest it into Pipelex storage |
+
+## Rewritten-input shape: `url` carries the URI
+
+The runtime's canonical image/document content stores its reference in a **`url`** field. Preparation emits inputs the runtime interprets natively, so a rewritten input keeps the canonical content shape with `url` holding the `pipelex-storage://` value — exactly what the runtime's `input_normalizer` writes.
+
+The "uploaded reference is named `uri`" decision applies to the **upload surface**: the raw upload result and each upload record expose the storage reference as `uri`. Preparation must **not** invent a `uri` field inside rewritten inputs — that would produce inputs the runtime does not recognize.
+
+## Error and capability behavior
+
+Upload is a **hosted Pipelex-product capability**, even though the SDK can be pointed at other base URLs. A deployment that does not support upload must raise a specific, actionable exception — preparation must never silently leave a local path in place and let a later run fail obscurely.
+
+The contract distinguishes at least these semantic outcomes (exact typed exception classes are settled during implementation):
+
+- **invalid local source** — missing or unreadable path;
+- **rejected asset** — the server refused it (e.g. a `413` past the service-defined size cap — see "Storage policy" — surfaced as a clear rejection, not a raw transport error);
+- **unsupported server capability** — the configured deployment has no upload route;
+- **authentication / authorization failure** — `401` / `403`;
+- **transport failure** — network / server fault.
+
+All preparation failures are raised **before any run is created**.
+
+## Storage policy (inherited, Phase 1)
+
+The SDK ships against **today's route behavior**: a service-defined size cap (hosted default 50 MiB via `MAX_UPLOAD_MIB`, rejected with `413`), auth required, per-user keys, and nothing else — no MIME validation, retention, quotas, dedup, or cleanup. The SDK documents limits as **service-defined** and surfaces server rejections as clear "rejected asset" errors; it does **not** hardcode a client-side cap. Real storage policy (retention, quotas, org scoping, cleanup) is a later hosted-owner deliverable.
+
+## Stability across the future endpoint move
+
+The public abstraction sits deliberately **above** the HTTP route. Callers depend on `upload_file` / `prepare_inputs`, the `uri` result field, and the `pipelex-storage://` scheme — never on which backend service owns the route. The current transport is `POST /v1/upload` on `pipelex-api`; when hosted storage upload later moves to `pipelex-platform` (together with its paired resolution route, as one storage domain), the public path and wire shape are kept compatible so released SDK versions keep working, and any wire-protocol change is absorbed inside the SDK's upload transport. `upload_file`, `prepare_inputs`, and the prepared run-input shape stay stable across that move.
diff --git a/pipelex_sdk/build_models.py b/pipelex_sdk/build_models.py
new file mode 100644
index 0000000..a1c4086
--- /dev/null
+++ b/pipelex_sdk/build_models.py
@@ -0,0 +1,99 @@
+"""Wire models for the `/v1/build/inputs` route — the signature source `prepare_inputs`
+reads to resolve a pipe's declared inputs.
+
+The Python SDK had no `/v1/build/*` coverage; `prepare_inputs` needs the explicit
+inputs template, so this adds the `build_inputs` counterpart of `pipelex-sdk-js`'s
+`buildInputs` (only this route — the other build projections are not needed here).
+A produced verdict is a `200` discriminated on `is_valid`; a no-verdict condition
+(unknown `pipe_ref`, auth, server fault) throws `ApiResponseError`.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated, Any, Literal, Self, TypeAlias
+
+from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, model_validator
+
+from pipelex_sdk.validation_models import ValidationErrorItem
+
+InputsTemplateFormat = Literal["json", "toml"]
+
+
+class MthdsFileItem(BaseModel):
+    """One MTHDS file in a build closure. `source` is an optional provenance label the
+    server threads onto diagnostics raised from this file.
+    """
+
+    content: str
+    source: str | None = None
+
+
+class BuildInputsRequest(BaseModel):
+    """Request for `POST /v1/build/inputs`. The closure is supplied as inline `files`."""
+
+    files: list[MthdsFileItem]
+    pipe_ref: str | None = None
+    format: InputsTemplateFormat = "json"
+    explicit: bool = False
+
+
+class BuildInputsValidReport(BaseModel):
+    """The `is_valid: true` arm. The template rides `inputs` (json) or `inputs_toml` (toml)."""
+
+    model_config = ConfigDict(extra="allow")
+
+    is_valid: Literal[True]
+    pipe_ref: str
+    requested_pipe_ref: str | None = None
+    message: str
+    format: InputsTemplateFormat
+    explicit: bool
+    inputs: dict[str, Any] | None = None
+    inputs_toml: str | None = None
+
+    @model_validator(mode="after")
+    def _template_matches_format(self) -> Self:
+        # Honor the adapter's malformed-200 guarantee for the template shape too: a valid
+        # verdict must carry the template field its `format` selects (and not the other).
+        # Without this, an `is_valid: true` body missing both templates would parse as a valid
+        # report and only fail one layer down in `prepare_inputs`.
+        match self.format:
+            case "json":
+                if self.inputs is None:
+                    msg = "inputs is required when format is 'json'"
+                    raise ValueError(msg)
+                if self.inputs_toml is not None:
+                    msg = "inputs_toml must be absent when format is 'json'"
+                    raise ValueError(msg)
+            case "toml":
+                if self.inputs_toml is None:
+                    msg = "inputs_toml is required when format is 'toml'"
+                    raise ValueError(msg)
+                if self.inputs is not None:
+                    msg = "inputs must be absent when format is 'toml'"
+                    raise ValueError(msg)
+        return self
+
+
+class CrateInvalidReport(BaseModel):
+    """The `is_valid: false` arm shared by the build routes — an unresolvable closure is a
+    produced verdict on a `200`, never a thrown error. Branch on `is_valid`, not transport.
+    """
+
+    model_config = ConfigDict(extra="allow")
+
+    is_valid: Literal[False]
+    validation_errors: list[ValidationErrorItem]
+    message: str
+
+
+BuildInputsResponse: TypeAlias = Annotated[
+    BuildInputsValidReport | CrateInvalidReport,
+    Field(discriminator="is_valid"),
+]
+
+# The single parse path for a 200 `/build/inputs` body — discriminated on `is_valid`, built once at
+# import (TypeAdapter construction is expensive), mirroring `PipelexValidationResultAdapter`. A
+# malformed 200 (or an empty body) raises a clean `pydantic.ValidationError` rather than being
+# mistaken for a valid verdict.
+BuildInputsResponseAdapter: TypeAdapter[BuildInputsResponse] = TypeAdapter(BuildInputsResponse)  # pylint: disable=invalid-name
diff --git a/pipelex_sdk/client.py b/pipelex_sdk/client.py
index e62a59e..114b48b 100644
--- a/pipelex_sdk/client.py
+++ b/pipelex_sdk/client.py
@@ -32,6 +32,7 @@
 from pydantic_core import to_json
 from typing_extensions import override
 
+from pipelex_sdk.build_models import BuildInputsRequest, BuildInputsResponse, BuildInputsResponseAdapter, MthdsFileItem
 from pipelex_sdk.errors import (
     ApiResponseError,
     ApiUnreachableError,
@@ -42,6 +43,8 @@
     RunTimeoutError,
 )
 from pipelex_sdk.execute_result import PipelexExecuteResult
+from pipelex_sdk.prepare_inputs import PreparedInputs
+from pipelex_sdk.prepare_inputs import prepare_inputs as _prepare_inputs_impl
 from pipelex_sdk.product_models import (
     BillingPortalResponse,
     ChangePlanResponse,
@@ -71,6 +74,8 @@
     RunStatus,
     WaitForResultOptions,
 )
+from pipelex_sdk.upload import UploadRecord, UploadSource
+from pipelex_sdk.upload import upload_file as _upload_file_impl
 from pipelex_sdk.validation_models import PipelexValidationResultAdapter, ValidationErrorItem
 
 if TYPE_CHECKING:
@@ -817,6 +822,49 @@ async def upload(self, upload_input: UploadInput) -> UploadedFile:
         body = upload_input.model_dump(mode="json", exclude_none=True)
         return UploadedFile.model_validate(await self._request_product("POST", "upload", body=body))
 
+    async def build_inputs(self, request: BuildInputsRequest) -> BuildInputsResponse:
+        """Project a pipe's declared inputs as a fill-in template — `POST /v1/build/inputs`.
+
+        Returns a 200 verdict: branch on `is_valid` before reading the arm — an unresolvable
+        closure comes back as `is_valid: false` with `validation_errors`, not a thrown error.
+        A no-verdict condition (unknown `pipe_ref`, auth, server fault) raises `ApiResponseError`.
+        This is the signature source `prepare_inputs` reads (with `explicit=True`).
+        """
+        body = request.model_dump(mode="json", exclude_none=True)
+        raw = await self._request_product("POST", "build/inputs", body=body)
+        return BuildInputsResponseAdapter.validate_python(raw)
+
+    async def upload_file(
+        self,
+        source: UploadSource,
+        *,
+        filename: str | None = None,
+        content_type: str | None = None,
+    ) -> UploadRecord:
+        """Upload one local asset and return its `UploadRecord` — the single-asset convenience
+        over `upload`. `source` is a filesystem path (`str`/`Path`) or raw `bytes`. The record
+        guarantees `uri`, `content_type`, `size`, and `filename`. Transport failures surface as
+        the semantic input-preparation errors (rejected asset, auth, unsupported capability,
+        transport). See `docs/input-preparation.md`.
+        """
+        return await _upload_file_impl(self, source, filename=filename, content_type=content_type)
+
+    async def prepare_inputs(
+        self,
+        *,
+        files: list[MthdsFileItem],
+        pipe_ref: str | None = None,
+        inputs: dict[str, Any],
+    ) -> PreparedInputs:
+        """Prepare a pipe's inputs — resolve the declared signature, upload the file-bearing
+        assets, and return copy-on-write rewritten inputs (canonical content carrying
+        `pipelex-storage://` in `url`) plus one upload record per prepared asset. HTTP(S) URLs
+        and existing `pipelex-storage://` URIs pass through unchanged; all failures are raised
+        before any run is created. The caller supplies the method closure as inline `files`.
+        See `docs/input-preparation.md`.
+        """
+        return await _prepare_inputs_impl(self, files=files, pipe_ref=pipe_ref, inputs=inputs)
+
     async def list_runs(self, method_id: str) -> list[PipelineRun]:
         """List a method's runs — `GET /v1/runs?method_id={methodId}`."""
         result = await self._request_product("GET", f"{_RUNS}?method_id={quote(method_id, safe='')}")
diff --git a/pipelex_sdk/errors.py b/pipelex_sdk/errors.py
index 997e93a..f7afb63 100644
--- a/pipelex_sdk/errors.py
+++ b/pipelex_sdk/errors.py
@@ -161,3 +161,55 @@ class RunLifecycleUnavailableError(PipelineRequestError):
     def __init__(self, message: str, api_url: str) -> None:
         super().__init__(message)
         self.api_url = api_url
+
+
+class InputPreparationError(PipelineRequestError):
+    """Base class for every failure raised by input preparation (`upload_file` /
+    `prepare_inputs`).
+
+    Catch this to handle any preparation failure; catch a subclass to branch on the
+    semantic category. All preparation failures are raised BEFORE any run is created —
+    a run never triggers a hidden upload. Mirrors `pipelex-sdk-js`'s
+    `InputPreparationError` family.
+    """
+
+
+class InvalidLocalSourceError(InputPreparationError):
+    """A local asset could not be turned into bytes — a missing or unreadable path.
+    `source` is the offending path.
+    """
+
+    def __init__(self, message: str, source: str) -> None:
+        super().__init__(message)
+        self.source = source
+
+
+class RejectedAssetError(InputPreparationError):
+    """The server refused the asset — most commonly a `413` past the service-defined
+    size cap. The SDK imposes no client-side cap; it surfaces the server's rejection.
+    `filename` and `status` locate it.
+    """
+
+    def __init__(self, message: str, filename: str, status: int) -> None:
+        super().__init__(message)
+        self.filename = filename
+        self.status = status
+
+
+class UnsupportedUploadCapabilityError(InputPreparationError):
+    """The configured deployment does not support upload (no `/v1/upload` route, seen
+    as a `404`). Upload is a hosted Pipelex-product capability even though the SDK can
+    be pointed at other base URLs.
+    """
+
+
+class UploadAuthenticationError(InputPreparationError):
+    """Upload was not authorized — a `401`/`403` from the upload route."""
+
+    def __init__(self, message: str, status: int) -> None:
+        super().__init__(message)
+        self.status = status
+
+
+class UploadTransportError(InputPreparationError):
+    """A network or server fault reaching the upload route (unreachable host, `5xx`)."""
diff --git a/pipelex_sdk/prepare_inputs.py b/pipelex_sdk/prepare_inputs.py
new file mode 100644
index 0000000..7857ad8
--- /dev/null
+++ b/pipelex_sdk/prepare_inputs.py
@@ -0,0 +1,208 @@
+"""`prepare_inputs` — signature-driven input preparation. Resolves the target pipe's
+declared inputs via the explicit inputs template, interprets the caller's compact inputs
+top-down against it, uploads the file-bearing values, and returns rewritten inputs
+(canonical content carrying `pipelex-storage://` in `url`) plus one upload record per
+prepared asset. Python counterpart of `pipelex-sdk-js`'s `prepareInputs`.
+
+The classification mirrors the runtime: `pipelex`'s `input_normalizer` walks
+Image/Document contents (recognized by their `url`-bearing shape, incl. nested in
+structured content) and `resolve_uri` decides upload vs pass-through. The declared
+signature comes from the explicit template (`build_inputs`, `explicit=True`), whose
+canonical content shape is the classifier — the file signal is a value that is a dict
+containing a `url` key. See the shared behavior matrix (`wip/upload/behavior-matrix.md`)
+and `docs/input-preparation.md`.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import re
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Protocol, cast
+from urllib.parse import unquote_to_bytes
+
+from pydantic import BaseModel
+
+from pipelex_sdk.build_models import BuildInputsRequest, BuildInputsResponse, CrateInvalidReport, MthdsFileItem
+from pipelex_sdk.errors import InputPreparationError
+from pipelex_sdk.upload import UploadRecord, UploadSource, upload_file
+
+if TYPE_CHECKING:
+    from pipelex_sdk.product_models import UploadedFile, UploadInput
+
+PIPELEX_STORAGE_SCHEME = "pipelex-storage://"
+_HTTP_URL_RE = re.compile(r"^https?://", re.IGNORECASE)
+
+
+class PreparedInputs(BaseModel):
+    """The result of `prepare_inputs`: rewritten inputs (copy-on-write) plus upload records.
+
+    `inputs` is a copy of the caller's inputs with each file-bearing value rewritten to
+    canonical content carrying `pipelex-storage://` in `url`. `uploads` carries one record
+    per uploaded asset — pass-through references (http(s), existing storage URIs) produce none.
+    """
+
+    inputs: dict[str, Any]
+    uploads: list[UploadRecord]
+
+
+class _PrepareClient(Protocol):
+    """The client surface `prepare_inputs` needs: raw `upload` plus the `build_inputs` signature source."""
+
+    async def upload(self, upload_input: UploadInput) -> UploadedFile: ...
+
+    async def build_inputs(self, request: BuildInputsRequest) -> BuildInputsResponse: ...
+
+
+class _PrepareContext:
+    """Mutable state threaded through one preparation walk."""
+
+    def __init__(self, client: _PrepareClient) -> None:
+        self.client = client
+        self.uploads: list[UploadRecord] = []
+        # Dedup by source identity: same source (str/bytes/Path value) uploads once.
+        self.dedup: dict[UploadSource, str] = {}
+
+
+def _is_file_content(node: Any) -> bool:
+    """A canonical Image/Document content is a dict carrying a `url` key."""
+    return isinstance(node, dict) and "url" in node
+
+
+def _decode_data_url(data_url: str) -> tuple[bytes, str]:
+    """Decode a `data:` URL into bytes plus its MIME type.
+
+    A base64 payload is decoded with `validate=True` so junk characters are rejected rather
+    than silently discarded (which would upload corrupted bytes), and a decode failure (bad
+    padding or non-alphabet input) surfaces as a typed `InputPreparationError` — never a raw
+    `binascii.Error` escaping the preparation contract. A non-base64 payload decodes straight
+    to bytes via `unquote_to_bytes`, so percent-encoded binary keeps its exact bytes (decoding
+    it as UTF-8 text first would corrupt any byte ≥ 0x80).
+    """
+    comma = data_url.find(",")
+    if comma < 0:
+        msg = f"Malformed data URL (no comma separator): {data_url[:32]}…"
+        raise InputPreparationError(msg)
+    header = data_url[5:comma]  # strip "data:"
+    payload = data_url[comma + 1 :]
+    content_type = header.split(";")[0] or "application/octet-stream"
+    if ";base64" in header.lower():
+        try:
+            decoded = base64.b64decode(payload, validate=True)
+        except binascii.Error as exc:
+            msg = f"Malformed data URL: the base64 payload is not valid ({exc})."
+            raise InputPreparationError(msg) from exc
+        return decoded, content_type
+    return unquote_to_bytes(payload), content_type
+
+
+async def _do_resolve_source(ctx: _PrepareContext, source: Any) -> str:
+    """Resolve one source to the URL/URI to write."""
+    if isinstance(source, str):
+        if source.startswith(PIPELEX_STORAGE_SCHEME):
+            return source  # already prepared
+        if _HTTP_URL_RE.match(source):
+            return source  # reachable URL — pass through
+        if source.startswith("data:"):
+            data, content_type = _decode_data_url(source)
+            record = await upload_file(ctx.client, data, content_type=content_type)
+            ctx.uploads.append(record)
+            return record.uri
+        # Anything else is a local filesystem path.
+        record = await upload_file(ctx.client, source)
+        ctx.uploads.append(record)
+        return record.uri
+    if isinstance(source, (bytes, Path)):
+        record = await upload_file(ctx.client, source)
+        ctx.uploads.append(record)
+        return record.uri
+    # An unrecognized value sits at a file-bearing position (neither a source string,
+    # bytes/Path, nor a canonical {url} content dict). Fail with a typed error rather than
+    # passing an unusable value through to a later run.
+    msg = (
+        "Unsupported value at a file input: expected a path (str/Path), bytes, a data URL, "
+        f"an http(s)/pipelex-storage:// URL, or canonical {{url}} content; got {type(source).__name__}."
+    )
+    raise InputPreparationError(msg)
+
+
+async def _resolve_source(ctx: _PrepareContext, source: Any) -> str:
+    """Resolve a source, deduped by identity (same source uploads once)."""
+    hashable = isinstance(source, (str, bytes, Path))
+    if hashable and source in ctx.dedup:
+        return ctx.dedup[source]
+    resolved = await _do_resolve_source(ctx, source)
+    if hashable:
+        ctx.dedup[source] = resolved
+    return resolved
+
+
+async def _resolve_file_position(ctx: _PrepareContext, caller_value: Any) -> Any:
+    """Resolve a value known to sit at a file position into canonical content with a rewritten `url`."""
+    if isinstance(caller_value, dict) and "url" in caller_value:
+        content = cast("dict[str, Any]", caller_value)
+        resolved = await _resolve_source(ctx, content["url"])
+        return {**content, "url": resolved}
+    resolved = await _resolve_source(ctx, caller_value)
+    return {"url": resolved}
+
+
+async def _resolve_node(ctx: _PrepareContext, template_node: Any, caller_value: Any) -> Any:
+    """Template-guided walk: a template node that is canonical file content marks a file position."""
+    if _is_file_content(template_node):
+        return await _resolve_file_position(ctx, caller_value)
+    if isinstance(template_node, list) and template_node:
+        element_template = cast("list[Any]", template_node)[0]
+        if isinstance(caller_value, list):
+            items = cast("list[Any]", caller_value)
+            return [await _resolve_node(ctx, element_template, item) for item in items]
+        return caller_value  # shape mismatch — leave it for the run to reject
+    if isinstance(template_node, dict) and isinstance(caller_value, dict):
+        template_dict = cast("dict[str, Any]", template_node)
+        caller_dict = cast("dict[str, Any]", caller_value)
+        result: dict[str, Any] = dict(caller_dict)
+        for key in template_dict:
+            if key in caller_dict:
+                result[key] = await _resolve_node(ctx, template_dict[key], caller_dict[key])
+        return result
+    return caller_value  # scalar (text/number/…) or shape mismatch — pass through
+
+
+async def prepare_inputs(
+    client: _PrepareClient,
+    *,
+    files: list[MthdsFileItem],
+    pipe_ref: str | None = None,
+    inputs: dict[str, Any],
+) -> PreparedInputs:
+    """Prepare a pipe's inputs: upload local/byte/data-URL assets at the signature's
+    file-bearing positions and return copy-on-write rewritten inputs plus upload records.
+
+    HTTP(S) URLs and existing `pipelex-storage://` URIs pass through unchanged. All failures
+    are raised before any run is created. The declared signature is resolved from the inline
+    `files` closure; a closure that does not resolve raises `InputPreparationError`. No-verdict
+    conditions from the signature route (unknown `pipe_ref`, auth, server fault) surface as the
+    build route's `ApiResponseError`.
+    """
+    report = await client.build_inputs(BuildInputsRequest(files=files, pipe_ref=pipe_ref, format="json", explicit=True))
+    if isinstance(report, CrateInvalidReport):
+        first = report.validation_errors[0].message if report.validation_errors else report.message
+        msg = f"Cannot prepare inputs: the method signature did not resolve — {first}"
+        raise InputPreparationError(msg)
+    if report.format != "json" or report.inputs is None:
+        msg = f'Cannot prepare inputs: expected a JSON inputs template, got "{report.format}".'
+        raise InputPreparationError(msg)
+    template = report.inputs
+
+    ctx = _PrepareContext(client)
+    rewritten = dict(inputs)
+    for name, caller_value in inputs.items():
+        entry = template.get(name)
+        if not isinstance(entry, dict) or "content" not in entry:
+            # Not a declared input (or an unexpected envelope) — pass through untouched.
+            continue
+        content = cast("dict[str, Any]", entry)["content"]
+        rewritten[name] = await _resolve_node(ctx, content, caller_value)
+
+    return PreparedInputs(inputs=rewritten, uploads=ctx.uploads)
diff --git a/pipelex_sdk/upload.py b/pipelex_sdk/upload.py
new file mode 100644
index 0000000..dc8edcc
--- /dev/null
+++ b/pipelex_sdk/upload.py
@@ -0,0 +1,132 @@
+"""`upload_file` — the single-asset upload convenience over the raw `upload()` wire
+call. Accepts `str`/`pathlib.Path` filesystem paths and raw `bytes`, and returns an
+`UploadRecord` assembled client-side. Python counterpart of `pipelex-sdk-js`'s
+`uploadFile`. See `docs/input-preparation.md`.
+
+The MIME type and size are known client-side at upload time, so the record is built
+without extending the `/v1/upload` response.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import mimetypes
+from pathlib import Path
+from typing import Protocol
+
+from pydantic import BaseModel
+
+from pipelex_sdk.errors import (
+    ApiResponseError,
+    ApiUnreachableError,
+    InputPreparationError,
+    InvalidLocalSourceError,
+    RejectedAssetError,
+    UnsupportedUploadCapabilityError,
+    UploadAuthenticationError,
+    UploadTransportError,
+)
+from pipelex_sdk.product_models import UploadedFile, UploadInput
+
+DEFAULT_CONTENT_TYPE = "application/octet-stream"
+DEFAULT_FILENAME = "upload.bin"
+
+# A local asset `upload_file` accepts. A bare string in `upload_file` is a filesystem
+# path; `prepare_inputs` classifies strings (url vs path) before they reach here.
+UploadSource = str | Path | bytes
+
+
+class UploadRecord(BaseModel):
+    """The record `upload_file` returns for a prepared asset. Beyond the source identity it
+    guarantees the resulting `uri`, the MIME `content_type`, the `size` in bytes, and the
+    `filename`. A content checksum is deliberately not included — best-effort at most, and
+    within-preparation dedup keys on source identity.
+    """
+
+    uri: str
+    filename: str
+    content_type: str
+    size: int
+
+
+class _UploadClient(Protocol):
+    """The client surface `upload_file` needs — the raw base64 `upload` wire call."""
+
+    async def upload(self, upload_input: UploadInput) -> UploadedFile: ...
+
+
+def _guess_content_type(filename: str) -> str:
+    """MIME guess from a filename extension; `application/octet-stream` when unknown."""
+    guessed, _ = mimetypes.guess_type(filename)
+    return guessed or DEFAULT_CONTENT_TYPE
+
+
+def _read_path(path: Path) -> bytes:
+    """Read a filesystem path into bytes, mapping read failures to `InvalidLocalSourceError`."""
+    try:
+        return path.read_bytes()
+    except OSError as exc:
+        msg = f'Local file cannot be read: "{path}" ({type(exc).__name__}).'
+        raise InvalidLocalSourceError(msg, source=str(path)) from exc
+
+
+def _to_asset_bytes(source: UploadSource, filename: str | None, content_type: str | None) -> tuple[bytes, str, str]:
+    """Normalize any accepted asset form into bytes plus a filename and MIME."""
+    if isinstance(source, bytes):
+        resolved_name = filename or DEFAULT_FILENAME
+        return source, resolved_name, content_type or _guess_content_type(resolved_name)
+    path = Path(source)
+    data = _read_path(path)
+    resolved_name = filename or path.name or DEFAULT_FILENAME
+    return data, resolved_name, content_type or _guess_content_type(resolved_name)
+
+
+def _map_upload_error(error: ApiResponseError | ApiUnreachableError, filename: str) -> InputPreparationError:
+    """Translate a raw `upload()` transport error into the matching preparation error."""
+    if isinstance(error, ApiUnreachableError):
+        msg = f'Upload of "{filename}" could not reach the Pipelex API ({error.code or "unreachable"}).'
+        return UploadTransportError(msg)
+    match error.status:
+        case 413:
+            detail = error.server_message or "asset exceeds the service size limit"
+            return RejectedAssetError(f'The server rejected "{filename}": {detail}.', filename=filename, status=error.status)
+        case 401 | 403:
+            return UploadAuthenticationError(
+                f'Upload of "{filename}" was not authorized ({error.status}). Check the configured Pipelex API key.',
+                status=error.status,
+            )
+        case 404:
+            return UnsupportedUploadCapabilityError(
+                "The configured Pipelex deployment does not support file upload (no /v1/upload route). Upload is a hosted Pipelex capability."
+            )
+        case _:
+            detail = error.server_message or error.status_text
+            return UploadTransportError(f'Upload of "{filename}" failed ({error.status}): {detail}.')
+
+
+async def upload_file(
+    client: _UploadClient,
+    source: UploadSource,
+    *,
+    filename: str | None = None,
+    content_type: str | None = None,
+) -> UploadRecord:
+    """Upload one local asset and return its `UploadRecord`.
+
+    `source` is a filesystem path (`str`/`Path`) or raw `bytes`. Maps the raw `upload()`
+    transport errors onto the semantic input-preparation errors: a `413` is a rejected
+    asset, `401`/`403` an auth failure, `404` an unsupported upload capability, an
+    unreachable host a transport failure.
+    """
+    # Offload the (possibly large) synchronous file read off the event loop — `read_bytes`
+    # releases the GIL during the underlying os.read, so other coroutines run during disk I/O.
+    # base64 stays inline on purpose: CPython's binascii holds the GIL, so threading it would
+    # not free the loop (and this matches the JS SDK, which also reads off-loop but encodes inline).
+    data, resolved_name, resolved_type = await asyncio.to_thread(_to_asset_bytes, source, filename, content_type)
+    encoded = base64.b64encode(data).decode("ascii")
+    try:
+        uploaded = await client.upload(UploadInput(filename=resolved_name, data=encoded, content_type=resolved_type))
+    except (ApiResponseError, ApiUnreachableError) as exc:
+        raise _map_upload_error(exc, resolved_name) from exc
+    return UploadRecord(uri=uploaded.uri, filename=uploaded.filename, content_type=resolved_type, size=len(data))
diff --git a/pyproject.toml b/pyproject.toml
index 3157342..92e83a9 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -318,6 +318,7 @@ convention = "google"
   "INP001",  # Allow test files to not have __init__.py in their directories (avoids namespace collisions)
   "SLF001",  # Unit tests legitimately probe private transport/error helpers (e.g. _request_product, _request_json)
   "PLC2701", # Unit tests legitimately import private module helpers under test (e.g. _parse_error_body)
+  "ARG002",  # Test-double methods match a Protocol signature; an unused param (e.g. a fake build_inputs ignoring `request`) is intentional
 ]
 "examples/**/*.py" = [
   "INP001", # Runnable demo scripts, not an importable package
diff --git a/tests/unit/test_build_inputs.py b/tests/unit/test_build_inputs.py
new file mode 100644
index 0000000..3e07a78
--- /dev/null
+++ b/tests/unit/test_build_inputs.py
@@ -0,0 +1,107 @@
+"""The `build_inputs` route — the signature source `prepare_inputs` reads. Pins the verb +
+path + body, the 200-verdict discipline (branch on `is_valid`), and the no-verdict throw.
+
+Ports the relevant slice of `pipelex-sdk-js/tests/build-routes.test.ts` for `/v1/build/inputs`.
+`_send` is mocked; a produced verdict is a 200 discriminated on `is_valid`, a no-verdict
+condition throws `ApiResponseError`.
+"""
+
+import asyncio
+import json
+
+import httpx
+import pytest
+from pydantic import ValidationError
+from pytest_mock import MockerFixture, MockType
+
+from pipelex_sdk.build_models import BuildInputsRequest, BuildInputsResponseAdapter, BuildInputsValidReport, CrateInvalidReport, MthdsFileItem
+from pipelex_sdk.client import PipelexAPIClient
+from pipelex_sdk.errors import ApiResponseError
+
+_BASE_URL = "http://localhost:8081"
+
+
+def _response(status_code: int, *, json_body: object | None = None) -> httpx.Response:
+    request = httpx.Request("POST", f"{_BASE_URL}/x")
+    if json_body is not None:
+        return httpx.Response(status_code, json=json_body, request=request)
+    return httpx.Response(status_code, request=request)
+
+
+class TestBuildInputs:
+    def _client(self) -> PipelexAPIClient:
+        return PipelexAPIClient(api_key="test-token", base_url=_BASE_URL)
+
+    def _mock_send(self, mocker: MockerFixture, client: PipelexAPIClient, response: httpx.Response) -> MockType:
+        return mocker.patch.object(client, "_send", mocker.AsyncMock(return_value=response))
+
+    def test_posts_files_and_flags_to_build_inputs(self, mocker: MockerFixture) -> None:
+        client = self._client()
+        valid = {
+            "is_valid": True,
+            "pipe_ref": "demo.main",
+            "message": "ok",
+            "format": "json",
+            "explicit": True,
+            "inputs": {"photo": {"concept": "demo.Photo", "content": {"url": "https://mock/p.png"}}},
+        }
+        send = self._mock_send(mocker, client, _response(200, json_body=valid))
+
+        report = asyncio.run(
+            client.build_inputs(BuildInputsRequest(files=[MthdsFileItem(content='domain = "demo"', source="b.mthds")], format="json", explicit=True))
+        )
+
+        call = send.call_args
+        assert call.args[0] == "POST"
+        assert call.args[1] == f"{_BASE_URL}/v1/build/inputs"
+        body = json.loads(call.kwargs["content"])
+        assert body == {"files": [{"content": 'domain = "demo"', "source": "b.mthds"}], "format": "json", "explicit": True}
+        assert isinstance(report, BuildInputsValidReport)
+        assert report.pipe_ref == "demo.main"
+        assert report.inputs is not None
+
+    def test_invalid_closure_is_a_200_verdict(self, mocker: MockerFixture) -> None:
+        client = self._client()
+        invalid = {
+            "is_valid": False,
+            "message": "closure did not validate",
+            "validation_errors": [{"category": "blueprint_validation", "message": "unknown pipe type"}],
+        }
+        self._mock_send(mocker, client, _response(200, json_body=invalid))
+
+        report = asyncio.run(client.build_inputs(BuildInputsRequest(files=[MthdsFileItem(content="x")])))
+
+        assert isinstance(report, CrateInvalidReport)
+        assert report.validation_errors[0].message == "unknown pipe type"
+
+    @pytest.mark.parametrize(
+        "body",
+        [
+            # format=json but carrying no template at all — the flagged malformed-200 shape.
+            {"is_valid": True, "pipe_ref": "demo.main", "message": "ok", "format": "json", "explicit": True},
+            # format=json but carrying the toml template (mismatched/opposite shape).
+            {"is_valid": True, "pipe_ref": "demo.main", "message": "ok", "format": "json", "explicit": True, "inputs_toml": "x = 1"},
+            # format=toml but carrying no toml template.
+            {"is_valid": True, "pipe_ref": "demo.main", "message": "ok", "format": "toml", "explicit": True},
+        ],
+    )
+    def test_valid_report_without_matching_template_is_rejected(self, body: dict[str, object]) -> None:
+        # A valid verdict must carry the template its `format` selects — the adapter's
+        # malformed-200 guarantee, now honored for the template shape too.
+        with pytest.raises(ValidationError):
+            BuildInputsResponseAdapter.validate_python(body)
+
+    def test_valid_toml_report_is_accepted(self) -> None:
+        body = {"is_valid": True, "pipe_ref": "demo.main", "message": "ok", "format": "toml", "explicit": True, "inputs_toml": "photo = 1"}
+        report = BuildInputsResponseAdapter.validate_python(body)
+        assert isinstance(report, BuildInputsValidReport)
+        assert report.inputs_toml == "photo = 1"
+
+    def test_no_verdict_422_raises_api_response_error(self, mocker: MockerFixture) -> None:
+        client = self._client()
+        problem = {"detail": "Unknown pipe_ref", "error_type": "PipeNotFound"}
+        self._mock_send(mocker, client, _response(422, json_body=problem))
+
+        with pytest.raises(ApiResponseError) as exc_info:
+            asyncio.run(client.build_inputs(BuildInputsRequest(files=[MthdsFileItem(content="x")], pipe_ref="demo.nope")))
+        assert exc_info.value.status == 422
diff --git a/tests/unit/test_prepare_inputs.py b/tests/unit/test_prepare_inputs.py
new file mode 100644
index 0000000..18c79a8
--- /dev/null
+++ b/tests/unit/test_prepare_inputs.py
@@ -0,0 +1,246 @@
+"""`prepare_inputs` — signature-driven input preparation. Cases derive from the shared
+behavior matrix (`wip/upload/behavior-matrix.md`): file-bearing positions are found from the
+explicit template's canonical content shape (a `{"url": …}` dict), assets are uploaded and
+rewritten to `pipelex-storage://` in `url`, http(s)/storage references pass through, dedup
+keys on source identity, and the call is copy-on-write.
+
+Ports `pipelex-sdk-js/tests/prepare-inputs.test.ts`. The fake client returns a canned explicit
+template from `build_inputs` and a counting `upload`; one wiring test drives the real client.
+"""
+
+import asyncio
+import base64
+from pathlib import Path
+from typing import Any
+
+import httpx
+import pytest
+from pytest_mock import MockerFixture
+
+from pipelex_sdk.build_models import BuildInputsRequest, BuildInputsResponse, BuildInputsValidReport, CrateInvalidReport, MthdsFileItem
+from pipelex_sdk.client import PipelexAPIClient
+from pipelex_sdk.errors import ApiResponseError, InputPreparationError, RejectedAssetError
+from pipelex_sdk.prepare_inputs import prepare_inputs
+from pipelex_sdk.product_models import UploadedFile, UploadInput
+
+_BASE_URL = "http://localhost:8081"
+_FILES = [MthdsFileItem(content='domain = "demo"')]
+
+
+def _entry(concept: str, content: Any) -> dict[str, Any]:
+    return {"concept": concept, "content": content}
+
+
+class _FakePrepareClient:
+    """Fake client: `build_inputs` returns the given envelope template; `upload` counts calls."""
+
+    def __init__(self, template: dict[str, Any], *, report: BuildInputsResponse | None = None, upload_error: Exception | None = None) -> None:
+        self._template = template
+        self._report = report
+        self._upload_error = upload_error
+        self.upload_calls: list[UploadInput] = []
+        self._counter = 0
+
+    async def build_inputs(self, request: BuildInputsRequest) -> BuildInputsResponse:
+        if self._report is not None:
+            return self._report
+        return BuildInputsValidReport(is_valid=True, pipe_ref="demo.main", message="ok", format="json", explicit=True, inputs=self._template)
+
+    async def upload(self, upload_input: UploadInput) -> UploadedFile:
+        if self._upload_error is not None:
+            raise self._upload_error
+        self._counter += 1
+        self.upload_calls.append(upload_input)
+        return UploadedFile(uri=f"pipelex-storage://user/assets/{self._counter}.bin", filename=upload_input.filename)
+
+
+class TestPrepareInputs:
+    def test_uploads_top_level_image_bytes(self) -> None:
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})})
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": bytes([1, 2, 3])}))
+
+        assert prepared.inputs == {"photo": {"url": "pipelex-storage://user/assets/1.bin"}}
+        assert len(prepared.uploads) == 1
+        assert prepared.uploads[0].uri == "pipelex-storage://user/assets/1.bin"
+
+    def test_passes_http_url_through(self) -> None:
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})})
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": "https://example.com/real.png"}))
+
+        assert prepared.inputs == {"photo": {"url": "https://example.com/real.png"}}
+        assert prepared.uploads == []
+        assert client.upload_calls == []
+
+    def test_passes_existing_storage_uri_through(self) -> None:
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})})
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": "pipelex-storage://user/assets/already.png"}))
+
+        assert prepared.inputs == {"photo": {"url": "pipelex-storage://user/assets/already.png"}}
+        assert prepared.uploads == []
+
+    def test_decodes_and_uploads_data_url(self) -> None:
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})})
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": "data:image/png;base64,AQIDBA=="}))
+
+        assert prepared.inputs == {"photo": {"url": "pipelex-storage://user/assets/1.bin"}}
+        assert client.upload_calls[0].content_type == "image/png"
+        assert client.upload_calls[0].data == "AQIDBA=="
+
+    @pytest.mark.parametrize(
+        "data_url",
+        [
+            "data:image/png;base64,AQI",  # bad padding — binascii.Error
+            "data:image/png;base64,AQID!!!!",  # non-alphabet junk — rejected by validate=True
+        ],
+    )
+    def test_malformed_base64_data_url_raises_typed_error(self, data_url: str) -> None:
+        # A malformed base64 data URL must surface as the typed `InputPreparationError`
+        # (never a raw binascii.Error), and must never upload silently-corrupted bytes.
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})})
+
+        with pytest.raises(InputPreparationError):
+            asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": data_url}))
+        assert client.upload_calls == []
+
+    def test_percent_encoded_binary_data_url_keeps_exact_bytes(self) -> None:
+        # A non-base64 data URL carrying percent-encoded binary must upload its exact bytes;
+        # decoding as UTF-8 text first would corrupt any byte >= 0x80 (e.g. %FF).
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})})
+
+        asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": "data:application/octet-stream,%00%ff%01"}))
+
+        assert base64.b64decode(client.upload_calls[0].data) == bytes([0x00, 0xFF, 0x01])
+
+    def test_uploads_each_element_of_declared_multiple(self) -> None:
+        client = _FakePrepareClient({"exhibits": _entry("demo.Exhibit", [{"url": "https://mock/d.pdf"}])})
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"exhibits": [bytes([1]), bytes([2])]}))
+
+        assert prepared.inputs == {"exhibits": [{"url": "pipelex-storage://user/assets/1.bin"}, {"url": "pipelex-storage://user/assets/2.bin"}]}
+        assert len(prepared.uploads) == 2
+
+    def test_leaves_text_input_untouched(self) -> None:
+        client = _FakePrepareClient({"question": _entry("demo.Question", {"text": "text_value"})})
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"question": "notes/summary.txt"}))
+
+        assert prepared.inputs == {"question": "notes/summary.txt"}
+        assert client.upload_calls == []
+
+    def test_uploads_only_nested_image_of_structured_input(self) -> None:
+        client = _FakePrepareClient(
+            {"dossier": _entry("demo.Dossier", {"title": "title_value", "cover": {"url": "https://mock/c.png", "mime_type": "image/png"}})}
+        )
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"dossier": {"title": "Q3 report", "cover": bytes([7, 7])}}))
+
+        assert prepared.inputs == {"dossier": {"title": "Q3 report", "cover": {"url": "pipelex-storage://user/assets/1.bin"}}}
+        assert len(prepared.uploads) == 1
+
+    def test_does_not_path_interpret_bare_string_at_dynamic_input(self) -> None:
+        client = _FakePrepareClient({"freeform": _entry("native.Anything", {"whatever": "value"})})
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"freeform": "resembles/a/path"}))
+
+        assert prepared.inputs == {"freeform": "resembles/a/path"}
+        assert client.upload_calls == []
+
+    def test_uploads_canonical_image_nested_in_dynamic(self) -> None:
+        client = _FakePrepareClient({"data": _entry("native.Composite", {"text": "t", "images": [{"url": "https://mock/i.png"}]})})
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"data": {"text": "hi", "images": [bytes([5])]}}))
+
+        assert prepared.inputs == {"data": {"text": "hi", "images": [{"url": "pipelex-storage://user/assets/1.bin"}]}}
+        assert len(prepared.uploads) == 1
+
+    def test_dedups_by_source_identity(self) -> None:
+        client = _FakePrepareClient({"exhibits": _entry("demo.Exhibit", [{"url": "https://mock/d.pdf"}])})
+        shared = bytes([9, 9, 9])
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"exhibits": [shared, shared]}))
+
+        assert len(client.upload_calls) == 1
+        exhibits = prepared.inputs["exhibits"]
+        assert exhibits[0]["url"] == exhibits[1]["url"]
+
+    def test_is_copy_on_write(self) -> None:
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})})
+        original = {"photo": bytes([1, 2, 3])}
+
+        asyncio.run(prepare_inputs(client, files=_FILES, inputs=original))
+
+        assert original["photo"] == bytes([1, 2, 3])
+
+    def test_passes_through_undeclared_input(self) -> None:
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})})
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": "https://example.com/p.png", "stray": "left alone"}))
+
+        assert prepared.inputs["stray"] == "left alone"
+
+    def test_uploads_real_local_path(self, tmp_path: Path) -> None:
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})})
+        path = tmp_path / "shot.png"
+        path.write_bytes(bytes([1, 2, 3, 4]))
+
+        prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": str(path)}))
+
+        assert prepared.inputs == {"photo": {"url": "pipelex-storage://user/assets/1.bin"}}
+        assert client.upload_calls[0].content_type == "image/png"
+
+    def test_raises_for_unrecognized_value_at_file_position(self) -> None:
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})})
+
+        # A plain object that is neither a canonical {url} content nor bytes — a realistic
+        # caller typo — must surface as a typed error, not pass through unresolved.
+        with pytest.raises(InputPreparationError):
+            asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": {"mimeType": "image/png", "bytes": [1, 2, 3]}}))
+        assert client.upload_calls == []
+
+    def test_raises_when_signature_does_not_resolve(self) -> None:
+        report = CrateInvalidReport(is_valid=False, message="closure did not validate", validation_errors=[])
+        client = _FakePrepareClient({}, report=report)
+
+        with pytest.raises(InputPreparationError):
+            asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": bytes([1])}))
+
+    def test_surfaces_rejected_asset_before_returning(self) -> None:
+        error = ApiResponseError(
+            "HTTP 413", api_url=f"{_BASE_URL}/v1/upload", status=413, status_text="Payload Too Large", response_body="", server_message="too big"
+        )
+        client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}, upload_error=error)
+
+        with pytest.raises(RejectedAssetError):
+            asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": bytes([1])}))
+
+    def test_wires_through_the_real_client(self, mocker: MockerFixture) -> None:
+        client = PipelexAPIClient(api_key="test-token", base_url=_BASE_URL)
+        build_body = {
+            "is_valid": True,
+            "pipe_ref": "demo.main",
+            "message": "ok",
+            "format": "json",
+            "explicit": True,
+            "inputs": {"photo": {"concept": "demo.Photo", "content": {"url": "https://mock/p.png"}}},
+        }
+        upload_body = {"uri": "pipelex-storage://user/assets/1.bin", "filename": "upload.bin"}
+        request = httpx.Request("POST", f"{_BASE_URL}/x")
+        mocker.patch.object(
+            client,
+            "_send",
+            mocker.AsyncMock(
+                side_effect=[
+                    httpx.Response(200, json=build_body, request=request),
+                    httpx.Response(200, json=upload_body, request=request),
+                ]
+            ),
+        )
+
+        prepared = asyncio.run(client.prepare_inputs(files=_FILES, inputs={"photo": bytes([1, 2, 3])}))
+
+        assert prepared.inputs == {"photo": {"url": "pipelex-storage://user/assets/1.bin"}}
+        assert len(prepared.uploads) == 1
diff --git a/tests/unit/test_upload.py b/tests/unit/test_upload.py
new file mode 100644
index 0000000..6c8b227
--- /dev/null
+++ b/tests/unit/test_upload.py
@@ -0,0 +1,160 @@
+"""`upload_file` — the single-asset upload convenience over the raw `upload()` wire call.
+Pins accepted asset forms (path str/Path, bytes), the client-side record assembly
+(uri/filename/content_type/size), base64 correctness, and the mapping of raw transport
+errors onto the semantic preparation errors.
+
+Ports `pipelex-sdk-js/tests/upload.test.ts`. Uses a fake `upload` client double for the
+logic, plus one wiring test through the real `PipelexAPIClient` with a mocked `_send`.
+"""
+
+import asyncio
+import base64
+import json
+from pathlib import Path
+
+import httpx
+import pytest
+from pytest_mock import MockerFixture
+
+from pipelex_sdk.client import PipelexAPIClient
+from pipelex_sdk.errors import (
+    ApiResponseError,
+    ApiUnreachableError,
+    InvalidLocalSourceError,
+    RejectedAssetError,
+    UnsupportedUploadCapabilityError,
+    UploadAuthenticationError,
+    UploadTransportError,
+)
+from pipelex_sdk.product_models import UploadedFile, UploadInput
+from pipelex_sdk.upload import _to_asset_bytes, upload_file
+
+_BASE_URL = "http://localhost:8081"
+
+
+class _FakeUploadClient:
+    """A fake `upload` client: records the wire body, returns a canned URI (or raises)."""
+
+    def __init__(self, *, uri: str = "pipelex-storage://user/assets/abc.bin", error: Exception | None = None) -> None:
+        self.calls: list[UploadInput] = []
+        self._uri = uri
+        self._error = error
+
+    async def upload(self, upload_input: UploadInput) -> UploadedFile:
+        if self._error is not None:
+            raise self._error
+        self.calls.append(upload_input)
+        return UploadedFile(uri=self._uri, filename=upload_input.filename)
+
+
+def _api_error(status: int, server_message: str = "boom") -> ApiResponseError:
+    return ApiResponseError(
+        f"HTTP {status}", api_url=f"{_BASE_URL}/v1/upload", status=status, status_text="Error", response_body="", server_message=server_message
+    )
+
+
+class TestUploadFile:
+    def test_uploads_bytes_with_base64_and_full_record(self) -> None:
+        client = _FakeUploadClient()
+        data = bytes([1, 2, 3, 4, 5])
+
+        record = asyncio.run(upload_file(client, data, filename="blob.png", content_type="image/png"))
+
+        assert len(client.calls) == 1
+        assert client.calls[0].data == base64.b64encode(data).decode("ascii")
+        assert client.calls[0].content_type == "image/png"
+        assert record.uri == "pipelex-storage://user/assets/abc.bin"
+        assert record.filename == "blob.png"
+        assert record.content_type == "image/png"
+        assert record.size == 5
+
+    def test_reads_a_local_path_str_deriving_filename_and_mime(self, tmp_path: Path) -> None:
+        client = _FakeUploadClient()
+        path = tmp_path / "diagram.png"
+        path.write_bytes(bytes([10, 20, 30]))
+
+        record = asyncio.run(upload_file(client, str(path)))
+
+        assert client.calls[0].filename == "diagram.png"
+        assert client.calls[0].content_type == "image/png"
+        assert record.size == 3
+
+    def test_accepts_a_pathlib_path(self, tmp_path: Path) -> None:
+        client = _FakeUploadClient()
+        path = tmp_path / "report.pdf"
+        path.write_bytes(bytes([1, 2]))
+
+        record = asyncio.run(upload_file(client, path))
+
+        assert client.calls[0].filename == "report.pdf"
+        assert client.calls[0].content_type == "application/pdf"
+        assert record.size == 2
+
+    def test_missing_path_raises_invalid_local_source(self, tmp_path: Path) -> None:
+        client = _FakeUploadClient()
+        missing = tmp_path / "nope.png"
+        with pytest.raises(InvalidLocalSourceError):
+            asyncio.run(upload_file(client, missing))
+
+    def test_413_maps_to_rejected_asset(self) -> None:
+        client = _FakeUploadClient(error=_api_error(413, "too big"))
+        with pytest.raises(RejectedAssetError) as exc_info:
+            asyncio.run(upload_file(client, bytes([1]), filename="big.pdf"))
+        assert exc_info.value.filename == "big.pdf"
+        assert exc_info.value.status == 413
+
+    @pytest.mark.parametrize("status", [401, 403])
+    def test_401_403_map_to_upload_authentication(self, status: int) -> None:
+        client = _FakeUploadClient(error=_api_error(status))
+        with pytest.raises(UploadAuthenticationError):
+            asyncio.run(upload_file(client, bytes([1])))
+
+    def test_404_maps_to_unsupported_capability(self) -> None:
+        client = _FakeUploadClient(error=_api_error(404))
+        with pytest.raises(UnsupportedUploadCapabilityError):
+            asyncio.run(upload_file(client, bytes([1])))
+
+    @pytest.mark.parametrize(
+        "error",
+        [
+            _api_error(500),
+            ApiUnreachableError("down", api_url=_BASE_URL, code="ECONNREFUSED"),
+        ],
+    )
+    def test_non_semantic_failures_map_to_transport(self, error: Exception) -> None:
+        client = _FakeUploadClient(error=error)
+        with pytest.raises(UploadTransportError):
+            asyncio.run(upload_file(client, bytes([1])))
+
+    def test_reads_the_local_file_off_the_event_loop(self, mocker: MockerFixture, tmp_path: Path) -> None:
+        # The (possibly large) file read is offloaded via asyncio.to_thread so it never blocks
+        # the event loop. `wraps` keeps the real behavior; we only assert the offload happened.
+        to_thread_spy = mocker.patch("pipelex_sdk.upload.asyncio.to_thread", wraps=asyncio.to_thread)
+        client = _FakeUploadClient()
+        path = tmp_path / "shot.png"
+        path.write_bytes(bytes([1, 2, 3, 4]))
+
+        record = asyncio.run(upload_file(client, path))
+
+        assert to_thread_spy.await_count == 1
+        await_args = to_thread_spy.await_args
+        assert await_args is not None
+        assert await_args.args[0] is _to_asset_bytes
+        assert record.size == 4  # real behavior preserved through the wrapped call
+
+    def test_wires_through_the_real_client(self, mocker: MockerFixture) -> None:
+        client = PipelexAPIClient(api_key="test-token", base_url=_BASE_URL)
+        uploaded = {"uri": "pipelex-storage://user/assets/z.bin", "filename": "x.png"}
+        request = httpx.Request("POST", f"{_BASE_URL}/v1/upload")
+        send = mocker.patch.object(client, "_send", mocker.AsyncMock(return_value=httpx.Response(200, json=uploaded, request=request)))
+
+        record = asyncio.run(client.upload_file(bytes([1, 2, 3]), filename="x.png", content_type="image/png"))
+
+        call = send.call_args
+        assert call.args[0] == "POST"
+        assert call.args[1] == f"{_BASE_URL}/v1/upload"
+        body = json.loads(call.kwargs["content"])
+        assert body["filename"] == "x.png"
+        assert body["data"] == base64.b64encode(bytes([1, 2, 3])).decode("ascii")
+        assert record.uri == "pipelex-storage://user/assets/z.bin"
+        assert record.size == 3