diff --git a/changelog/+artifact-composition.added.md b/changelog/+artifact-composition.added.md
new file mode 100644
index 00000000..5891fa4d
--- /dev/null
+++ b/changelog/+artifact-composition.added.md
@@ -0,0 +1 @@
+Add `artifact_content`, `file_object_content`, `from_json`, and `from_yaml` Jinja2 filters for artifact content composition in templates.
diff --git a/changelog/+artifact-composition.changed.md b/changelog/+artifact-composition.changed.md
new file mode 100644
index 00000000..4a9912c0
--- /dev/null
+++ b/changelog/+artifact-composition.changed.md
@@ -0,0 +1 @@
+Replace `FilterDefinition.trusted: bool` with flag-based `ExecutionContext` model (`CORE`, `WORKER`, `LOCAL`) for context-aware template validation. `validate()` now accepts an optional `context` parameter. Backward compatible.
diff --git a/dev/specs/infp-504-artifact-composition/contracts/filter-interfaces.md b/dev/specs/infp-504-artifact-composition/contracts/filter-interfaces.md
index e034bbba..7dc7874e 100644
--- a/dev/specs/infp-504-artifact-composition/contracts/filter-interfaces.md
+++ b/dev/specs/infp-504-artifact-composition/contracts/filter-interfaces.md
@@ -19,7 +19,7 @@ async def artifact_content(storage_id: str) -> str
 | Permission denied (401/403) | — | `JinjaFilterError("artifact_content", "permission denied for storage_id: {id}")` |
 | No client provided | — | `JinjaFilterError("artifact_content", "requires InfrahubClient", hint="pass client via Jinja2Template(client=...)")` |
 
-**Validation**: Blocked in `CORE` context. Allowed in `WORKER` context.
+**Validation**: Blocked in `CORE` context. Allowed in `WORKER` and `LOCAL` contexts.
 
 ### file_object_content
 
@@ -37,7 +37,7 @@ async def file_object_content(storage_id: str) -> str
 | Permission denied (401/403) | — | `JinjaFilterError("file_object_content", "permission denied for storage_id: {id}")` |
 | No client provided | — | `JinjaFilterError("file_object_content", "requires InfrahubClient", hint="pass client via Jinja2Template(client=...)")` |
 
-**Validation**: Blocked in `CORE` context. Allowed in `WORKER` context.
+**Validation**: Blocked in `CORE` context. Allowed in `WORKER` and `LOCAL` contexts.
 
 ### from_json
 
diff --git a/dev/specs/infp-504-artifact-composition/data-model.md b/dev/specs/infp-504-artifact-composition/data-model.md
index ad1e0d29..cc9bd419 100644
--- a/dev/specs/infp-504-artifact-composition/data-model.md
+++ b/dev/specs/infp-504-artifact-composition/data-model.md
@@ -104,7 +104,19 @@ def __init__(
 - New optional `client` parameter
 - When `client` provided: instantiate `InfrahubFilters`, register `artifact_content` and `file_object_content`
 - Always register `from_json` and `from_yaml` (no client needed)
-- File-based environment must add `enable_async=True` for async filter support
+- File-based environment already has `enable_async=True` (no change needed)
+
+### Jinja2Template.set_client() (new method)
+
+```python
+def set_client(self, client: InfrahubClient) -> None:
+```
+
+**Purpose**: Deferred client injection — allows creating a `Jinja2Template` first and adding the client later. Also supports replacing a previously set client.
+
+- Calls `_register_client_filters(client)` to bind real filter methods
+- If the Jinja2 environment was already created, patches it in place
+- Without calling `set_client()` (and without passing `client` to `__init__`), client-dependent filters raise `JinjaFilterError` with a descriptive message at render time
 
 ### Jinja2Template.validate() (modified signature)
 
@@ -140,9 +152,9 @@ async def get_file_by_storage_id(self, storage_id: str, tracker: str | None = No
 ```python
 # In AVAILABLE_FILTERS:
 
-# Infrahub client-dependent filters (worker context only)
-FilterDefinition("artifact_content", allowed_contexts=ExecutionContext.WORKER, source="infrahub"),
-FilterDefinition("file_object_content", allowed_contexts=ExecutionContext.WORKER, source="infrahub"),
+# Infrahub client-dependent filters (worker and local contexts)
+FilterDefinition("artifact_content", allowed_contexts=ExecutionContext.WORKER | ExecutionContext.LOCAL, source="infrahub"),
+FilterDefinition("file_object_content", allowed_contexts=ExecutionContext.WORKER | ExecutionContext.LOCAL, source="infrahub"),
 
 # Parsing filters (trusted, all contexts)
 FilterDefinition("from_json", allowed_contexts=ExecutionContext.ALL, source="infrahub"),
diff --git a/dev/specs/infp-504-artifact-composition/spec.md b/dev/specs/infp-504-artifact-composition/spec.md
index fa39a004..75d484a9 100644
--- a/dev/specs/infp-504-artifact-composition/spec.md
+++ b/dev/specs/infp-504-artifact-composition/spec.md
@@ -93,13 +93,13 @@ The existing single `restricted: bool` parameter on `validate()` is insufficient
 
 ### Functional requirements
 
-- **FR-001**: `Jinja2Template.__init__` MUST accept an optional `client` parameter of type `InfrahubClient | None` (default `None`). `InfrahubClientSync` is not supported.
-- **FR-002**: A dedicated class (for example, `InfrahubFilters`) MUST be introduced to hold the client reference and expose the Infrahub-specific filter callable methods. `Jinja2Template` instantiates this class when a client is provided and registers its filters into the Jinja2 environment.
+- **FR-001**: `Jinja2Template.__init__` MUST accept an optional `client` parameter of type `InfrahubClient | None` (default `None`). Additionally, `Jinja2Template` MUST expose a `set_client(client)` method for deferred client injection, allowing the template to be created first and the client added later. `InfrahubClientSync` is not supported.
+- **FR-002**: A dedicated class (for example, `InfrahubFilters`) MUST be introduced to hold the client reference and expose the Infrahub-specific filter callable methods. `Jinja2Template` instantiates this class when a client is provided (via `__init__` or `set_client()`) and registers its filters into the Jinja2 environment.
 - **FR-003**: The system MUST provide an `artifact_content` Jinja2 filter that accepts a `storage_id` string and returns the raw string content of the referenced artifact, using the artifact-specific API path.
 - **FR-004**: The system MUST provide a `file_object_content` Jinja2 filter that accepts a `storage_id` string and returns the raw string content of the referenced file object, using the file-object-specific API path or metadata handling — this implementation is distinct from `artifact_content`.
 - **FR-005**: Both `artifact_content` and `file_object_content` MUST raise `JinjaFilterError` when the input `storage_id` is null or empty, or when the object store cannot retrieve the content for any reason (not found, network failure, auth failure). Additionally, `file_object_content` MUST raise `JinjaFilterError` when the retrieved content has a non-text content type (i.e., not `text/*`, `application/json`, or `application/yaml`).
 - **FR-006**: Both `artifact_content` and `file_object_content` MUST raise `JinjaFilterError` when invoked and no `InfrahubClient` was supplied to `Jinja2Template` at construction time. The error message MUST name the filter and explain that an `InfrahubClient` is required.
-- **FR-007**: Both `artifact_content` and `file_object_content` MUST be registered with `allowed_contexts=ExecutionContext.WORKER` in the `FilterDefinition` registry. The `validate()` method accepts an `ExecutionContext` flag; these filters are blocked in the `CORE` context (API server computed attributes) and permitted in the `WORKER` context (Prefect workers, where an `InfrahubClient` is available). Within Infrahub, any Jinja2-based computed attributes that use these new filters should cause a schema violation when loading the schema.
+- **FR-007**: Both `artifact_content` and `file_object_content` MUST be registered with `allowed_contexts=ExecutionContext.WORKER | ExecutionContext.LOCAL` in the `FilterDefinition` registry. The `validate()` method accepts an `ExecutionContext` flag; these filters are blocked in the `CORE` context (API server computed attributes) and permitted in the `WORKER` context (Prefect workers) and `LOCAL` context (CLI/unrestricted rendering). Within Infrahub, any Jinja2-based computed attributes that use these new filters should cause a schema violation when loading the schema.
 - **FR-008**: The system MUST provide `from_json` and `from_yaml` Jinja2 filters (adding them only if not already present in the environment) that parse a string into a Python dict/list. Applying them to an empty string MUST return an empty dict without raising. Applying them to malformed content MUST raise `JinjaFilterError`.
 - **FR-009**: `from_json` and `from_yaml` MUST be registered as trusted filters (`trusted=True`) since they perform no external I/O.
 - **FR-010**: All new filters MUST work correctly with `InfrahubClient` (async). `InfrahubClientSync` is not a supported client type for `Jinja2Template`. Both the sandboxed environment (string-based templates) and the file-based environment MUST have `enable_async=True` to support async filter callables via Jinja2's `auto_await`.
diff --git a/dev/specs/infp-504-artifact-composition/tasks.md b/dev/specs/infp-504-artifact-composition/tasks.md
index 47e372bb..ca708923 100644
--- a/dev/specs/infp-504-artifact-composition/tasks.md
+++ b/dev/specs/infp-504-artifact-composition/tasks.md
@@ -22,8 +22,8 @@
 
 **CRITICAL**: No user story work can begin until this phase is complete
 
-- [ ] T001 [P] Create `JinjaFilterError` exception class in `infrahub_sdk/template/exceptions.py` — subclass of `JinjaTemplateError` with `filter_name`, `message`, and optional `hint` attributes. Include unit tests for instantiation, inheritance chain, and message formatting. (IFC-2367)
-- [ ] T002 [P] Implement `ExecutionContext` flag enum and migrate `FilterDefinition` in `infrahub_sdk/template/filters.py` — add `ExecutionContext(Flag)` with `CORE`, `WORKER`, `LOCAL`, `ALL` values. Replace `FilterDefinition.trusted: bool` with `allowed_contexts: ExecutionContext`. Add backward-compat `trusted` property. Migrate all 138 existing filter entries (`trusted=True` → `ALL`, `trusted=False` → `LOCAL`). Update `validate()` in `infrahub_sdk/template/__init__.py` to accept optional `context: ExecutionContext` parameter (takes precedence over `restricted`; `restricted=True` → `CORE`, `restricted=False` → `LOCAL`). Include unit tests for all 3 contexts with existing filters, backward compat path, and no regressions. (IFC-2368)
+- [x] T001 [P] Create `JinjaFilterError` exception class in `infrahub_sdk/template/exceptions.py` — subclass of `JinjaTemplateError` with `filter_name`, `message`, and optional `hint` attributes. Include unit tests for instantiation, inheritance chain, and message formatting. (IFC-2367)
+- [x] T002 [P] Implement `ExecutionContext` flag enum and migrate `FilterDefinition` in `infrahub_sdk/template/filters.py` — add `ExecutionContext(Flag)` with `CORE`, `WORKER`, `LOCAL`, `ALL` values. Replace `FilterDefinition.trusted: bool` with `allowed_contexts: ExecutionContext`. Add backward-compat `trusted` property. Migrate all 138 existing filter entries (`trusted=True` → `ALL`, `trusted=False` → `LOCAL`). Update `validate()` in `infrahub_sdk/template/__init__.py` to accept optional `context: ExecutionContext` parameter (takes precedence over `restricted`; `restricted=True` → `CORE`, `restricted=False` → `LOCAL`). Include unit tests for all 3 contexts with existing filters, backward compat path, and no regressions. (IFC-2368)
 
 **Checkpoint**: Foundation ready — JinjaFilterError and ExecutionContext available for all stories
 
@@ -37,9 +37,9 @@
 
 ### Implementation for US1 + US4
 
-- [ ] T003 [US1] Create `InfrahubFilters` class in `infrahub_sdk/template/infrahub_filters.py` — new file. Class holds `InfrahubClient` reference, exposes async filter methods. Methods are `async def` (Jinja2 `auto_await` handles them in async render mode per R-001). Raises `JinjaFilterError` when called without a client. Include unit tests for instantiation with/without client. (IFC-2371)
-- [ ] T004 [US1] Implement `artifact_content` async method on `InfrahubFilters` in `infrahub_sdk/template/infrahub_filters.py` — uses `self.client.object_store.get(identifier=storage_id)`. Raises `JinjaFilterError` on: null/empty storage_id, retrieval failure, permission denied (catch `AuthenticationError` per R-006). Artifacts are always text (no binary check needed per R-003). Include unit tests: happy path (mocked ObjectStore), null, empty, not-found, network error, permission denied, no-client error with descriptive message. (IFC-2372)
-- [ ] T005 [US1] [US4] Add `client` parameter to `Jinja2Template.__init__` and wire up filter registration in `infrahub_sdk/template/__init__.py` — add `client: InfrahubClient | None = None` param. When client provided: instantiate `InfrahubFilters`, register `artifact_content` into Jinja2 env filter map. Add `enable_async=True` to `_get_file_based_environment()` (per R-001 caveat). Register `artifact_content` in `FilterDefinition` registry with `allowed_contexts=ExecutionContext.WORKER`. Include unit tests: render with client (mocked), render without client (error), validation in CORE (blocked), WORKER (allowed), LOCAL (allowed). Verify existing untrusted filters like `safe` remain blocked in WORKER context (US4 AC3). (IFC-2375 partial + IFC-2376 partial)
+- [x] T003 [US1] Create `InfrahubFilters` class in `infrahub_sdk/template/infrahub_filters.py` — new file. Class holds `InfrahubClient` reference, exposes async filter methods. Methods are `async def` (Jinja2 `auto_await` handles them in async render mode per R-001). Raises `JinjaFilterError` when called without a client. Include unit tests for instantiation with/without client. (IFC-2371)
+- [x] T004 [US1] Implement `artifact_content` async method on `InfrahubFilters` in `infrahub_sdk/template/infrahub_filters.py` — uses `self.client.object_store.get(identifier=storage_id)`. Raises `JinjaFilterError` on: null/empty storage_id, retrieval failure, permission denied (catch `AuthenticationError` per R-006). Artifacts are always text (no binary check needed per R-003). Include unit tests: happy path (mocked ObjectStore), null, empty, not-found, network error, permission denied, no-client error with descriptive message. (IFC-2372)
+- [x] T005 [US1] [US4] Add `client` parameter to `Jinja2Template.__init__` and wire up filter registration in `infrahub_sdk/template/__init__.py` — add `client: InfrahubClient | None = None` param. When client provided: instantiate `InfrahubFilters`, register `artifact_content` into Jinja2 env filter map. Add `enable_async=True` to `_get_file_based_environment()` (per R-001 caveat). Register `artifact_content` in `FilterDefinition` registry with `allowed_contexts=ExecutionContext.WORKER`. Include unit tests: render with client (mocked), render without client (error), validation in CORE (blocked), WORKER (allowed), LOCAL (allowed). Verify existing untrusted filters like `safe` remain blocked in WORKER context (US4 AC3). Also added `set_client()` setter for deferred client injection per PR #885 feedback. (IFC-2375 partial + IFC-2376 partial)
 
 **Checkpoint**: US1 + US4 fully functional. `artifact_content` renders in WORKER context, blocked in CORE. MVP complete.
 
@@ -53,9 +53,9 @@
 
 ### Implementation for US2
 
-- [ ] T006 [P] [US2] Add `get_file_by_storage_id()` method to `ObjectStore` in `infrahub_sdk/object_store.py` — async method using endpoint `GET /api/files/by-storage-id/{storage_id}`. Check `content-type` response header: allow `text/*`, `application/json`, `application/yaml`, `application/x-yaml`; reject all others with descriptive error. Handle 401/403 as `AuthenticationError`. Include unit tests: text response, binary rejection, 404, auth failure, network error. (IFC-2373)
-- [ ] T007 [US2] Implement `file_object_content` async method on `InfrahubFilters` in `infrahub_sdk/template/infrahub_filters.py` — uses new `self.client.object_store.get_file_by_storage_id(storage_id)`. Same error handling as `artifact_content` plus binary content error (delegated to ObjectStore). Include unit tests: happy path, all error conditions, binary content rejection. (IFC-2374)
-- [ ] T008 [US2] Register `file_object_content` filter in `Jinja2Template` and `FilterDefinition` in `infrahub_sdk/template/__init__.py` and `infrahub_sdk/template/filters.py` — register when client provided. `allowed_contexts=ExecutionContext.WORKER`. Include unit tests: render with client, validation in CORE (blocked), WORKER (allowed). (IFC-2375 partial + IFC-2376 partial)
+- [x] T006 [P] [US2] Add `get_file_by_storage_id()` method to `ObjectStore` in `infrahub_sdk/object_store.py` — async method using endpoint `GET /api/files/by-storage-id/{storage_id}`. Check `content-type` response header: allow `text/*`, `application/json`, `application/yaml`, `application/x-yaml`; reject all others with descriptive error. Handle 401/403 as `AuthenticationError`. Include unit tests: text response, binary rejection, 404, auth failure, network error. (IFC-2373)
+- [x] T007 [US2] Implement `file_object_content` async method on `InfrahubFilters` in `infrahub_sdk/template/infrahub_filters.py` — uses new `self.client.object_store.get_file_by_storage_id(storage_id)`. Same error handling as `artifact_content` plus binary content error (delegated to ObjectStore). Include unit tests: happy path, all error conditions, binary content rejection. (IFC-2374)
+- [x] T008 [US2] Register `file_object_content` filter in `Jinja2Template` and `FilterDefinition` in `infrahub_sdk/template/__init__.py` and `infrahub_sdk/template/filters.py` — register when client provided. `allowed_contexts=ExecutionContext.WORKER`. Include unit tests: render with client, validation in CORE (blocked), WORKER (allowed). (IFC-2375 partial + IFC-2376 partial)
 
 **Checkpoint**: US2 complete. `file_object_content` works alongside `artifact_content`.
 
@@ -69,9 +69,9 @@
 
 ### Implementation for US3
 
-- [ ] T009 [P] [US3] Implement `from_json` filter function in `infrahub_sdk/template/infrahub_filters.py` — pure sync function (no client needed). Empty string → `{}` (explicit special-case since `json.loads("")` raises). Malformed JSON → `JinjaFilterError`. Register in `FilterDefinition` with `allowed_contexts=ExecutionContext.ALL`. Register unconditionally in `Jinja2Template._set_filters()`. Include unit tests: valid JSON dict, valid JSON list, empty string → `{}`, malformed → error, render through template. (IFC-2369)
-- [ ] T010 [P] [US3] Implement `from_yaml` filter function in `infrahub_sdk/template/infrahub_filters.py` — pure sync function. Empty string → `{}` (explicit special-case since `yaml.safe_load("")` returns `None`). Malformed YAML → `JinjaFilterError`. Register in `FilterDefinition` with `allowed_contexts=ExecutionContext.ALL`. Register unconditionally in `Jinja2Template._set_filters()`. Include unit tests: valid YAML mapping, valid YAML list, empty string → `{}`, malformed → error, render through template. (IFC-2370)
-- [ ] T011 [US3] Integration test for filter chaining in `tests/unit/template/test_infrahub_filters.py` — test `artifact_content | from_json` and `artifact_content | from_yaml` end-to-end with mocked ObjectStore returning JSON/YAML content. Verify template can access parsed fields. (IFC-2376 partial, SC-006)
+- [x] T009 [P] [US3] Implement `from_json` filter function in `infrahub_sdk/template/infrahub_filters.py` — pure sync function (no client needed). Empty string → `{}` (explicit special-case since `json.loads("")` raises). Malformed JSON → `JinjaFilterError`. Register in `FilterDefinition` with `allowed_contexts=ExecutionContext.ALL`. Register unconditionally in `Jinja2Template._set_filters()`. Include unit tests: valid JSON dict, valid JSON list, empty string → `{}`, malformed → error, render through template. (IFC-2369)
+- [x] T010 [P] [US3] Implement `from_yaml` filter function in `infrahub_sdk/template/infrahub_filters.py` — pure sync function. Empty string → `{}` (explicit special-case since `yaml.safe_load("")` returns `None`). Malformed YAML → `JinjaFilterError`. Register in `FilterDefinition` with `allowed_contexts=ExecutionContext.ALL`. Register unconditionally in `Jinja2Template._set_filters()`. Include unit tests: valid YAML mapping, valid YAML list, empty string → `{}`, malformed → error, render through template. (IFC-2370)
+- [x] T011 [US3] Integration test for filter chaining in `tests/unit/template/test_infrahub_filters.py` — test `artifact_content | from_json` and `artifact_content | from_yaml` end-to-end with mocked ObjectStore returning JSON/YAML content. Verify template can access parsed fields. (IFC-2376 partial, SC-006)
 
 **Checkpoint**: US3 complete. All 4 filters work, chain correctly, and are validated per context.
 
diff --git a/infrahub_sdk/object_store.py b/infrahub_sdk/object_store.py
index bf5fc862..3305fffe 100644
--- a/infrahub_sdk/object_store.py
+++ b/infrahub_sdk/object_store.py
@@ -11,6 +11,14 @@
     from .client import InfrahubClient, InfrahubClientSync
 
 
+ALLOWED_TEXT_CONTENT_TYPES = {"application/json", "application/yaml", "application/x-yaml"}
+
+
+def _extract_content_type(response: httpx.Response) -> str:
+    """Extract and normalize the content-type from an HTTP response, stripping parameters."""
+    return response.headers.get("content-type", "").split(";")[0].strip().lower()
+
+
 class ObjectStoreBase:
     pass
 
@@ -62,6 +70,38 @@ async def upload(self, content: str, tracker: str | None = None) -> dict[str, st
 
         return resp.json()
 
+    async def get_file_by_storage_id(self, storage_id: str, tracker: str | None = None) -> str:
+        """Retrieve file object content by storage_id.
+
+        Raises an error if the content-type indicates binary content.
+        """
+        url = f"{self.client.address}/api/files/by-storage-id/{storage_id}"
+        headers = copy.copy(self.client.headers or {})
+        if self.client.insert_tracker and tracker:
+            headers["X-Infrahub-Tracker"] = tracker
+
+        try:
+            resp = await self.client._get(url=url, headers=headers)
+            resp.raise_for_status()
+        except ServerNotReachableError:
+            self.client.log.error(f"Unable to connect to {self.client.address} .. ")
+            raise
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code in {401, 403}:
+                response = exc.response.json()
+                errors = response.get("errors")
+                messages = [error.get("message") for error in errors]
+                raise AuthenticationError(" | ".join(messages)) from exc
+            raise
+
+        content_type = _extract_content_type(resp)
+        if not content_type.startswith("text/") and content_type not in ALLOWED_TEXT_CONTENT_TYPES:
+            raise ValueError(
+                f"Binary content not supported: content-type '{content_type}' for storage_id '{storage_id}'"
+            )
+
+        return resp.text
+
 
 class ObjectStoreSync(ObjectStoreBase):
     def __init__(self, client: InfrahubClientSync) -> None:
@@ -109,3 +149,35 @@ def upload(self, content: str, tracker: str | None = None) -> dict[str, str]:
                 raise AuthenticationError(" | ".join(messages)) from exc
 
         return resp.json()
+
+    def get_file_by_storage_id(self, storage_id: str, tracker: str | None = None) -> str:
+        """Retrieve file object content by storage_id.
+
+        Raises an error if the content-type indicates binary content.
+        """
+        url = f"{self.client.address}/api/files/by-storage-id/{storage_id}"
+        headers = copy.copy(self.client.headers or {})
+        if self.client.insert_tracker and tracker:
+            headers["X-Infrahub-Tracker"] = tracker
+
+        try:
+            resp = self.client._get(url=url, headers=headers)
+            resp.raise_for_status()
+        except ServerNotReachableError:
+            self.client.log.error(f"Unable to connect to {self.client.address} .. ")
+            raise
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code in {401, 403}:
+                response = exc.response.json()
+                errors = response.get("errors")
+                messages = [error.get("message") for error in errors]
+                raise AuthenticationError(" | ".join(messages)) from exc
+            raise
+
+        content_type = _extract_content_type(resp)
+        if not content_type.startswith("text/") and content_type not in ALLOWED_TEXT_CONTENT_TYPES:
+            raise ValueError(
+                f"Binary content not supported: content-type '{content_type}' for storage_id '{storage_id}'"
+            )
+
+        return resp.text
diff --git a/infrahub_sdk/template/__init__.py b/infrahub_sdk/template/__init__.py
index 6a7f7fe2..29394a2f 100644
--- a/infrahub_sdk/template/__init__.py
+++ b/infrahub_sdk/template/__init__.py
@@ -3,7 +3,7 @@
 import linecache
 from collections.abc import Callable
 from pathlib import Path
-from typing import Any, NoReturn
+from typing import TYPE_CHECKING, Any, NoReturn
 
 import jinja2
 from jinja2 import meta, nodes
@@ -19,9 +19,13 @@
     JinjaTemplateSyntaxError,
     JinjaTemplateUndefinedError,
 )
-from .filters import AVAILABLE_FILTERS
+from .filters import AVAILABLE_FILTERS, ExecutionContext
+from .infrahub_filters import InfrahubFilters, from_json, from_yaml, no_client_filter
 from .models import UndefinedJinja2Error
 
+if TYPE_CHECKING:
+    from infrahub_sdk.client import InfrahubClient
+
 netutils_filters = jinja2_convenience_function()
 
 
@@ -31,6 +35,7 @@ def __init__(
         template: str | Path,
         template_directory: Path | None = None,
         filters: dict[str, Callable] | None = None,
+        client: InfrahubClient | None = None,
     ) -> None:
         self.is_string_based = isinstance(template, str)
         self.is_file_based = isinstance(template, Path)
@@ -39,17 +44,42 @@ def __init__(
         self._environment: jinja2.Environment | None = None
 
         self._available_filters = [filter_definition.name for filter_definition in AVAILABLE_FILTERS]
-        self._trusted_filters = [
-            filter_definition.name for filter_definition in AVAILABLE_FILTERS if filter_definition.trusted
-        ]
 
         self._filters = filters or {}
+        self._user_filter_names: set[str] = set(self._filters.keys())
         for user_filter in self._filters:
             self._available_filters.append(user_filter)
-            self._trusted_filters.append(user_filter)
+
+        self._infrahub_filters: InfrahubFilters | None = None
+        self._register_client_filters(client=client)
+        self._register_filter("from_json", from_json)
+        self._register_filter("from_yaml", from_yaml)
 
         self._template_definition: jinja2.Template | None = None
 
+    def set_client(self, client: InfrahubClient) -> None:
+        """Set or replace the InfrahubClient used by client-dependent filters."""
+        self._register_client_filters(client=client)
+        if self._environment:
+            self._environment.filters["artifact_content"] = self._filters["artifact_content"]
+            self._environment.filters["file_object_content"] = self._filters["file_object_content"]
+
+    def _register_filter(self, name: str, func: Callable) -> None:
+        """Register a filter callable and make it available for validation."""
+        self._filters[name] = func
+        if name not in self._available_filters:
+            self._available_filters.append(name)
+
+    def _register_client_filters(self, client: InfrahubClient | None) -> None:
+        """Register client-dependent filters, using fallbacks if no client is provided."""
+        if client is not None:
+            self._infrahub_filters = InfrahubFilters(client=client)
+            self._register_filter("artifact_content", self._infrahub_filters.artifact_content)
+            self._register_filter("file_object_content", self._infrahub_filters.file_object_content)
+        else:
+            self._register_filter("artifact_content", no_client_filter("artifact_content"))
+            self._register_filter("file_object_content", no_client_filter("file_object_content"))
+
     def get_environment(self) -> jinja2.Environment:
         if self._environment:
             return self._environment
@@ -86,10 +116,16 @@ def get_variables(self) -> list[str]:
 
         return sorted(meta.find_undeclared_variables(template))
 
-    def validate(self, restricted: bool = True) -> None:
-        allowed_list = self._available_filters
-        if restricted:
-            allowed_list = self._trusted_filters
+    def validate(self, restricted: bool = True, context: ExecutionContext | None = None) -> None:
+        effective_context = (
+            context if context is not None else ExecutionContext.CORE if restricted else ExecutionContext.LOCAL
+        )
+
+        allowed_list = [fd.name for fd in AVAILABLE_FILTERS if fd.allowed_contexts & effective_context]
+        # User-supplied filters are always allowed (but not SDK-injected ones)
+        for user_filter in self._user_filter_names:
+            if user_filter not in allowed_list:
+                allowed_list.append(user_filter)
 
         env = self.get_environment()
         template_source = self._template
diff --git a/infrahub_sdk/template/exceptions.py b/infrahub_sdk/template/exceptions.py
index 6ef60b43..1b45c7cb 100644
--- a/infrahub_sdk/template/exceptions.py
+++ b/infrahub_sdk/template/exceptions.py
@@ -39,3 +39,13 @@ def __init__(self, message: str | None, errors: list[UndefinedJinja2Error]) -> N
 class JinjaTemplateOperationViolationError(JinjaTemplateError):
     def __init__(self, message: str | None = None) -> None:
         self.message = message or "Forbidden code found in the template"
+
+
+class JinjaFilterError(JinjaTemplateError):
+    def __init__(self, filter_name: str, message: str, hint: str | None = None) -> None:
+        self.filter_name = filter_name
+        self.hint = hint
+        full_message = f"Filter '{filter_name}': {message}"
+        if hint is not None:
+            full_message += f" — {hint}"
+        super().__init__(full_message)
diff --git a/infrahub_sdk/template/filters.py b/infrahub_sdk/template/filters.py
index 1d082b39..004011a8 100644
--- a/infrahub_sdk/template/filters.py
+++ b/infrahub_sdk/template/filters.py
@@ -1,151 +1,176 @@
 from dataclasses import dataclass
+from enum import Flag, auto
+
+
+class ExecutionContext(Flag):
+    CORE = auto()
+    WORKER = auto()
+    LOCAL = auto()
+    ALL = CORE | WORKER | LOCAL
 
 
 @dataclass
 class FilterDefinition:
     name: str
-    trusted: bool
+    allowed_contexts: ExecutionContext
     source: str
 
+    @property
+    def trusted(self) -> bool:
+        """Backward compatibility: trusted means allowed in all contexts."""
+        return self.allowed_contexts == ExecutionContext.ALL
+
 
 BUILTIN_FILTERS = [
-    FilterDefinition(name="abs", trusted=True, source="jinja2"),
-    FilterDefinition(name="attr", trusted=False, source="jinja2"),
-    FilterDefinition(name="batch", trusted=False, source="jinja2"),
-    FilterDefinition(name="capitalize", trusted=True, source="jinja2"),
-    FilterDefinition(name="center", trusted=True, source="jinja2"),
-    FilterDefinition(name="count", trusted=True, source="jinja2"),
-    FilterDefinition(name="d", trusted=True, source="jinja2"),
-    FilterDefinition(name="default", trusted=True, source="jinja2"),
-    FilterDefinition(name="dictsort", trusted=False, source="jinja2"),
-    FilterDefinition(name="e", trusted=True, source="jinja2"),
-    FilterDefinition(name="escape", trusted=True, source="jinja2"),
-    FilterDefinition(name="filesizeformat", trusted=True, source="jinja2"),
-    FilterDefinition(name="first", trusted=True, source="jinja2"),
-    FilterDefinition(name="float", trusted=True, source="jinja2"),
-    FilterDefinition(name="forceescape", trusted=True, source="jinja2"),
-    FilterDefinition(name="format", trusted=True, source="jinja2"),
-    FilterDefinition(name="groupby", trusted=False, source="jinja2"),
-    FilterDefinition(name="indent", trusted=True, source="jinja2"),
-    FilterDefinition(name="int", trusted=True, source="jinja2"),
-    FilterDefinition(name="items", trusted=False, source="jinja2"),
-    FilterDefinition(name="join", trusted=True, source="jinja2"),
-    FilterDefinition(name="last", trusted=True, source="jinja2"),
-    FilterDefinition(name="length", trusted=True, source="jinja2"),
-    FilterDefinition(name="list", trusted=True, source="jinja2"),
-    FilterDefinition(name="lower", trusted=True, source="jinja2"),
-    FilterDefinition(name="map", trusted=False, source="jinja2"),
-    FilterDefinition(name="max", trusted=True, source="jinja2"),
-    FilterDefinition(name="min", trusted=True, source="jinja2"),
-    FilterDefinition(name="pprint", trusted=False, source="jinja2"),
-    FilterDefinition(name="random", trusted=False, source="jinja2"),
-    FilterDefinition(name="reject", trusted=False, source="jinja2"),
-    FilterDefinition(name="rejectattr", trusted=False, source="jinja2"),
-    FilterDefinition(name="replace", trusted=True, source="jinja2"),
-    FilterDefinition(name="reverse", trusted=True, source="jinja2"),
-    FilterDefinition(name="round", trusted=True, source="jinja2"),
-    FilterDefinition(name="safe", trusted=False, source="jinja2"),
-    FilterDefinition(name="select", trusted=False, source="jinja2"),
-    FilterDefinition(name="selectattr", trusted=False, source="jinja2"),
-    FilterDefinition(name="slice", trusted=True, source="jinja2"),
-    FilterDefinition(name="sort", trusted=False, source="jinja2"),
-    FilterDefinition(name="string", trusted=True, source="jinja2"),
-    FilterDefinition(name="striptags", trusted=True, source="jinja2"),
-    FilterDefinition(name="sum", trusted=True, source="jinja2"),
-    FilterDefinition(name="title", trusted=True, source="jinja2"),
-    FilterDefinition(name="tojson", trusted=False, source="jinja2"),
-    FilterDefinition(name="trim", trusted=True, source="jinja2"),
-    FilterDefinition(name="truncate", trusted=True, source="jinja2"),
-    FilterDefinition(name="unique", trusted=False, source="jinja2"),
-    FilterDefinition(name="upper", trusted=True, source="jinja2"),
-    FilterDefinition(name="urlencode", trusted=True, source="jinja2"),
-    FilterDefinition(name="urlize", trusted=False, source="jinja2"),
-    FilterDefinition(name="wordcount", trusted=True, source="jinja2"),
-    FilterDefinition(name="wordwrap", trusted=True, source="jinja2"),
-    FilterDefinition(name="xmlattr", trusted=False, source="jinja2"),
+    FilterDefinition(name="abs", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="attr", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="batch", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="capitalize", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="center", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="count", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="d", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="default", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="dictsort", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="e", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="escape", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="filesizeformat", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="first", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="float", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="forceescape", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="format", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="groupby", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="indent", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="int", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="items", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="join", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="last", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="length", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="list", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="lower", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="map", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="max", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="min", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="pprint", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="random", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="reject", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="rejectattr", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="replace", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="reverse", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="round", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="safe", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="select", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="selectattr", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="slice", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="sort", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="string", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="striptags", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="sum", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="title", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="tojson", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="trim", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="truncate", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="unique", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="upper", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="urlencode", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="urlize", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
+    FilterDefinition(name="wordcount", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="wordwrap", allowed_contexts=ExecutionContext.ALL, source="jinja2"),
+    FilterDefinition(name="xmlattr", allowed_contexts=ExecutionContext.LOCAL, source="jinja2"),
 ]
 
 
 NETUTILS_FILTERS = [
-    FilterDefinition(name="abbreviated_interface_name", trusted=True, source="netutils"),
-    FilterDefinition(name="abbreviated_interface_name_list", trusted=True, source="netutils"),
-    FilterDefinition(name="asn_to_int", trusted=True, source="netutils"),
-    FilterDefinition(name="bits_to_name", trusted=True, source="netutils"),
-    FilterDefinition(name="bytes_to_name", trusted=True, source="netutils"),
-    FilterDefinition(name="canonical_interface_name", trusted=True, source="netutils"),
-    FilterDefinition(name="canonical_interface_name_list", trusted=True, source="netutils"),
-    FilterDefinition(name="cidr_to_netmask", trusted=True, source="netutils"),
-    FilterDefinition(name="cidr_to_netmaskv6", trusted=True, source="netutils"),
-    FilterDefinition(name="clean_config", trusted=True, source="netutils"),
-    FilterDefinition(name="compare_version_loose", trusted=True, source="netutils"),
-    FilterDefinition(name="compare_version_strict", trusted=True, source="netutils"),
-    FilterDefinition(name="config_compliance", trusted=True, source="netutils"),
-    FilterDefinition(name="config_section_not_parsed", trusted=True, source="netutils"),
-    FilterDefinition(name="delimiter_change", trusted=True, source="netutils"),
-    FilterDefinition(name="diff_network_config", trusted=True, source="netutils"),
-    FilterDefinition(name="feature_compliance", trusted=True, source="netutils"),
-    FilterDefinition(name="find_unordered_cfg_lines", trusted=True, source="netutils"),
-    FilterDefinition(name="fqdn_to_ip", trusted=False, source="netutils"),
-    FilterDefinition(name="get_all_host", trusted=False, source="netutils"),
-    FilterDefinition(name="get_broadcast_address", trusted=True, source="netutils"),
-    FilterDefinition(name="get_first_usable", trusted=True, source="netutils"),
-    FilterDefinition(name="get_ips_sorted", trusted=True, source="netutils"),
-    FilterDefinition(name="get_nist_urls", trusted=True, source="netutils"),
-    FilterDefinition(name="get_nist_vendor_platform_urls", trusted=True, source="netutils"),
-    FilterDefinition(name="get_oui", trusted=True, source="netutils"),
-    FilterDefinition(name="get_peer_ip", trusted=True, source="netutils"),
-    FilterDefinition(name="get_range_ips", trusted=True, source="netutils"),
-    FilterDefinition(name="get_upgrade_path", trusted=True, source="netutils"),
-    FilterDefinition(name="get_usable_range", trusted=True, source="netutils"),
-    FilterDefinition(name="hash_data", trusted=True, source="netutils"),
-    FilterDefinition(name="int_to_asdot", trusted=True, source="netutils"),
-    FilterDefinition(name="interface_range_compress", trusted=True, source="netutils"),
-    FilterDefinition(name="interface_range_expansion", trusted=True, source="netutils"),
-    FilterDefinition(name="ip_addition", trusted=True, source="netutils"),
-    FilterDefinition(name="ip_subtract", trusted=True, source="netutils"),
-    FilterDefinition(name="ip_to_bin", trusted=True, source="netutils"),
-    FilterDefinition(name="ip_to_hex", trusted=True, source="netutils"),
-    FilterDefinition(name="ipaddress_address", trusted=True, source="netutils"),
-    FilterDefinition(name="ipaddress_interface", trusted=True, source="netutils"),
-    FilterDefinition(name="ipaddress_network", trusted=True, source="netutils"),
-    FilterDefinition(name="is_classful", trusted=True, source="netutils"),
-    FilterDefinition(name="is_fqdn_resolvable", trusted=False, source="netutils"),
-    FilterDefinition(name="is_ip", trusted=True, source="netutils"),
-    FilterDefinition(name="is_ip_range", trusted=True, source="netutils"),
-    FilterDefinition(name="is_ip_within", trusted=True, source="netutils"),
-    FilterDefinition(name="is_netmask", trusted=True, source="netutils"),
-    FilterDefinition(name="is_network", trusted=True, source="netutils"),
-    FilterDefinition(name="is_reversible_wildcardmask", trusted=True, source="netutils"),
-    FilterDefinition(name="is_valid_mac", trusted=True, source="netutils"),
-    FilterDefinition(name="longest_prefix_match", trusted=True, source="netutils"),
-    FilterDefinition(name="mac_normalize", trusted=True, source="netutils"),
-    FilterDefinition(name="mac_to_format", trusted=True, source="netutils"),
-    FilterDefinition(name="mac_to_int", trusted=True, source="netutils"),
-    FilterDefinition(name="mac_type", trusted=True, source="netutils"),
-    FilterDefinition(name="name_to_bits", trusted=True, source="netutils"),
-    FilterDefinition(name="name_to_bytes", trusted=True, source="netutils"),
-    FilterDefinition(name="name_to_name", trusted=True, source="netutils"),
-    FilterDefinition(name="netmask_to_cidr", trusted=True, source="netutils"),
-    FilterDefinition(name="netmask_to_wildcardmask", trusted=True, source="netutils"),
-    FilterDefinition(name="normalise_delimiter_caret_c", trusted=True, source="netutils"),
-    FilterDefinition(name="paloalto_panos_brace_to_set", trusted=True, source="netutils"),
-    FilterDefinition(name="paloalto_panos_clean_newlines", trusted=True, source="netutils"),
-    FilterDefinition(name="regex_findall", trusted=False, source="netutils"),
-    FilterDefinition(name="regex_match", trusted=False, source="netutils"),
-    FilterDefinition(name="regex_search", trusted=False, source="netutils"),
-    FilterDefinition(name="regex_split", trusted=False, source="netutils"),
-    FilterDefinition(name="regex_sub", trusted=False, source="netutils"),
-    FilterDefinition(name="sanitize_config", trusted=True, source="netutils"),
-    FilterDefinition(name="section_config", trusted=True, source="netutils"),
-    FilterDefinition(name="sort_interface_list", trusted=True, source="netutils"),
-    FilterDefinition(name="split_interface", trusted=True, source="netutils"),
-    FilterDefinition(name="uptime_seconds_to_string", trusted=True, source="netutils"),
-    FilterDefinition(name="uptime_string_to_seconds", trusted=True, source="netutils"),
-    FilterDefinition(name="version_metadata", trusted=True, source="netutils"),
-    FilterDefinition(name="vlanconfig_to_list", trusted=True, source="netutils"),
-    FilterDefinition(name="vlanlist_to_config", trusted=True, source="netutils"),
-    FilterDefinition(name="wildcardmask_to_netmask", trusted=True, source="netutils"),
+    FilterDefinition(name="abbreviated_interface_name", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="abbreviated_interface_name_list", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="asn_to_int", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="bits_to_name", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="bytes_to_name", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="canonical_interface_name", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="canonical_interface_name_list", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="cidr_to_netmask", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="cidr_to_netmaskv6", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="clean_config", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="compare_version_loose", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="compare_version_strict", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="config_compliance", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="config_section_not_parsed", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="delimiter_change", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="diff_network_config", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="feature_compliance", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="find_unordered_cfg_lines", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="fqdn_to_ip", allowed_contexts=ExecutionContext.LOCAL, source="netutils"),
+    FilterDefinition(name="get_all_host", allowed_contexts=ExecutionContext.LOCAL, source="netutils"),
+    FilterDefinition(name="get_broadcast_address", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="get_first_usable", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="get_ips_sorted", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="get_nist_urls", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="get_nist_vendor_platform_urls", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="get_oui", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="get_peer_ip", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="get_range_ips", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="get_upgrade_path", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="get_usable_range", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="hash_data", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="int_to_asdot", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="interface_range_compress", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="interface_range_expansion", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="ip_addition", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="ip_subtract", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="ip_to_bin", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="ip_to_hex", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="ipaddress_address", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="ipaddress_interface", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="ipaddress_network", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="is_classful", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="is_fqdn_resolvable", allowed_contexts=ExecutionContext.LOCAL, source="netutils"),
+    FilterDefinition(name="is_ip", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="is_ip_range", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="is_ip_within", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="is_netmask", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="is_network", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="is_reversible_wildcardmask", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="is_valid_mac", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="longest_prefix_match", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="mac_normalize", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="mac_to_format", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="mac_to_int", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="mac_type", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="name_to_bits", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="name_to_bytes", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="name_to_name", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="netmask_to_cidr", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="netmask_to_wildcardmask", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="normalise_delimiter_caret_c", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="paloalto_panos_brace_to_set", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="paloalto_panos_clean_newlines", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="regex_findall", allowed_contexts=ExecutionContext.LOCAL, source="netutils"),
+    FilterDefinition(name="regex_match", allowed_contexts=ExecutionContext.LOCAL, source="netutils"),
+    FilterDefinition(name="regex_search", allowed_contexts=ExecutionContext.LOCAL, source="netutils"),
+    FilterDefinition(name="regex_split", allowed_contexts=ExecutionContext.LOCAL, source="netutils"),
+    FilterDefinition(name="regex_sub", allowed_contexts=ExecutionContext.LOCAL, source="netutils"),
+    FilterDefinition(name="sanitize_config", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="section_config", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="sort_interface_list", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="split_interface", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="uptime_seconds_to_string", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="uptime_string_to_seconds", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="version_metadata", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="vlanconfig_to_list", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="vlanlist_to_config", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+    FilterDefinition(name="wildcardmask_to_netmask", allowed_contexts=ExecutionContext.ALL, source="netutils"),
+]
+
+
+INFRAHUB_FILTERS = [
+    FilterDefinition(
+        name="artifact_content", allowed_contexts=ExecutionContext.WORKER | ExecutionContext.LOCAL, source="infrahub"
+    ),
+    FilterDefinition(
+        name="file_object_content", allowed_contexts=ExecutionContext.WORKER | ExecutionContext.LOCAL, source="infrahub"
+    ),
+    FilterDefinition(name="from_json", allowed_contexts=ExecutionContext.ALL, source="infrahub"),
+    FilterDefinition(name="from_yaml", allowed_contexts=ExecutionContext.ALL, source="infrahub"),
 ]
 
 
-AVAILABLE_FILTERS = BUILTIN_FILTERS + NETUTILS_FILTERS
+AVAILABLE_FILTERS = BUILTIN_FILTERS + NETUTILS_FILTERS + INFRAHUB_FILTERS
diff --git a/infrahub_sdk/template/infrahub_filters.py b/infrahub_sdk/template/infrahub_filters.py
new file mode 100644
index 00000000..a5261399
--- /dev/null
+++ b/infrahub_sdk/template/infrahub_filters.py
@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import json
+from collections.abc import Callable, Coroutine
+from typing import TYPE_CHECKING, Any
+
+import yaml
+
+from infrahub_sdk.exceptions import AuthenticationError
+from infrahub_sdk.template.exceptions import JinjaFilterError
+
+if TYPE_CHECKING:
+    from infrahub_sdk.client import InfrahubClient
+
+
+class InfrahubFilters:
+    """Holds an InfrahubClient and exposes async filter methods for Jinja2 templates."""
+
+    def __init__(self, client: InfrahubClient) -> None:
+        self.client = client
+
+    async def artifact_content(self, storage_id: str) -> str:
+        """Retrieve artifact content by storage_id."""
+        if storage_id is None:
+            raise JinjaFilterError(
+                filter_name="artifact_content",
+                message="storage_id is null",
+                hint="ensure the GraphQL query returns a valid storage_id value",
+            )
+        if not storage_id:
+            raise JinjaFilterError(
+                filter_name="artifact_content",
+                message="storage_id is empty",
+                hint="ensure the GraphQL query returns a non-empty storage_id value",
+            )
+        try:
+            return await self.client.object_store.get(identifier=storage_id)
+        except AuthenticationError as exc:
+            raise JinjaFilterError(
+                filter_name="artifact_content", message=f"permission denied for storage_id: {storage_id}"
+            ) from exc
+        except Exception as exc:
+            raise JinjaFilterError(
+                filter_name="artifact_content",
+                message=f"failed to retrieve content for storage_id: {storage_id}",
+                hint=str(exc),
+            ) from exc
+
+    async def file_object_content(self, storage_id: str) -> str:
+        """Retrieve file object content by storage_id."""
+        if storage_id is None:
+            raise JinjaFilterError(
+                filter_name="file_object_content",
+                message="storage_id is null",
+                hint="ensure the GraphQL query returns a valid storage_id value",
+            )
+        if not storage_id:
+            raise JinjaFilterError(
+                filter_name="file_object_content",
+                message="storage_id is empty",
+                hint="ensure the GraphQL query returns a non-empty storage_id value",
+            )
+        try:
+            return await self.client.object_store.get_file_by_storage_id(storage_id=storage_id)
+        except AuthenticationError as exc:
+            raise JinjaFilterError(
+                filter_name="file_object_content", message=f"permission denied for storage_id: {storage_id}"
+            ) from exc
+        except ValueError as exc:
+            raise JinjaFilterError(filter_name="file_object_content", message=str(exc)) from exc
+        except Exception as exc:
+            raise JinjaFilterError(
+                filter_name="file_object_content",
+                message=f"failed to retrieve content for storage_id: {storage_id}",
+                hint=str(exc),
+            ) from exc
+
+
+def no_client_filter(filter_name: str) -> Callable[[str], Coroutine[Any, Any, str]]:
+    """Create a filter function that raises JinjaFilterError because no client was provided."""
+
+    async def _filter(storage_id: str) -> str:  # noqa: ARG001
+        raise JinjaFilterError(
+            filter_name=filter_name,
+            message="requires an InfrahubClient",
+            hint="pass a client via Jinja2Template(client=...)",
+        )
+
+    return _filter
+
+
+def from_json(value: str) -> dict | list:
+    """Parse a JSON string into a Python dict or list."""
+    if not value:
+        return {}
+    try:
+        return json.loads(value)
+    except (json.JSONDecodeError, TypeError) as exc:
+        raise JinjaFilterError(filter_name="from_json", message=f"invalid JSON: {exc}") from exc
+
+
+def from_yaml(value: str) -> dict | list:
+    """Parse a YAML string into a Python dict or list."""
+    if not value:
+        return {}
+    try:
+        result = yaml.safe_load(value)
+        # yaml.safe_load("") returns None, normalize to {}
+        if result is None:
+            return {}
+        return result
+    except yaml.YAMLError as exc:
+        raise JinjaFilterError(filter_name="from_yaml", message=f"invalid YAML: {exc}") from exc
diff --git a/tests/unit/sdk/test_infrahub_filters.py b/tests/unit/sdk/test_infrahub_filters.py
new file mode 100644
index 00000000..1606cd9c
--- /dev/null
+++ b/tests/unit/sdk/test_infrahub_filters.py
@@ -0,0 +1,348 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import httpx
+import pytest
+
+from infrahub_sdk.template import Jinja2Template
+from infrahub_sdk.template.exceptions import JinjaFilterError, JinjaTemplateError, JinjaTemplateOperationViolationError
+from infrahub_sdk.template.filters import INFRAHUB_FILTERS, ExecutionContext, FilterDefinition
+from infrahub_sdk.template.infrahub_filters import from_json, from_yaml, no_client_filter
+
+if TYPE_CHECKING:
+    from pytest_httpx import HTTPXMock
+
+    from infrahub_sdk import InfrahubClient
+
+
+pytestmark = pytest.mark.httpx_mock(can_send_already_matched_responses=True)
+
+ARTIFACT_CONTENT_URL = "http://mock/api/storage/object"
+FILE_OBJECT_CONTENT_URL = "http://mock/api/files/by-storage-id"
+
+CLIENT_FILTER_PARAMS = [
+    pytest.param(
+        "artifact_content",
+        "{{ storage_id | artifact_content }}",
+        f"{ARTIFACT_CONTENT_URL}/test-id",
+        {"content-type": "text/plain"},
+        id="artifact_content",
+    ),
+    pytest.param(
+        "file_object_content",
+        "{{ storage_id | file_object_content }}",
+        f"{FILE_OBJECT_CONTENT_URL}/test-id",
+        {"content-type": "text/plain"},
+        id="file_object_content",
+    ),
+]
+
+
+class TestJinjaFilterError:
+    def test_instantiation_without_hint(self) -> None:
+        exc = JinjaFilterError(filter_name="my_filter", message="something broke")
+        assert exc.filter_name == "my_filter"
+        assert exc.hint is None
+        assert exc.message == "Filter 'my_filter': something broke"
+
+    def test_instantiation_with_hint(self) -> None:
+        exc = JinjaFilterError(filter_name="my_filter", message="something broke", hint="try harder")
+        assert exc.filter_name == "my_filter"
+        assert exc.hint == "try harder"
+        assert exc.message == "Filter 'my_filter': something broke — try harder"
+
+
+class TestFilterDefinition:
+    def test_trusted_when_all_contexts(self) -> None:
+        fd = FilterDefinition(name="abs", allowed_contexts=ExecutionContext.ALL, source="jinja2")
+        assert fd.trusted is True
+
+    def test_not_trusted_when_local_only(self) -> None:
+        fd = FilterDefinition(name="safe", allowed_contexts=ExecutionContext.LOCAL, source="jinja2")
+        assert fd.trusted is False
+
+    def test_not_trusted_when_worker_and_local(self) -> None:
+        fd = FilterDefinition(
+            name="artifact_content",
+            allowed_contexts=ExecutionContext.WORKER | ExecutionContext.LOCAL,
+            source="infrahub",
+        )
+        assert fd.trusted is False
+
+    def test_not_trusted_when_core_only(self) -> None:
+        fd = FilterDefinition(name="custom", allowed_contexts=ExecutionContext.CORE, source="test")
+        assert fd.trusted is False
+
+    def test_infrahub_filters_list_sorted(self) -> None:
+        """Infrahub filter names should be in alphabetical order."""
+        names = [fd.name for fd in INFRAHUB_FILTERS]
+        assert names == sorted(names)
+
+
+class TestValidateContext:
+    def test_restricted_true_blocks_untrusted_filters(self) -> None:
+        """restricted=True behaves like ExecutionContext.CORE -- blocks LOCAL-only filters."""
+        jinja = Jinja2Template(template="{{ network | get_all_host }}")
+        with pytest.raises(JinjaTemplateOperationViolationError) as exc:
+            jinja.validate(restricted=True)
+        assert exc.value.message == "The 'get_all_host' filter isn't allowed to be used"
+
+    def test_restricted_false_allows_all_filters(self) -> None:
+        """restricted=False behaves like ExecutionContext.LOCAL -- allows everything."""
+        jinja = Jinja2Template(template="{{ network | get_all_host }}")
+        jinja.validate(restricted=False)
+
+    def test_context_core_blocks_artifact_content(self) -> None:
+        jinja = Jinja2Template(template="{{ sid | artifact_content }}")
+        with pytest.raises(JinjaTemplateOperationViolationError) as exc:
+            jinja.validate(context=ExecutionContext.CORE)
+        assert exc.value.message == "The 'artifact_content' filter isn't allowed to be used"
+
+    def test_context_worker_allows_artifact_content(self) -> None:
+        jinja = Jinja2Template(template="{{ sid | artifact_content }}")
+        jinja.validate(context=ExecutionContext.WORKER)
+
+    def test_context_worker_blocks_local_only_filters(self) -> None:
+        """WORKER context should still block LOCAL-only filters like 'safe'."""
+        jinja = Jinja2Template(template="{{ data | safe }}")
+        with pytest.raises(JinjaTemplateOperationViolationError) as exc:
+            jinja.validate(context=ExecutionContext.WORKER)
+        assert exc.value.message == "The 'safe' filter isn't allowed to be used"
+
+    def test_context_local_allows_local_only_filters(self) -> None:
+        jinja = Jinja2Template(template="{{ data | safe }}")
+        jinja.validate(context=ExecutionContext.LOCAL)
+
+    def test_context_local_allows_artifact_content(self) -> None:
+        """LOCAL context allows artifact_content (WORKER | LOCAL)."""
+        jinja = Jinja2Template(template="{{ sid | artifact_content }}")
+        jinja.validate(context=ExecutionContext.LOCAL)
+
+    @pytest.mark.parametrize("context", [ExecutionContext.CORE, ExecutionContext.WORKER])
+    def test_user_filters_always_allowed(self, context: ExecutionContext) -> None:
+        def my_custom_filter(value: str) -> str:
+            return value.upper()
+
+        jinja = Jinja2Template(template="{{ name | my_custom }}", filters={"my_custom": my_custom_filter})
+        jinja.validate(context=context)
+
+    def test_context_core_allows_from_json(self) -> None:
+        jinja = Jinja2Template(template="{{ '{\"a\":1}' | from_json }}")
+        jinja.validate(context=ExecutionContext.CORE)
+
+    def test_context_core_blocks_file_object_content(self) -> None:
+        jinja = Jinja2Template(template="{{ sid | file_object_content }}")
+        with pytest.raises(JinjaTemplateOperationViolationError) as exc:
+            jinja.validate(context=ExecutionContext.CORE)
+        assert exc.value.message == "The 'file_object_content' filter isn't allowed to be used"
+
+    def test_context_worker_allows_file_object_content(self) -> None:
+        jinja = Jinja2Template(template="{{ sid | file_object_content }}")
+        jinja.validate(context=ExecutionContext.WORKER)
+
+
+class TestClientDependentFilters:
+    @pytest.mark.parametrize(("filter_name", "template", "url", "headers"), CLIENT_FILTER_PARAMS)
+    async def test_happy_path(
+        self,
+        filter_name: str,
+        template: str,
+        url: str,
+        headers: dict[str, str],
+        client: InfrahubClient,
+        httpx_mock: HTTPXMock,
+    ) -> None:
+        httpx_mock.add_response(method="GET", url=url, text="rendered content", headers=headers)
+        jinja = Jinja2Template(template=template, client=client)
+        result = await jinja.render(variables={"storage_id": "test-id"})
+        assert result == "rendered content"
+
+    @pytest.mark.parametrize(("filter_name", "template", "url", "headers"), CLIENT_FILTER_PARAMS)
+    @pytest.mark.parametrize(
+        ("storage_id_value", "expected_message"),
+        [
+            pytest.param(
+                None,
+                "Filter '{filter_name}': storage_id is null"
+                " — ensure the GraphQL query returns a valid storage_id value",
+                id="null",
+            ),
+            pytest.param(
+                "",
+                "Filter '{filter_name}': storage_id is empty"
+                " — ensure the GraphQL query returns a non-empty storage_id value",
+                id="empty",
+            ),
+        ],
+    )
+    async def test_invalid_storage_id(
+        self,
+        filter_name: str,
+        template: str,
+        url: str,
+        headers: dict[str, str],
+        storage_id_value: str | None,
+        expected_message: str,
+        client: InfrahubClient,
+    ) -> None:
+        jinja = Jinja2Template(template=template, client=client)
+        with pytest.raises(JinjaTemplateError) as exc:
+            await jinja.render(variables={"storage_id": storage_id_value})
+        assert exc.value.message == expected_message.format(filter_name=filter_name)
+
+    @pytest.mark.parametrize(
+        ("template", "url"),
+        [
+            pytest.param(
+                "{{ storage_id | artifact_content }}",
+                f"{ARTIFACT_CONTENT_URL}/abc-123",
+                id="artifact_content",
+            ),
+            pytest.param(
+                "{{ storage_id | file_object_content }}",
+                f"{FILE_OBJECT_CONTENT_URL}/abc-123",
+                id="file_object_content",
+            ),
+        ],
+    )
+    async def test_store_exception_is_wrapped(
+        self, template: str, url: str, client: InfrahubClient, httpx_mock: HTTPXMock
+    ) -> None:
+        httpx_mock.add_exception(httpx.ConnectError("connection timeout"), method="GET", url=url)
+        jinja = Jinja2Template(template=template, client=client)
+        with pytest.raises(JinjaTemplateError):
+            await jinja.render(variables={"storage_id": "abc-123"})
+
+    @pytest.mark.parametrize(
+        ("template", "url", "storage_id", "expected_message"),
+        [
+            pytest.param(
+                "{{ storage_id | artifact_content }}",
+                f"{ARTIFACT_CONTENT_URL}/sid-x",
+                "sid-x",
+                "Filter 'artifact_content': permission denied for storage_id: sid-x",
+                id="artifact_content",
+            ),
+            pytest.param(
+                "{{ storage_id | file_object_content }}",
+                f"{FILE_OBJECT_CONTENT_URL}/fid-x",
+                "fid-x",
+                "Filter 'file_object_content': permission denied for storage_id: fid-x",
+                id="file_object_content",
+            ),
+        ],
+    )
+    async def test_auth_error(
+        self,
+        template: str,
+        url: str,
+        storage_id: str,
+        expected_message: str,
+        client: InfrahubClient,
+        httpx_mock: HTTPXMock,
+    ) -> None:
+        httpx_mock.add_response(method="GET", url=url, status_code=403, json={"errors": [{"message": "forbidden"}]})
+        jinja = Jinja2Template(template=template, client=client)
+        with pytest.raises(JinjaTemplateError) as exc:
+            await jinja.render(variables={"storage_id": storage_id})
+        assert exc.value.message == expected_message
+
+    async def test_file_object_content_binary_content_rejected(
+        self, client: InfrahubClient, httpx_mock: HTTPXMock
+    ) -> None:
+        httpx_mock.add_response(
+            method="GET",
+            url=f"{FILE_OBJECT_CONTENT_URL}/fid-bin",
+            content=b"\x00\x01\x02",
+            headers={"content-type": "application/octet-stream"},
+        )
+        jinja = Jinja2Template(template="{{ storage_id | file_object_content }}", client=client)
+        with pytest.raises(JinjaTemplateError) as exc:
+            await jinja.render(variables={"storage_id": "fid-bin"})
+        assert (
+            exc.value.message == "Filter 'file_object_content': Binary content not supported:"
+            " content-type 'application/octet-stream' for storage_id 'fid-bin'"
+        )
+
+
+class TestFromJsonFilter:
+    def test_valid_json(self) -> None:
+        result = from_json('{"key": "value", "num": 42}')
+        assert result == {"key": "value", "num": 42}
+
+    def test_valid_json_list(self) -> None:
+        result = from_json("[1, 2, 3]")
+        assert result == [1, 2, 3]
+
+    def test_empty_string_returns_empty_dict(self) -> None:
+        assert from_json("") == {}
+
+    def test_malformed_json_raises_error(self) -> None:
+        with pytest.raises(JinjaFilterError) as exc:
+            from_json("{not valid json}")
+        assert exc.value.filter_name == "from_json"
+        assert exc.value.message is not None
+        assert exc.value.message.startswith("Filter 'from_json': invalid JSON: Expecting property name")
+
+    async def test_render_through_template(self) -> None:
+        jinja = Jinja2Template(template="{{ data | from_json }}")
+        result = await jinja.render(variables={"data": '{"a": 1}'})
+        assert result == "{'a': 1}"
+
+
+class TestFromYamlFilter:
+    def test_valid_yaml(self) -> None:
+        result = from_yaml("key: value\nnum: 42")
+        assert result == {"key": "value", "num": 42}
+
+    def test_valid_yaml_list(self) -> None:
+        result = from_yaml("- one\n- two\n- three")
+        assert result == ["one", "two", "three"]
+
+    def test_empty_string_returns_empty_dict(self) -> None:
+        assert from_yaml("") == {}
+
+    def test_malformed_yaml_raises_error(self) -> None:
+        with pytest.raises(JinjaFilterError) as exc:
+            from_yaml("key:\n\t- broken: [unclosed")
+        assert exc.value.filter_name == "from_yaml"
+        assert exc.value.message is not None
+        assert exc.value.message.startswith("Filter 'from_yaml': invalid YAML: while scanning for the next token")
+
+    async def test_render_through_template(self) -> None:
+        jinja = Jinja2Template(template="{{ data | from_yaml }}")
+        result = await jinja.render(variables={"data": "key: value"})
+        assert result == "{'key': 'value'}"
+
+
+class TestFilterChaining:
+    async def test_artifact_content_piped_to_from_json(self, client: InfrahubClient, httpx_mock: HTTPXMock) -> None:
+        json_payload = '{"hostname": "router1", "interfaces": ["eth0", "eth1"]}'
+        httpx_mock.add_response(method="GET", url=f"{ARTIFACT_CONTENT_URL}/store-789", text=json_payload)
+        jinja = Jinja2Template(template="{{ storage_id | artifact_content | from_json }}", client=client)
+        result = await jinja.render(variables={"storage_id": "store-789"})
+        assert result == "{'hostname': 'router1', 'interfaces': ['eth0', 'eth1']}"
+
+
+class TestClientFilter:
+    @pytest.mark.parametrize("filter_name", ["artifact_content", "file_object_content"])
+    async def test_no_client_filter_raises(self, filter_name: str) -> None:
+        fallback = no_client_filter(filter_name)
+        with pytest.raises(JinjaFilterError) as exc:
+            await fallback("some-id")
+        assert exc.value.message == (
+            f"Filter '{filter_name}': requires an InfrahubClient — pass a client via Jinja2Template(client=...)"
+        )
+        assert exc.value.filter_name == filter_name
+
+    async def test_set_client_enables_artifact_content(self, client: InfrahubClient, httpx_mock: HTTPXMock) -> None:
+        httpx_mock.add_response(method="GET", url=f"{ARTIFACT_CONTENT_URL}/abc", text="deferred content")
+        tpl = Jinja2Template(template="{{ sid | artifact_content }}")
+
+        with pytest.raises(JinjaTemplateError):
+            await tpl.render(variables={"sid": "abc"})
+
+        tpl.set_client(client=client)
+        result = await tpl.render(variables={"sid": "abc"})
+        assert result == "deferred content"