Skip to content

feat(adapters): adapter middleware framework (base) #1518

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Glorf wants to merge 2 commits into from
Closed

feat(adapters): adapter middleware framework (base) #1518

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d7e5c20
feat(adapters): adapter middleware framework (base)
Glorf Jun 3, 2026
35a1b91
test(adapters): cover endpoint interceptor and adapter internals
Glorf Jun 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions fern/versions/latest/pages/model-server/adapters.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
Comment on lines +26 to +33

@ananthsub ananthsub Jun 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shouldnt have this in the docs


## 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>
6 changes: 6 additions & 0 deletions fern/versions/latest/pages/model-server/index.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,3 +32,9 @@ Self-hosted inference with vLLM for maximum control.
[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.
58 changes: 58 additions & 0 deletions nemo_gym/adapters/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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",
]
15 changes: 15 additions & 0 deletions nemo_gym/adapters/interceptors/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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."""
180 changes: 180 additions & 0 deletions nemo_gym/adapters/interceptors/endpoint.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,180 @@
# 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
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

@staticmethod
def _normalize_content(body: dict[str, Any]) -> None:
for choice in body.get("choices", []):
msg = choice.get("message") or choice.get("delta") or {}
if "content" in msg and msg["content"] is None:
msg["content"] = ""

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")
delay = float(retry_after) if retry_after else 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

if isinstance(parsed, dict):
self._normalize_content(parsed)

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
Loading
Loading