bruin-data
diff --git a/‎src/bruin/_context.py‎
Lines changed: 62 additions & 0 deletions b/‎src/bruin/_context.py‎
Lines changed: 62 additions & 0 deletions
diff --git a/‎tests/e2e/test_bruin_run.py‎
Lines changed: 96 additions & 0 deletions b/‎tests/e2e/test_bruin_run.py‎
Lines changed: 96 additions & 0 deletions
@@ -32,6 +32,58 @@ def _parse_datetime(env_var: str) -> "datetime.datetime | None":
         )
 
 
+def _coerce_value(value, type_def: dict):
+    """Coerce a single value to match a JSON Schema type definition."""
+    if value is None:
+        return None
+    schema_type = type_def.get("type")
+    if schema_type == "string":
+        return str(value)
+    if schema_type == "integer":
+        return int(value)
+    if schema_type == "number":
+        return float(value)
+    if schema_type == "boolean":
+        if isinstance(value, str):
+            if value.lower() in ("true", "1"):
+                return True
+            if value.lower() in ("false", "0"):
+                return False
+            raise ValueError(f"Cannot convert '{value}' to boolean")
+        return bool(value)
+    if schema_type == "array":
+        if not isinstance(value, list):
+            return value
+        items_def = type_def.get("items")
+        if items_def:
+            return [_coerce_value(item, items_def) for item in value]
+        return value
+    if schema_type == "object":
+        if not isinstance(value, dict):
+            return value
+        props = type_def.get("properties", {})
+        return {k: _coerce_value(v, props[k]) if k in props else v for k, v in value.items()}
+    return value  # unknown type → passthrough
+
+
+def _coerce_vars(values: dict, schema: dict) -> dict:
+    """Apply schema-based type coercion to all variables."""
+    result = {}
+    for key, val in values.items():
+        type_def = schema.get(key)
+        if type_def:
+            try:
+                result[key] = _coerce_value(val, type_def)
+            except (ValueError, TypeError) as exc:
+                target = type_def.get("type")
+                raise ValueError(
+                    f"Cannot coerce variable '{key}' (value={val!r}) to {target}: {exc}"
+                ) from exc
+        else:
+            result[key] = val
+    return result
+
+
 class _BruinContext:
     """Lazy accessor for BRUIN_* environment variables injected by ``bruin run``.
 
@@ -92,6 +144,16 @@ def vars(self) -> dict:
             raise BruinError(
                 f"Invalid BRUIN_VARS value: expected a JSON object, got {type(parsed).__name__}."
             )
+        schema_raw = os.environ.get("BRUIN_VARS_SCHEMA")
+        if schema_raw:
+            try:
+                schema = json.loads(schema_raw)
+            except (json.JSONDecodeError, TypeError):
+                return parsed  # bad schema → return raw values
+            try:
+                return _coerce_vars(parsed, schema)
+            except (ValueError, TypeError) as exc:
+                raise BruinError(f"Cannot coerce BRUIN_VARS: {exc}") from exc
         return parsed
 
 
 
@@ -136,6 +136,102 @@ def test_context_available_in_asset(self, bruin_bin, pipeline_dir):
         assert output["has_bruin_this"] is True
 
 
+class TestVarsCoercion:
+    """Verify that BRUIN_VARS_SCHEMA is injected and the SDK coerces types."""
+
+    def test_vars_coerced_to_schema_types(self, bruin_bin, pipeline_dir):
+        root, pipe_dir, assets_dir = pipeline_dir
+
+        # Pipeline with typed variables
+        (pipe_dir / "pipeline.yml").write_text(
+            textwrap.dedent("""\
+            name: test_pipeline
+            schedule: daily
+            variables:
+              env:
+                type: string
+                default: dev
+              count:
+                type: integer
+                default: 10
+              rate:
+                type: number
+                default: 1.5
+              debug:
+                type: boolean
+                default: false
+        """)
+        )
+
+        output_file = root / "vars_output.json"
+        asset_code = textwrap.dedent(f'''\
+            """ @bruin
+
+            name: test_vars
+            type: python
+
+            @bruin """
+
+            import json
+            import os
+            from bruin import context
+
+            v = context.vars
+            result = {{
+                "vars": v,
+                "types": {{k: type(val).__name__ for k, val in v.items()}},
+                "has_schema": "BRUIN_VARS_SCHEMA" in os.environ,
+            }}
+
+            with open("{output_file}", "w") as f:
+                json.dump(result, f)
+        ''')
+        (assets_dir / "test_vars.py").write_text(asset_code)
+
+        result = subprocess.run(
+            [
+                bruin_bin,
+                "run",
+                "--start-date",
+                "2024-06-01",
+                "--end-date",
+                "2024-06-02",
+                "--var",
+                "count=42",
+                "--var",
+                "debug=true",
+                str(pipe_dir / "assets" / "test_vars.py"),
+            ],
+            capture_output=True,
+            text=True,
+            cwd=str(root),
+            timeout=120,
+        )
+
+        assert result.returncode == 0, (
+            f"bruin run failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
+        )
+        assert output_file.exists(), (
+            f"Asset did not produce output file.\nstdout: {result.stdout}\nstderr: {result.stderr}"
+        )
+
+        output = json.loads(output_file.read_text())
+
+        if not output["has_schema"]:
+            pytest.skip("bruin binary does not inject BRUIN_VARS_SCHEMA yet")
+
+        # CLI overrides were strings — SDK should have coerced them
+        assert output["vars"]["count"] == 42
+        assert output["types"]["count"] == "int"
+        assert output["vars"]["debug"] is True
+        assert output["types"]["debug"] == "bool"
+        # Defaults should keep their types
+        assert output["vars"]["env"] == "dev"
+        assert output["types"]["env"] == "str"
+        assert output["vars"]["rate"] == 1.5
+        assert output["types"]["rate"] == "float"
+
+
 class TestConnectionEnvVar:
     """Verify BRUIN_CONNECTION is injected when asset has a connection field.