@@ -280,3 +280,73 @@ def test_output_identical_with_and_without_has_finalize(self, client_transport:
         assert len(with_finalize) == 1
         assert len(without_finalize) == 1
         assert with_finalize[0].column("i").to_pylist() == without_finalize[0].column("i").to_pylist()
+
+
+class TestBlendedAny:
+    """`blended_any` / `blended_any_varargs`: positional input columns declared ANY.
+
+    An ANY arg names no concrete Arrow type, so the output type can only come from
+    the input schema the client supplies at bind. These drive that over the wire:
+    the same registration must bind to whatever type each call resolves to, and
+    echo the column back untouched (nested types and nulls included).
+    """
+
+    @pytest.mark.parametrize(
+        "array",
+        [
+            pa.array(["a", None, "c"], type=pa.string()),
+            pa.array([1, 2, None], type=pa.int32()),
+            pa.array(
+                [{"message": "lost pkg", "tier": "gold"}, None, {"message": None, "tier": "free"}],
+                type=pa.struct([("message", pa.string()), ("tier", pa.string())]),
+            ),
+            pa.array([[1, 2], [], None], type=pa.list_(pa.int64())),
+        ],
+        ids=["string", "int32", "struct", "list"],
+    )
+    def test_output_is_typed_from_the_input_and_echoed(self, fixture_worker: str, array: pa.Array[Any]) -> None:
+        """One registration, bound per call to the resolved input type."""
+        batch = pa.record_batch({"value": array})
+        bound: list[pa.Schema] = []
+
+        with Client(fixture_worker) as client:
+            results = list(
+                client.table_in_out_function(
+                    function_name="blended_any",
+                    schema_path=["main"],
+                    input=iter([batch]),
+                    bind_result_callback=lambda r: bound.append(r.output_schema),
+                    has_finalize=False,
+                )
+            )
+
+        assert [s.names for s in bound] == [["value"]]
+        assert bound[0].field("value").type == array.type
+        (out,) = results
+        assert out.column("value").type == array.type
+        # A null struct and a struct with a null field must stay distinct.
+        assert out.column("value").to_pylist() == array.to_pylist()
+
+    def test_varargs_columns_each_keep_their_own_type(self, fixture_worker: str) -> None:
+        """The vararg element type is ANY, so every runtime column types itself."""
+        batch = pa.record_batch(
+            {
+                "col0": pa.array(["a", "b"], type=pa.string()),
+                "col1": pa.array([[1], [2, 3]], type=pa.list_(pa.int32())),
+                "col2": pa.array([{"k": 1.5}, None], type=pa.struct([("k", pa.float64())])),
+            }
+        )
+
+        with Client(fixture_worker) as client:
+            (out,) = list(
+                client.table_in_out_function(
+                    function_name="blended_any_varargs",
+                    schema_path=["main"],
+                    input=iter([batch]),
+                    has_finalize=False,
+                )
+            )
+
+        assert out.schema.names == ["col0", "col1", "col2"]
+        assert out.schema.types == batch.schema.types
+        assert out.to_pylist() == batch.to_pylist()
@@ -148,3 +148,43 @@ class Meta:
 
         with pytest.raises(TypeError, match="at least one positional Arg"):
             resolve_metadata(_NamedOnly)
+
+
+class TestBlendedAnyResolution:
+    """A blended function may declare its positional input column(s) ANY.
+
+    The ANY marker lives on the argument *spec* — the ``vgi_type=any`` field
+    metadata the client reads to learn it must take the column's type from the
+    call rather than from the declaration.
+    """
+
+    def test_any_positional_is_a_blended_input_column(self) -> None:
+        """The ANY arg resolves as a positional input column, not a bind constant."""
+        from vgi._test_fixtures.table_in_out import BlendedAnyFunction
+        from vgi.argument_spec import extract_argument_specs
+
+        m = resolve_metadata(BlendedAnyFunction)
+        assert m.input_from_args is True
+        assert m.has_finalize is False
+        (spec,) = extract_argument_specs(BlendedAnyFunction)
+        assert (spec.name, spec.position, spec.is_any_type, spec.is_varargs) == ("value", 0, True, False)
+
+    def test_any_varargs_is_a_blended_input_column(self) -> None:
+        """The varargs form keeps both markers: ANY element type and varargs."""
+        from vgi._test_fixtures.table_in_out import BlendedAnyVarargsFunction
+        from vgi.argument_spec import extract_argument_specs
+
+        assert resolve_metadata(BlendedAnyVarargsFunction).input_from_args is True
+        (spec,) = extract_argument_specs(BlendedAnyVarargsFunction)
+        assert (spec.name, spec.position, spec.is_any_type, spec.is_varargs) == ("values", 0, True, True)
+
+    def test_any_marker_survives_the_wire_schema(self) -> None:
+        """The client learns the arg is ANY from the serialized arguments schema."""
+        from vgi._test_fixtures.table_in_out import BlendedAnyFunction, BlendedAnyVarargsFunction
+        from vgi.argument_spec import argument_specs_to_schema, extract_argument_specs, schema_to_argument_specs
+
+        for function in (BlendedAnyFunction, BlendedAnyVarargsFunction):
+            specs = extract_argument_specs(function)
+            wire = schema_to_argument_specs(argument_specs_to_schema(specs))
+            assert [s.is_any_type for s in wire] == [True]
+            assert [s.is_varargs for s in wire] == [s.is_varargs for s in specs]
@@ -51,7 +51,7 @@
 from vgi_rpc.log import Level
 from vgi_rpc.rpc import OutputCollector
 
-from vgi.arguments import Arg, Setting, TableInput
+from vgi.arguments import AnyArrow, Arg, Setting, TableInput
 from vgi.cache_control import CacheControl
 from vgi.invocation import BindResponse
 from vgi.metadata import FunctionExample
@@ -94,6 +94,8 @@ class _LogDrainState(ArrowSerializableDataclass):
     "GeoEncode3Function",
     "RowSumFunction",
     "BlendedDropFunction",
+    "BlendedAnyFunction",
+    "BlendedAnyVarargsFunction",
     "SubstreamPartialSumFunction",
     "MultiBatchFinishFunction",
     "SumAllColumnsFunction",
@@ -1880,6 +1882,87 @@ def process(
         cast("VgiOutputCollector", out).emit(out_batch, parent_rows=parent_rows)
 
 
+@dataclass(slots=True, frozen=True, kw_only=True)
+class _AnyArgs:
+    """One positional ANY-typed input column."""
+
+    value: Annotated[AnyArrow, Arg(0, doc="Input column of any type (echoed back unchanged)")]
+
+
+class BlendedAnyFunction(RowTransformFunction[_AnyArgs]):
+    """Blended 1->1 echo of a single ANY-typed input column.
+
+    Proves a blended positional arg may be declared ANY: the declared type names
+    no concrete Arrow type, so the client must build the worker's input schema
+    from the type DuckDB actually resolved for the call (a STRUCT built per row,
+    a whole-row struct, a LIST, a plain VARCHAR ...) rather than from the
+    declaration. The output column ``value`` is bound to that resolved input
+    type at bind time and the column is passed through untouched, so a test
+    asserts on DuckDB's own ``typeof(value)`` and on the round-tripped values —
+    nothing here depends on how an SDK happens to render a type name.
+    """
+
+    class Meta:
+        name = "blended_any"
+        description = "Blended 1->1 echo of one ANY-typed input column (output typed from the input)"
+        categories = ["blended", "test"]
+
+    @classmethod
+    def on_bind(cls, params: BindParams[_AnyArgs]) -> BindResponse:
+        input_schema = params.bind_call.input_schema
+        assert input_schema is not None
+        return BindResponse(output_schema=pa.schema([pa.field("value", input_schema.field(0).type)]))
+
+    @classmethod
+    def process(
+        cls,
+        params: ProcessParams[_AnyArgs],
+        state: None,
+        batch: pa.RecordBatch,
+        out: OutputCollector,
+    ) -> None:
+        out.emit(pa.record_batch([batch.column(0)], schema=params.output_schema))
+
+
+@dataclass(slots=True, frozen=True, kw_only=True)
+class _AnyVarargsArgs:
+    """N positional ANY-typed input columns, each free to be a different type."""
+
+    values: Annotated[list[AnyArrow], Arg(0, varargs=True, doc="Input columns of any types (echoed back)")]
+
+
+class BlendedAnyVarargsFunction(RowTransformFunction[_AnyVarargsArgs]):
+    """Blended 1->1 echo of N ANY-typed VARARGS input columns.
+
+    The varargs counterpart of ``blended_any``: every runtime column may resolve
+    to a different concrete type, so the client must take each column's type
+    from the call rather than from the (ANY) vararg element type. Output columns
+    are ``col0..colN-1`` — the names the client generates for varargs blended
+    input — each bound to its own resolved input type.
+    """
+
+    class Meta:
+        name = "blended_any_varargs"
+        description = "Blended 1->1 echo of N ANY-typed varargs input columns (col0..colN-1)"
+        categories = ["blended", "test"]
+
+    @classmethod
+    def on_bind(cls, params: BindParams[_AnyVarargsArgs]) -> BindResponse:
+        input_schema = params.bind_call.input_schema
+        assert input_schema is not None
+        return BindResponse(output_schema=pa.schema([pa.field(f"col{i}", f.type) for i, f in enumerate(input_schema)]))
+
+    @classmethod
+    def process(
+        cls,
+        params: ProcessParams[_AnyVarargsArgs],
+        state: None,
+        batch: pa.RecordBatch,
+        out: OutputCollector,
+    ) -> None:
+        out.emit(pa.record_batch(cls.input_columns(batch), schema=params.output_schema))
+
+
 class CachedExplodeFunction(RowTransformFunction[_ExplodeArgs]):
     """Cacheable blended 1->N fan-out advertising vgi.cache.per_value.
 
 
@@ -267,6 +267,8 @@
 )
 from vgi._test_fixtures.table_in_out import (
     BatchIndexBufferInputFunction,
+    BlendedAnyFunction,
+    BlendedAnyVarargsFunction,
     BlendedDropFunction,
     BlendedExplodeFunction,
     BufferEmitWideFunction,
@@ -432,6 +434,8 @@ def _build_fixture_stats() -> tuple[
                 GeoEncode3Function,
                 RowSumFunction,
                 BlendedDropFunction,
+                BlendedAnyFunction,
+                BlendedAnyVarargsFunction,
                 BlendedExplodeFunction,
                 CachedExplodeFunction,
                 ProjectableBlendedFunction,