feat(adapters): adapter middleware framework (base)

fern/versions/latest/pages/model-server/adapters.mdx

@@ -0,0 +1,68 @@
+---
+title: "Adapter Middleware"
+description: "Interceptor-based middleware framework for Model, Agent, and Resources servers."
+position: 3
+---
+
+Adapter middleware adds an interceptor chain to a server's request/response path. Each interceptor can observe or mutate the payload without the host server changing. Use it to inject system prompts, drop unsupported params, cache responses, count turns, normalize reasoning fields, log tokens, and so on.
+
+`adapters` is opt-in. It is declared on `BaseResponsesAPIModelConfig`, `BaseResponsesAPIAgentConfig`, and `BaseResourcesServerConfig`, so every in-tree server inheriting from `SimpleResponsesAPIModel` / `SimpleResponsesAPIAgent` / `SimpleResourcesServer` accepts an `adapters` block automatically. Omitting it leaves behavior identical to the base server.
+
+## Quickstart
+
+Add an `adapters` list to any server config:
+
+```yaml
+policy_model:
+  responses_api_models:
+    openai_model:
+      openai_base_url: https://api.openai.com/v1
+      openai_api_key: ...
+      openai_model: gpt-4.1
+      adapters:                       # ← toggle: omit or null = OFF
+        - {name: logging, config: {}}
+```
+
+This PR ships the **framework** (`AdapterPipeline`, `InterceptorRegistry`, `install_middleware`, `start_adapter_proxy`) plus two built-in interceptors:
+
+| Name | Stage | Purpose |
+|------|-------|---------|
+| `logging` | request + response | Log request body keys and response status/latency. Canonical "did the chain fire?" probe. |
+| `endpoint` | request → response | Drive the upstream HTTP call directly. **Only used by `start_adapter_proxy`** (standalone host mode); forbidden inside `install_middleware`. |
+
+Additional interceptor families (observability, caching, request rewriting) ship in follow-on PRs.
+
+## Host modes
+
+The same `AdapterPipeline` runs in either of two hosting modes:
+
+**Middleware mode** — `install_middleware(app, adapters)` attaches the pipeline to an existing FastAPI app via `app.middleware("http")`. The host server's own routing performs the upstream call via `call_next`. Used by every Model/Agent/Resources server when `adapters` is set on its config.
+
+**Proxy mode** — `start_adapter_proxy(upstream_url, adapters)` launches a localhost uvicorn that hosts the pipeline with its own forwarding logic. Used by agents that bring their own SDK client (e.g. `claude_code_agent` with `anthropic_base_url`); the agent points its SDK's `*_BASE_URL` at the proxy URL via the `adapter_proxy` field on `BaseResponsesAPIAgentConfig`.
+
+## Path-Based Session Scoping
+
+Posts to `/s/<hex-id>/<path>` have the prefix stripped before forwarding, and `<hex-id>` is recorded as `ctx.extra["session_id"]`. Follow-on interceptors that key per-session (e.g. `turn_counter`, `caching`) use this id.
+
+## Custom Interceptors
+
+Register a class at runtime via `InterceptorRegistry.register`:
+
+```python
+from nemo_gym.adapters import InterceptorRegistry
+
+InterceptorRegistry.register("my_interceptor", "myproject.adapters.my_interceptor")
+```
+
+The target module must expose a class named `Interceptor` that subclasses one of `RequestInterceptor`, `RequestToResponseInterceptor`, or `ResponseInterceptor` from `nemo_gym.adapters.types`. The class is instantiated with the YAML `config` dict as kwargs.
+
+## Configuration Reference
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `adapters` | `list[dict] \| null` | `null` | Ordered interceptor specs on each server config. Each entry is `{name: <str>, config: <dict>}`. `null` or `[]` disables the middleware. |
+| `adapter_proxy` | `AdapterProxyConfig \| null` | `null` | On `BaseResponsesAPIAgentConfig` only. Configures a localhost proxy in front of an external inference upstream. Fields: `upstream_url`, `adapters`, `host` (default `127.0.0.1`), `port` (default `0` → kernel-assigned), `request_timeout`, `unsafe_allow_remote`. |
+
+<Warning>
+`start_adapter_proxy` refuses any `host` other than `127.0.0.1`/`localhost` unless you pass `unsafe_allow_remote=True`. The proxy forwards the client's `Authorization` header verbatim to the upstream, so binding to `0.0.0.0` would leak the upstream API key to any caller on the network.
+</Warning>

fern/versions/latest/pages/model-server/index.mdx

@@ -38,3 +38,9 @@ Let NeMo Gym launch and manage the vLLM server for you.
 [Model Server Fields](/reference/configuration#model-server-fields) for server configuration syntax and fields.
 
 </Note>
+
+## Middleware
+
+Model servers can attach an interceptor chain that observes or mutates request and response payloads — useful for logging, caching, system-prompt injection, turn budgeting, reasoning-field normalization, and similar cross-cutting hooks.
+
+See [Adapter Middleware](/model-server/adapters) for the framework, built-in interceptors, and configuration syntax.

nemo_gym/adapters/__init__.py

@@ -0,0 +1,58 @@
+# 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.
+"""Gym adapter framework — interceptor-based middleware for responses_api_models.
+
+Public API:
+    install_middleware(app, interceptor_specs)  — attach pipeline to a FastAPI app
+    start_adapter_proxy(upstream_url, adapters) — host pipeline as a localhost uvicorn
+    AdapterPipeline                             — the async interceptor chain
+    InterceptorRegistry                         — name → class resolution
+"""
+
+from nemo_gym.adapters.middleware import install_middleware
+from nemo_gym.adapters.pipeline import AdapterPipeline
+from nemo_gym.adapters.proxy import ProxyHandle, start_adapter_proxy
+from nemo_gym.adapters.registry import InterceptorRegistry
+from nemo_gym.adapters.types import (
+    AdapterProxyConfig,
+    AdapterRequest,
+    AdapterResponse,
+    GracefulError,
+    InterceptorContext,
+    InterceptorSpec,
+    RequestInterceptor,
+    RequestToResponseInterceptor,
+    ResponseInterceptor,
+    Stage,
+)
+
+
+__all__ = [
+    "AdapterPipeline",
+    "AdapterProxyConfig",
+    "AdapterRequest",
+    "AdapterResponse",
+    "GracefulError",
+    "InterceptorContext",
+    "InterceptorRegistry",
+    "InterceptorSpec",
+    "ProxyHandle",
+    "RequestInterceptor",
+    "RequestToResponseInterceptor",
+    "ResponseInterceptor",
+    "Stage",
+    "install_middleware",
+    "start_adapter_proxy",
+]

nemo_gym/adapters/interceptors/__init__.py

@@ -0,0 +1,15 @@
+# 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.
+"""Built-in interceptors for the adapter pipeline."""

nemo_gym/adapters/interceptors/endpoint.py

@@ -0,0 +1,178 @@
+# 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 __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+from typing import Any
+
+import aiohttp  # type-name imports only (ClientTimeout, ClientError); HTTP calls go through nemo_gym.server_utils.request
+
+from nemo_gym.adapters.types import (
+    AdapterRequest,
+    AdapterResponse,
+    RequestToResponseInterceptor,
+)
+from nemo_gym.server_utils import request as global_request
+
+
+logger = logging.getLogger(__name__)
+
+
+class Interceptor(RequestToResponseInterceptor):
+    def __init__(
+        self,
+        *,
+        upstream_url: str,
+        api_key: str | None = None,
+        extra_body: dict[str, Any] | None = None,
+        request_timeout: float = 120,
+        max_retries: int = 0,
+        retry_on_status: list[int] | None = None,
+        max_concurrent: int = 64,
+    ) -> None:
+        clean = upstream_url.rstrip("/")
+        for suffix in ("/chat/completions", "/completions", "/embeddings"):
+            if clean.endswith(suffix):
+                clean = clean[: -len(suffix)]
+                break
+        # ``req.path`` already carries the ``/v1/...`` prefix, so strip a trailing
+        # ``/v1`` from the base to avoid ``.../v1/v1/...`` double-prefixing.
+        if clean.endswith("/v1"):
+            clean = clean[: -len("/v1")]
+        self._upstream_url = clean
+        self._api_key = api_key
+        self._extra_body = extra_body or {}
+        self._request_timeout = float(request_timeout)
+        self._timeout = aiohttp.ClientTimeout(total=request_timeout)
+        self._max_retries = max_retries
+        self._retry_on_status = set(retry_on_status or [429, 502, 503, 504])
+        # ``max_concurrent`` is config-shape compatibility only; connector
+        # limits live on the global aiohttp client.
+        self._max_concurrent = max_concurrent
+
+    async def intercept_request(
+        self,
+        req: AdapterRequest,
+    ) -> AdapterRequest | AdapterResponse:
+        url = f"{self._upstream_url}{req.path}"
+
+        body = {**req.body, **self._extra_body}
+        headers = {
+            k: v for k, v in req.headers.items() if k.lower() not in ("host", "content-length", "transfer-encoding")
+        }
+        if self._api_key:
+            headers["Authorization"] = f"Bearer {self._api_key}"
+        headers.setdefault("Content-Type", "application/json")
+
+        attempt = 0
+        while True:
+            t0 = time.perf_counter()
+            try:
+                resp = await global_request(
+                    method="POST",
+                    url=url,
+                    data=json.dumps(body),
+                    headers=headers,
+                    timeout=self._timeout,
+                )
+                async with resp:
+                    raw = await resp.read()
+                    latency = (time.perf_counter() - t0) * 1000
+                    # Iterate ``resp.headers.items()`` to preserve multi-valued
+                    # keys (e.g. Set-Cookie) that a plain ``dict()`` collapses.
+                    resp_headers: list[tuple[bytes, bytes]] = [
+                        (k.encode("latin-1"), v.encode("latin-1")) for k, v in resp.headers.items()
+                    ]
+                    status = resp.status
+
+                    if status in self._retry_on_status and attempt < self._max_retries:
+                        retry_after = resp.headers.get("Retry-After")
+                        try:
+                            # RFC 7231 allows an HTTP-date here; fall back on parse failure.
+                            delay = float(retry_after) if retry_after else min(2**attempt, 60)
+                        except ValueError:
+                            delay = min(2**attempt, 60)
+                        logger.warning(
+                            "endpoint: %s returned %d, retry %d/%d in %.1fs",
+                            url,
+                            status,
+                            attempt + 1,
+                            self._max_retries,
+                            delay,
+                        )
+                        attempt += 1
+                        await asyncio.sleep(delay)
+                        continue
+
+                    try:
+                        parsed = json.loads(raw)
+                    except (json.JSONDecodeError, UnicodeDecodeError):
+                        parsed = raw
+
+                    return AdapterResponse(
+                        status_code=status,
+                        headers=resp_headers,
+                        body=parsed,
+                        latency_ms=latency,
+                        ctx=req.ctx,
+                    )
+
+            except asyncio.TimeoutError:
+                latency = (time.perf_counter() - t0) * 1000
+                if attempt < self._max_retries:
+                    delay = min(2**attempt, 60)
+                    logger.warning(
+                        "endpoint: %s timed out, retry %d/%d in %.1fs",
+                        url,
+                        attempt + 1,
+                        self._max_retries,
+                        delay,
+                    )
+                    attempt += 1
+                    await asyncio.sleep(delay)
+                    continue
+                logger.error("endpoint: %s timed out after %d attempts", url, attempt + 1)
+                return AdapterResponse(
+                    status_code=504,
+                    headers={},
+                    body={
+                        "error": {"message": f"Upstream timed out after {self._request_timeout}s", "type": "timeout"}
+                    },
+                    latency_ms=latency,
+                    ctx=req.ctx,
+                )
+
+            except aiohttp.ClientError as exc:
+                latency = (time.perf_counter() - t0) * 1000
+                if attempt < self._max_retries:
+                    delay = min(2**attempt, 60)
+                    logger.warning(
+                        "endpoint: %s failed (%s), retry %d/%d in %.1fs",
+                        url,
+                        exc,
+                        attempt + 1,
+                        self._max_retries,
+                        delay,
+                    )
+                    attempt += 1
+                    await asyncio.sleep(delay)
+                    continue
+                raise
+
+    async def close(self) -> None:
+        return None

nemo_gym/adapters/interceptors/request_logging.py

@@ -0,0 +1,66 @@
+# 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 __future__ import annotations
+
+import json
+import logging
+
+from nemo_gym.adapters.types import (
+    AdapterRequest,
+    AdapterResponse,
+    RequestInterceptor,
+    ResponseInterceptor,
+)
+
+
+logger = logging.getLogger(__name__)
+
+_MAX = 512
+
+
+def _trunc_preview(obj: object) -> str:
+    if isinstance(obj, bytes):
+        text = obj.decode(errors="replace")
+    elif isinstance(obj, dict):
+        text = json.dumps(obj, default=str, ensure_ascii=False)
+    else:
+        text = str(obj)
+    if len(text) <= _MAX:
+        return text
+    return text[:_MAX] + "..."
+
+
+class Interceptor(RequestInterceptor, ResponseInterceptor):
+    best_effort = True
+
+    async def intercept_request(self, req: AdapterRequest) -> AdapterRequest:
+        keys = list(req.body.keys()) if isinstance(req.body, dict) else []
+        logger.info(
+            "request %s %s body_keys=%s body_preview=%s",
+            req.method,
+            req.path,
+            keys,
+            _trunc_preview(req.body),
+        )
+        return req
+
+    async def intercept_response(self, resp: AdapterResponse) -> AdapterResponse:
+        logger.info(
+            "response status=%d latency_ms=%.2f body_preview=%s",
+            resp.status_code,
+            resp.latency_ms,
+            _trunc_preview(resp.body),
+        )
+        return resp