[https://nvbugs/6781271][fix] Responses API: agent-client fixes and disaggregated serving

docs/source/features/disagg-serving.md

@@ -224,6 +224,18 @@ curl http://localhost:8000/v1/completions \
     }' -w "\n"
 ```
 
+#### Responses API
+
+The disaggregated server also serves `/v1/responses`, with both `context_first` and `generation_first` scheduling, and `/v1/models`, which it answers from a context worker. A streamed response carries the generation worker's events unchanged. Unlike a single `trtllm-serve` instance, the disaggregated server keeps no response history:
+
+- A request with `previous_response_id` is rejected with a 400, because a stored response lives in one worker's memory and a later request may reach another worker. Send the earlier turns in `input` instead.
+- `store` is ignored: nothing is persisted, and the server logs a warning once when a client asks for it.
+
+Two limitations apply to `/v1/responses` on every server, disaggregated or not:
+
+- `parallel_tool_calls: false`, and a `tool_choice` other than `auto` or `none` on non-Harmony models, are accepted but not enforced. The server logs a warning once.
+- An image in the input, in a message or in a tool output, is replaced by a text placeholder that names what was dropped, since the endpoint accepts text only.
+
 #### Launching disaggregated servers on SLURM clusters
 
 Please refer to [Disaggregated Inference Benchmark Scripts](source:examples/disaggregated/slurm).

docs/source/release-notes.md

@@ -6,8 +6,14 @@ All published functionality in the Release Notes has been fully tested and verif
 
 ## TensorRT-LLM Release 1.3
 
+### Key Features and Enhancements
+
+- Disaggregated serving supports the Responses API: `trtllm-serve disaggregated` serves `/v1/responses` with context-first and generation-first scheduling, and `/v1/models`. `previous_response_id` is rejected and `store` is ignored there, since no response history is kept. See [Disaggregated Serving](features/disagg-serving.md#responses-api).
+
 ### API Changes
 
+- `/v1/responses`: a response cut short by `max_output_tokens` now reports `incomplete_details.reason` `max_output_tokens`, and when streamed it ends with `response.incomplete` instead of `response.completed`, as in the OpenAI API.
+
 - NVFP4 cold-page compression remains opt-in and disabled by default. When enabled for DeepSeek-V4, NoPE uses NVFP4 and RoPE now defaults to 2FP4 (`nvfp4_residual_dim=64`). Set `skip_rope_quantization=True` to keep RoPE at its original precision, as before, or `nvfp4_residual_dim=0` to use single NVFP4. The active GPU KV-cache dtype is unchanged. See the [NVFP4 cold-page compression example](source:examples/kv_cache_compression/nvfp4_cold_page.md#deepseek-v4).
 
 - **[DEPRECATION]** The TRITON MoE backend (`TritonFusedMoE`, `moe_config.backend="TRITON"`) is deprecated as of TensorRT-LLM 1.3 (2026-09) and will be removed after the 3-month migration period. Its only remaining role is a modest performance edge for GPT-OSS on Hopper with `W4A16_MXFP4` — the single configuration `AUTO` resolves to TRITON, and the format an MXFP4 GPT-OSS checkpoint takes on SM90. As the model set and the supported platforms keep growing, a single-scenario MoE path is no longer worth its maintenance cost. `moe_config.backend="CUTLASS"` replaces it functionally on Hopper: it serves `W4A16_MXFP4` on SM90 along with the unquantized BF16 and FP8 per-tensor paths, and MoE backend resolution already degrades to it automatically when TRITON declines a layer. During the migration period TRITON keeps working and logs a one-time warning. See the [deprecation policy](https://github.com/NVIDIA/TensorRT-LLM#deprecation-policy).

tensorrt_llm/serve/openai_client.py

@@ -50,6 +50,8 @@
     ChatCompletionResponse,
     CompletionRequest,
     CompletionResponse,
+    ResponsesRequest,
+    ResponsesResponse,
     UCompletionRequest,
     UCompletionResponse,
 )
@@ -82,7 +84,20 @@
 # returned object to the type the caller asked for, so a mismatch is a type
 # error rather than an Any that silently propagates.
 class _SerializableRequest(Protocol):
-    def model_dump(self, *, mode: str = ..., exclude_unset: bool = ...) -> dict: ...
+    def model_dump(
+        self, *, mode: str = ..., exclude_unset: bool = ..., by_alias: bool = ...
+    ) -> dict: ...
+
+
+def _encode_request_body(request: _SerializableRequest) -> bytes:
+    """Encode a request forwarded to a worker.
+
+    Fields go out under their wire names, e.g. ``schema``, which the model
+    holds as ``schema_``.
+    """
+    return _msgpack_encoder.encode(
+        request.model_dump(mode="json", exclude_unset=True, by_alias=True)
+    )
 
 
 _ResponseT = TypeVar("_ResponseT", bound=BaseModel)
@@ -113,6 +128,10 @@ async def send_request(
                 hooks,
                 req_id,
             )
+        elif isinstance(request, ResponsesRequest):
+            return await self._send_request(
+                "v1/responses", request, ResponsesResponse, server, hooks, req_id
+            )
         else:
             raise ValueError(f"Invalid request type: {type(request)}")
 
@@ -153,6 +172,16 @@ async def post_json(
         """
         ...
 
+    @abstractmethod
+    async def get_json(
+        self,
+        endpoint: str,
+        response_type: Type[_ResponseT],
+        server: str,
+    ) -> _ResponseT:
+        """GET a worker endpoint, outside routing state, like post_json."""
+        ...
+
     async def shutdown(self) -> None: ...
 
     @abstractmethod
@@ -235,7 +264,7 @@ async def post_json(
         # same way -- hence the same msgpack transport as _send_request. The
         # worker decodes it because _MsgspecRoute is the app's route_class, so
         # it covers every route rather than just the completion ones.
-        body = _msgpack_encoder.encode(request.model_dump(mode="json", exclude_unset=True))
+        body = _encode_request_body(request)
         headers = dict(MSGPACK_HEADERS)
         async with self._session.post(url, data=body, headers=headers) as response:
             if response.status >= 400:
@@ -249,6 +278,26 @@ async def post_json(
                 )
             return response_type(**await response.json())
 
+    async def get_json(
+        self,
+        endpoint: str,
+        response_type: Type[_ResponseT],
+        server: str,
+    ) -> _ResponseT:
+        server_url = server if server.startswith("http") else f"http://{server}"
+        url = f"{server_url.rstrip('/')}/{endpoint}"
+        async with self._session.get(url) as response:
+            if response.status >= 400:
+                error_body = await response.text()
+                raise aiohttp.ClientResponseError(
+                    response.request_info,
+                    response.history,
+                    status=response.status,
+                    message=f"{response.reason}: {error_body[:2048]}",
+                    headers=response.headers,
+                )
+            return response_type(**await response.json())
+
     async def _send_request(
         self,
         endpoint: str,
@@ -390,7 +439,7 @@ async def _post_with_retry(
             # Serialize once on the orchestrator's single event-loop thread.
             # Content-Type stays application/json so FastAPI still routes the
             # body through Request.json(); the header picks the decoder.
-            body = _msgpack_encoder.encode(request.model_dump(mode="json", exclude_unset=True))
+            body = _encode_request_body(request)
             headers = dict(MSGPACK_HEADERS)
             if self._request_perf_metrics:
                 headers[RETURN_METRICS_HEADER] = "1"

tensorrt_llm/serve/openai_disagg_server.py

@@ -61,14 +61,16 @@
     OpenAIDisaggregatedService, ResponseHooks)
 from tensorrt_llm.serve.openai_protocol import (
     ChatCompletionRequest, ChatCompletionResponse, CompletionRequest,
-    ConversationParams, ErrorResponse, UCompletionRequest, UCompletionResponse,
-    ensure_request_chat_template_allowed)
+    ConversationParams, ErrorResponse, ResponsesRequest, UCompletionRequest,
+    UCompletionResponse, ensure_request_chat_template_allowed)
 from tensorrt_llm.serve.perf_metrics import (DisaggPerfMetricsCollector,
                                              PerfMetricsJsonlWriter,
                                              PerfMetricsMiddleware,
                                              combine_disagg_metrics)
 from tensorrt_llm.serve.responses_utils import (ServerArrivalTimeMiddleware,
-                                                get_steady_clock_now_in_seconds)
+                                                get_steady_clock_now_in_seconds,
+                                                guard_responses_stream,
+                                                stream_error_event)
 from tensorrt_llm.serve.router import Router
 from tensorrt_llm.usage import TerminalOutcome, record_termination_observation
 from tensorrt_llm.version import __version__ as VERSION
@@ -345,6 +347,8 @@ def register_routes(self):
         self.app.add_api_route("/v1/chat/completions", self._wrap_entry_point(self._service.openai_chat_completion, ChatCompletionRequest), methods=["POST"])
         self.app.add_api_route("/v1/messages", self.anthropic_messages, methods=["POST"])
         self.app.add_api_route("/v1/messages/count_tokens", self.anthropic_count_tokens, methods=["POST"])
+        self.app.add_api_route("/v1/responses", self._wrap_entry_point(self._service.openai_responses, ResponsesRequest), methods=["POST"])
+        self.app.add_api_route("/v1/models", self.get_model, methods=["GET"])
         self.app.add_api_route("/health", self.health, methods=["GET"])
         self.app.add_api_route("/cluster_info", self.cluster_info, methods=["GET"])
         self.app.add_api_route("/version", self.version, methods=["GET"])
@@ -437,10 +441,18 @@ async def wrapper(req: request_type, raw_req: Request) -> Response:
                 response_or_generator = await entry_point(req, hooks)
                 self._perf_metrics_collector.total_responses.inc()
                 if req.stream:
+                    stream = response_or_generator
+                    if isinstance(req, ResponsesRequest):
+                        # A relayed Responses stream that fails early ends with
+                        # an `error` event; other protocols end on a sentinel.
+                        stream = guard_responses_stream(stream,
+                                                        stream_error_event)
                     return StreamingResponse(
-                        content=response_or_generator,
+                        content=stream,
                         media_type="text/event-stream")
-                return JSONResponse(content=response_or_generator.model_dump())
+                # by_alias: e.g. text.format.schema goes out under its wire name.
+                return JSONResponse(
+                    content=response_or_generator.model_dump(by_alias=True))
             except Exception as e:
                 # Usually raises; returns a Response for worker errors that
                 # carry a machine-readable code (context_length_exceeded).
@@ -497,6 +509,26 @@ async def anthropic_messages(self, request: AnthropicMessagesRequest,
         return JSONResponse(content=anthropic_response.model_dump(
             exclude_none=True))
 
+    async def get_model(self) -> Response:
+        """List the served model, as a context worker reports it."""
+        try:
+            return JSONResponse(
+                content=(await self._service.get_model()).model_dump())
+        except aiohttp.ClientResponseError as error:
+            status, message = error.status or 500, _upstream_error_message(
+                error)
+        except (RuntimeError, ValueError) as error:
+            # No context worker yet; the client may retry.
+            status, message = 503, str(error)
+        return JSONResponse(
+            status_code=status,
+            content=ErrorResponse(
+                message=message,
+                type=("invalid_request_error"
+                      if 400 <= status < 500 else "api_error"),
+                code=status,
+            ).model_dump())
+
     async def anthropic_count_tokens(
             self, request: AnthropicCountTokensRequest) -> Response:
         """Count Anthropic input tokens on a context worker."""