@@ -285,6 +285,53 @@ def test_table_functions_have_correct_type(self, example_worker: str) -> None:
         assert by_name["range"]["function_type"] == "table"
         assert by_name["sum_all_columns"]["function_type"] == "table"
 
+    def test_varargs_function_shows_varargs_in_arguments(
+        self, example_worker: str
+    ) -> None:
+        """Varargs functions show varargs indicator in arguments."""
+        runner = CliRunner()
+
+        # Attach and get contents
+        attach_result = runner.invoke(
+            cli,
+            ["catalog", "attach", "example", "--server", example_worker],
+        )
+        attach_id = json.loads(attach_result.output)["attach_id"]
+
+        contents_result = runner.invoke(
+            cli,
+            [
+                "catalog",
+                "schema",
+                "contents",
+                "main",
+                "--attach-id",
+                attach_id,
+                "--server",
+                example_worker,
+            ],
+        )
+        assert contents_result.exit_code == 0
+
+        # Parse and find sum_columns function
+        lines = contents_result.output.strip().split("\n")
+        items = [json.loads(line) for line in lines if line.strip()]
+        by_name = {item["name"]: item for item in items}
+
+        # Verify sum_columns function exists
+        assert "sum_columns" in by_name, "sum_columns function not found"
+
+        sum_func = by_name["sum_columns"]
+
+        # Verify arguments include varargs indicator
+        arguments = sum_func.get("arguments", [])
+        columns_arg = next(
+            (a for a in arguments if a.get("name") == "columns"),
+            None,
+        )
+        assert columns_arg is not None, "columns argument not found"
+        assert columns_arg.get("varargs") is True, "varargs indicator missing"
+
 
 class TestCLISchemaList:
     """Tests for listing schemas via CLI."""
 
@@ -25,8 +25,9 @@
 from vgi.scalar_function import ScalarFunction
 
 __all__ = [
-    "DoubleColumnFunction",
     "AddNumericColumnsFunction",
+    "DoubleColumnFunction",
+    "SumColumnsFunction",
     "UpperCaseFunction",
 ]
 
@@ -225,3 +226,65 @@ def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
     def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
         """Convert the column values to uppercase."""
         return pc.utf8_upper(batch.column(self.column))  # type: ignore[no-matching-overload]
+
+
+class SumColumnsFunction(ScalarFunction):
+    """Sums values from multiple numeric columns.
+
+    Uses varargs with type_bound to accept any number of numeric columns
+    and validates that all columns are addable types at bind time.
+
+    Example:
+        Input:  a=[1, 2], b=[10, 20], c=[100, 200]
+        Args:   columns=('a', 'b', 'c')
+        Output: result=[111, 222]
+
+    """
+
+    class Meta:
+        """Function metadata."""
+
+        name = "sum_columns"
+        description = "Sum values from multiple numeric columns"
+        examples = [
+            FunctionExample(
+                sql="SELECT sum_columns(price, tax, shipping) FROM orders",
+                description="Calculate total cost from multiple columns",
+            ),
+        ]
+
+    # Varargs with type_bound validates all columns are numeric
+    columns = Arg[AnyArrow](
+        0,
+        varargs=True,
+        type_bound=_is_addable_type,
+        doc="Columns to sum (must be numeric)",
+    )
+
+    _output_type: pa.DataType
+
+    def bind(self) -> None:
+        """Compute output type from first column, promoted for overflow safety."""
+        # With varargs=True, self.columns is a tuple of column names
+        first_col = self.columns[0]  # type: ignore[index]
+        first_type = self.input_schema.field(first_col).type
+        self._output_type = _promote_for_addition(first_type)
+
+    @classmethod
+    def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
+        """Output type depends on input column types."""
+        return AnyArrow
+
+    @property
+    def output_type(self) -> pa.DataType:
+        """Return the computed output type based on first column."""
+        return self._output_type
+
+    def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
+        """Sum values from all specified columns."""
+        # With varargs=True, self.columns is a tuple of column names at runtime
+        columns: tuple[str, ...] = self.columns  # type: ignore[assignment]
+        result = batch.column(columns[0])
+        for col_name in columns[1:]:
+            result = pc.add(result, batch.column(col_name))
+        return result
@@ -16,6 +16,7 @@
 from vgi.examples.scalar import (
     AddNumericColumnsFunction,
     DoubleColumnFunction,
+    SumColumnsFunction,
     UpperCaseFunction,
 )
 from vgi.examples.table import (
@@ -74,8 +75,9 @@ class ExampleWorker(Worker):
         ProjectedDataFunction,
         SettingsAwareFunction,
         # ScalarFunctionGenerator - transform to single-column output
-        DoubleColumnFunction,
         AddNumericColumnsFunction,
+        DoubleColumnFunction,
+        SumColumnsFunction,
         UpperCaseFunction,
     ]
 
 
@@ -235,7 +235,10 @@ def __init__(
     ):
         """Initialize the scalar function with invocation data and logger."""
         super().__init__(invocation=invocation, logger=logger)
-        if invocation.input_schema is None:
+
+    def _validate_input_schema_requirement(self) -> None:
+        """Validate that input_schema is provided for scalar functions."""
+        if self.invocation.input_schema is None:
             raise ValueError(
                 f"{type(self).__name__} requires an input schema, but none was "
                 f"provided. ScalarFunction processes input batches and requires "
 
@@ -509,7 +509,10 @@ def __init__(
     ):
         """Initialize the function with invocation data and logger."""
         super().__init__(invocation=invocation, logger=logger)
-        if invocation.input_schema is None:
+
+    def _validate_input_schema_requirement(self) -> None:
+        """Validate that input_schema is provided for table-in-out functions."""
+        if self.invocation.input_schema is None:
             raise ValueError(
                 f"{type(self).__name__} requires an input schema, but none was "
                 f"provided. TableInOutGenerator processes input batches and "