diff --git a/nemoguardrails/http/__init__.py b/nemoguardrails/http/__init__.py
index 48c22c277f..d599a01e87 100644
--- a/nemoguardrails/http/__init__.py
+++ b/nemoguardrails/http/__init__.py
@@ -23,7 +23,9 @@
     HTTPStatusError,
     HTTPTimeoutError,
 )
+from nemoguardrails.http.request import http_call
 from nemoguardrails.http.retry import RetryingHTTPClient, RetryPolicy
+from nemoguardrails.http.runtime import create_http_client
 from nemoguardrails.http.transport import HttpxHTTPClient
 from nemoguardrails.http.types import HTTPRequest, HTTPResponse
 
@@ -40,4 +42,6 @@
     "HttpxHTTPClient",
     "RetryPolicy",
     "RetryingHTTPClient",
+    "create_http_client",
+    "http_call",
 ]
diff --git a/nemoguardrails/http/errors.py b/nemoguardrails/http/errors.py
index 4e8c49cdb7..e3e72be2f4 100644
--- a/nemoguardrails/http/errors.py
+++ b/nemoguardrails/http/errors.py
@@ -58,6 +58,8 @@ def __init__(self, response: "HTTPResponse", request: "HTTPRequest | None" = Non
         super().__init__(message)
         self.response = response
         self.request = request
+        retry_count = response.extensions.get("retry_count", 0)
+        self.retry_count = retry_count if isinstance(retry_count, int) else 0
 
 
 class HTTPResponseDecodeError(HTTPClientError):
diff --git a/nemoguardrails/http/request.py b/nemoguardrails/http/request.py
new file mode 100644
index 0000000000..d85a7b2de0
--- /dev/null
+++ b/nemoguardrails/http/request.py
@@ -0,0 +1,107 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Managed request lifecycle helpers for outbound HTTP calls."""
+
+from collections.abc import AsyncIterator, Callable
+from contextlib import asynccontextmanager
+from typing import Any, Mapping
+
+from nemoguardrails.http.client import ClosableHTTPClient, HTTPClient
+from nemoguardrails.http.runtime import create_http_client
+from nemoguardrails.http.types import HTTPRequest, HTTPResponse
+
+
+@asynccontextmanager
+async def _resolve_http_client(
+    client: HTTPClient | None,
+    *,
+    factory: Callable[[], ClosableHTTPClient],
+) -> AsyncIterator[HTTPClient]:
+    """Yield an injected client or create and close an owned client."""
+
+    if client is not None:
+        yield client
+        return
+
+    owned = factory()
+    if not isinstance(owned, ClosableHTTPClient):
+        raise TypeError("HTTP client factory must return a closable HTTP client")
+    try:
+        yield owned
+    finally:
+        await owned.close()
+
+
+async def http_call(
+    client: HTTPClient | None,
+    method: str,
+    url: str,
+    *,
+    headers: Mapping[str, str] | None = None,
+    params: Mapping[str, Any] | None = None,
+    json: Any = None,
+    content: bytes | str | None = None,
+    timeout: float | None = None,
+    raise_for_status: bool = True,
+    factory: Callable[[], ClosableHTTPClient] = create_http_client,
+) -> HTTPResponse:
+    """Execute one outbound HTTP request with deterministic client ownership.
+
+    Args:
+        client: Optional caller-owned client. When omitted, ``factory`` creates
+            a client that is closed after the request.
+        method: HTTP method.
+        url: Absolute request URL.
+        headers: Optional request headers.
+        params: Optional query parameters.
+        json: Optional JSON-serializable request body.
+        content: Optional raw request body.
+        timeout: Optional total request timeout in seconds.
+        raise_for_status: Whether responses with status 400 or greater raise.
+        factory: Factory for an owned client when ``client`` is omitted.
+
+    Returns:
+        The materialized transport-neutral response.
+
+    Raises:
+        HTTPStatusError: If ``raise_for_status`` is enabled and the response
+            status is 400 or greater.
+        TypeError: If ``factory`` returns a client that cannot be closed.
+    """
+
+    async with _resolve_http_client(client, factory=factory) as resolved:
+        response = await resolved.request(
+            method,
+            url,
+            headers=headers,
+            params=params,
+            json=json,
+            content=content,
+            timeout=timeout,
+        )
+    if raise_for_status and response.status_code >= 400:
+        response.raise_for_status(
+            HTTPRequest(
+                method=method,
+                url=url,
+                headers=headers,
+                params=params,
+                json=json,
+                content=content,
+                timeout=timeout,
+            )
+        )
+    return response
diff --git a/nemoguardrails/http/runtime.py b/nemoguardrails/http/runtime.py
new file mode 100644
index 0000000000..e0853473e6
--- /dev/null
+++ b/nemoguardrails/http/runtime.py
@@ -0,0 +1,64 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Construction helpers for managed outbound HTTP clients."""
+
+import httpx
+
+from nemoguardrails.http.client import ClosableHTTPClient
+from nemoguardrails.http.retry import RetryingHTTPClient, RetryPolicy
+from nemoguardrails.http.transport import HttpxHTTPClient
+
+
+def create_http_client(
+    *,
+    httpx_client: httpx.AsyncClient | None = None,
+    timeout: float | None = 30.0,
+    limits: httpx.Limits | None = None,
+    retry_policy: RetryPolicy | None = None,
+    follow_redirects: bool = False,
+) -> ClosableHTTPClient:
+    """Create a transport-neutral HTTP client with optional retry behavior.
+
+    Args:
+        httpx_client: Optional caller-owned HTTPX client.
+        timeout: Total timeout for a client created by this function.
+        limits: Connection-pool limits for a client created by this function.
+        retry_policy: Optional policy applied to created or injected clients.
+        follow_redirects: Whether a client created by this function follows
+            redirects.
+
+    Returns:
+        A closable transport-neutral HTTP client.
+
+    Raises:
+        ValueError: If non-default owned-client options are supplied with an
+            injected client.
+
+    When ``httpx_client`` is provided, it remains caller-owned and ``timeout``,
+    ``limits``, and ``follow_redirects`` must retain their defaults.
+    ``retry_policy`` is still applied when supplied.
+    """
+
+    transport = HttpxHTTPClient(
+        httpx_client,
+        timeout=timeout,
+        limits=limits,
+        follow_redirects=follow_redirects,
+    )
+    client: ClosableHTTPClient = transport
+    if retry_policy is not None:
+        client = RetryingHTTPClient(transport, retry_policy)
+    return client
diff --git a/tests/http/test_request.py b/tests/http/test_request.py
new file mode 100644
index 0000000000..c407cbcdfe
--- /dev/null
+++ b/tests/http/test_request.py
@@ -0,0 +1,155 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest import mock
+
+import pytest
+
+from nemoguardrails.http import HTTPConnectionError, HTTPResponse, HTTPStatusError, http_call
+from nemoguardrails.http.types import HTTPRequest
+from nemoguardrails.testing import RecordingHTTPClient
+
+
+@pytest.mark.asyncio
+async def test_http_call_skips_request_context_for_successful_response():
+    client = RecordingHTTPClient([HTTPResponse(status_code=200)])
+
+    with mock.patch("nemoguardrails.http.request.HTTPRequest") as request_factory:
+        await http_call(client, "GET", "https://example.com/check")
+
+    request_factory.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_http_call_forwards_the_request_and_returns_response():
+    response = HTTPResponse(status_code=200, content=b"ok")
+    client = RecordingHTTPClient([response])
+    headers = {"Authorization": "Bearer secret"}
+    params = {"version": 1}
+    payload = {"text": "hello"}
+
+    result = await http_call(
+        client,
+        "POST",
+        "https://example.com/check",
+        headers=headers,
+        params=params,
+        json=payload,
+        timeout=3.0,
+    )
+
+    assert result is response
+    assert client.requests == [
+        HTTPRequest(
+            method="POST",
+            url="https://example.com/check",
+            headers=headers,
+            params=params,
+            json=payload,
+            timeout=3.0,
+        )
+    ]
+
+
+@pytest.mark.asyncio
+async def test_http_call_raises_status_error_with_retry_context():
+    response = HTTPResponse(status_code=503, extensions={"retry_count": 2})
+    client = RecordingHTTPClient([response])
+
+    with pytest.raises(HTTPStatusError) as exc_info:
+        await http_call(
+            client,
+            "GET",
+            "https://user:password@example.com/check?token=secret",
+        )
+
+    assert exc_info.value.response is response
+    assert exc_info.value.retry_count == 2
+    assert "https://example.com/check" in str(exc_info.value)
+    assert "password" not in str(exc_info.value)
+    assert "secret" not in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_http_call_ignores_invalid_retry_metadata():
+    response = HTTPResponse(status_code=503, extensions={"retry_count": "unknown"})
+    client = RecordingHTTPClient([response])
+
+    with pytest.raises(HTTPStatusError) as exc_info:
+        await http_call(client, "GET", "https://example.com/check")
+
+    assert exc_info.value.retry_count == 0
+
+
+@pytest.mark.asyncio
+async def test_http_call_can_return_error_response_without_raising():
+    response = HTTPResponse(status_code=404)
+    client = RecordingHTTPClient([response])
+
+    result = await http_call(
+        client,
+        "GET",
+        "https://example.com/missing",
+        raise_for_status=False,
+    )
+
+    assert result is response
+
+
+@pytest.mark.asyncio
+async def test_http_call_preserves_client_error():
+    error = HTTPConnectionError("unavailable")
+    client = RecordingHTTPClient([error])
+
+    with pytest.raises(HTTPConnectionError) as exc_info:
+        await http_call(client, "GET", "https://example.com/check")
+
+    assert exc_info.value is error
+
+
+@pytest.mark.asyncio
+async def test_http_call_closes_only_the_client_it_creates():
+    response = HTTPResponse(status_code=200, content=b'{"ok": true}')
+    owned = RecordingHTTPClient([response])
+
+    result = await http_call(None, "GET", "https://example.com/check", factory=lambda: owned)
+
+    assert owned.close_calls == 1
+    assert result.json() == {"ok": True}
+
+    injected = RecordingHTTPClient([HTTPResponse(status_code=200)])
+    await http_call(injected, "GET", "https://example.com/check")
+
+    assert injected.close_calls == 0
+
+
+@pytest.mark.asyncio
+async def test_http_call_closes_owned_client_when_request_raises():
+    owned = RecordingHTTPClient()
+
+    with pytest.raises(RuntimeError, match="No HTTP responses available"):
+        await http_call(None, "GET", "https://example.com/check", factory=lambda: owned)
+
+    assert owned.close_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_http_call_rejects_unmanaged_factory_result():
+    class UnmanagedClient:
+        async def request(self, method, url, **kwargs):
+            return HTTPResponse(status_code=200)
+
+    with pytest.raises(TypeError, match="closable HTTP client"):
+        await http_call(None, "GET", "https://example.com/check", factory=lambda: UnmanagedClient())
diff --git a/tests/http/test_runtime.py b/tests/http/test_runtime.py
new file mode 100644
index 0000000000..f3f571b7ba
--- /dev/null
+++ b/tests/http/test_runtime.py
@@ -0,0 +1,82 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest import mock
+
+import httpx
+import pytest
+
+from nemoguardrails.http import (
+    ClosableHTTPClient,
+    HTTPClient,
+    RetryPolicy,
+    create_http_client,
+)
+
+
+def test_default_factory_forwards_owned_client_options():
+    limits = httpx.Limits(max_connections=12, max_keepalive_connections=4)
+
+    with mock.patch("nemoguardrails.http.transport.httpx.AsyncClient") as factory:
+        create_http_client(limits=limits, follow_redirects=True)
+
+    factory.assert_called_once_with(
+        timeout=None,
+        limits=limits,
+        follow_redirects=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_default_factory_composes_a_closable_client():
+    async def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"ok": True}, request=request)
+
+    injected = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+    client = create_http_client(
+        httpx_client=injected,
+        retry_policy=RetryPolicy(max_attempts=1),
+    )
+
+    response = await client.request("GET", "https://example.com/check")
+
+    assert isinstance(client, HTTPClient)
+    assert isinstance(client, ClosableHTTPClient)
+    assert response.json() == {"ok": True}
+    assert response.extensions["retry_count"] == 0
+    await client.close()
+    assert not injected.is_closed
+    await injected.aclose()
+
+
+@pytest.mark.asyncio
+async def test_default_factory_does_not_retry():
+    request_count = 0
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal request_count
+        request_count += 1
+        return httpx.Response(503, request=request)
+
+    injected = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+    client = create_http_client(httpx_client=injected)
+
+    response = await client.request("POST", "https://example.com/check")
+
+    assert response.status_code == 503
+    assert request_count == 1
+    assert "retry_count" not in response.extensions
+    await client.close()
+    await injected.aclose()