From b0c128cbf3e85cc6018a821fc4e3176363e9af4c Mon Sep 17 00:00:00 2001 From: mschwab Date: Fri, 17 Jul 2026 10:37:44 -0700 Subject: [PATCH 01/10] =?UTF-8?q?feat(agents):=20external=20NAT=20agents?= =?UTF-8?q?=20=E2=80=94=20backend=20(register,=20chat,=20eval,=20health)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a first-class "external" agent that runs outside NeMo Platform: registered by URL, discovered via its A2A card, reachable without a NAT deployment. Entity + create: - Agent gains source (managed|external), endpoint, card - POST /agents takes config XOR url; url fetches the A2A card (timeout + size-capped, generic error to avoid a port-scan oracle) and stores an external agent; structural URL validation A2A bridge (dependency-free JSON-RPC in a2a.py): - chat/completions <-> message/send + message/stream (streaming), reached via the Studio route and the CLI/SDK agent-name proxy (/-/v1/chat/completions) - NAT /generate/full <-> message/send, so external agents evaluate through the existing `nat eval` pipeline with unchanged eval configs - multi-turn: NAT's A2A executor is message-only, so the OpenAI history is folded into the message (single turn verbatim; multi-turn as a transcript) - byte-capped reads; A2A failures logged; targets the vetted registered endpoint, never the card's self-reported url Lifecycle + gating: - POST /agents/{name}/refresh (re-fetch card), GET /agents/{name}/reachability - external agents cannot be deployed (400); is_external_agent predicate ~100 unit tests (respx + real-socket); ruff + ty clean. Signed-off-by: mschwab --- plugins/nemo-agents/openapi/openapi.yaml | 132 +++++++- .../nemo-agents/src/nemo_agents_plugin/a2a.py | 302 ++++++++++++++++++ .../src/nemo_agents_plugin/api/v2/agents.py | 99 +++++- .../nemo_agents_plugin/api/v2/deployments.py | 15 +- .../src/nemo_agents_plugin/api/v2/gateway.py | 279 +++++++++++++++- .../src/nemo_agents_plugin/entities.py | 39 ++- .../src/nemo_agents_plugin/schema.py | 43 ++- plugins/nemo-agents/tests/unit/test_a2a.py | 200 ++++++++++++ .../nemo-agents/tests/unit/test_agents_api.py | 165 ++++++++++ .../tests/unit/test_deployments_api.py | 63 ++++ .../nemo-agents/tests/unit/test_gateway.py | 208 ++++++++++++ 11 files changed, 1515 insertions(+), 30 deletions(-) create mode 100644 plugins/nemo-agents/src/nemo_agents_plugin/a2a.py create mode 100644 plugins/nemo-agents/tests/unit/test_a2a.py create mode 100644 plugins/nemo-agents/tests/unit/test_deployments_api.py diff --git a/plugins/nemo-agents/openapi/openapi.yaml b/plugins/nemo-agents/openapi/openapi.yaml index ea7fead01a..ec3eb473e2 100644 --- a/plugins/nemo-agents/openapi/openapi.yaml +++ b/plugins/nemo-agents/openapi/openapi.yaml @@ -8,7 +8,11 @@ paths: tags: - Agents summary: Create Agent - description: Create a new agent from a NAT workflow config. + description: "Create an agent.\n\nManaged (``config`` supplied): stores the\ + \ NAT workflow the platform will run.\nExternal (``url`` supplied): fetches\ + \ the running agent's A2A card and stores\na pointer \u2014 the platform never\ + \ runs it. ``CreateAgentRequest`` enforces that\nexactly one of ``config``/``url``\ + \ is present." operationId: create_agent_apis_agents_v2_workspaces__workspace__agents_post parameters: - name: workspace @@ -154,6 +158,77 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/agents/v2/workspaces/{workspace}/agents/{name}/reachability: + get: + tags: + - Agents + summary: External Agent Reachability + description: Liveness probe for an external agent's endpoint (for a health badge). + operationId: external_agent_reachability_apis_agents_v2_workspaces__workspace__agents__name__reachability_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: name + in: path + required: true + schema: + type: string + title: Name + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/AgentReachability' + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + /apis/agents/v2/workspaces/{workspace}/agents/{name}/refresh: + post: + tags: + - Agents + summary: Refresh External Agent + description: 'Re-fetch an external agent''s A2A card and update the stored copy. + + + The card is captured once at registration; this refreshes it (e.g. after the + + agent''s skills change). Returns 502 if the agent can''t be reached.' + operationId: refresh_external_agent_apis_agents_v2_workspaces__workspace__agents__name__refresh_post + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: name + in: path + required: true + schema: + type: string + title: Name + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/Agent' + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /apis/agents/v2/workspaces/{workspace}/deployments: post: tags: @@ -2281,6 +2356,7 @@ components: type: object title: Config description: Agent config dict interpreted according to config_format. + Empty for external agents. config_format: type: string title: Config Format @@ -2288,6 +2364,29 @@ components: `nat-workflow-v1` is the default legacy NAT workflow format; `nemo-agents-spec-v1` identifies the Platform-owned agent.yaml spec format. default: nat-workflow-v1 + source: + type: string + enum: + - managed + - external + title: Source + description: '''managed'' (default) agents are deployed and run by NeMo + Platform. ''external'' agents run outside the platform; NMP holds only + a pointer to them.' + default: managed + endpoint: + type: string + title: Endpoint + description: Base URL of an external agent (e.g. http://host:10000). Empty + for managed agents, whose runtime address lives on the AgentDeployment + instead. + default: '' + card: + additionalProperties: true + type: object + title: Card + description: A2A agent card fetched from an external agent at registration + (name, description, skills). Empty for managed agents. id: type: string title: Id @@ -2475,6 +2574,17 @@ components: \ ``agent_deployment``\nLifecycle: pending \u2192 starting \u2192 running\ \ | failed.\nThe :class:`~nemo_agents_plugin.runner.controller.AgentDeploymentController`\n\ drives state transitions by reconciling this entity against the\n:class:`~nemo_agents_plugin.runner.backend.RunnerBackend`." + AgentReachability: + properties: + reachable: + type: boolean + title: Reachable + description: Whether the external agent's endpoint answered a liveness probe. + type: object + required: + - reachable + title: AgentReachability + description: Response for ``GET /v2/workspaces/{workspace}/agents/{name}/reachability``. AnalyzeBatchConfig: properties: batch: @@ -2653,24 +2763,34 @@ components: description: type: string title: Description - description: Human-readable description. + description: Human-readable description. For external agents, falls back + to the agent card's description. default: '' config: + title: Config + description: NAT workflow config dict. Required for a managed agent; omit + for external. additionalProperties: true type: object - title: Config - description: Agent config dict interpreted according to config_format. config_format: type: string title: Config Format description: Config format identifier. default: nat-workflow-v1 + url: + title: Url + description: Base URL of a running external agent (e.g. http://host:10000). + Provide instead of config. + type: string type: object required: - name - - config title: CreateAgentRequest - description: Request body for ``POST /v2/workspaces/{workspace}/agents``. + description: "Request body for ``POST /v2/workspaces/{workspace}/agents``.\n\ + \nCreates either a **managed** agent (supply ``config`` \u2014 a NAT workflow\ + \ the\nplatform will run) or an **external** agent (supply ``url`` \u2014\ + \ a pointer to\na NAT agent already running elsewhere, whose A2A card the\ + \ platform fetches\nat creation). Provide exactly one of ``config`` or ``url``." CreateDeploymentRequest: properties: agent: diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py new file mode 100644 index 0000000000..023dea2277 --- /dev/null +++ b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py @@ -0,0 +1,302 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""A2A (Agent-to-Agent) discovery helpers. + +An externally-running NAT agent served with ``nat a2a serve`` publishes an +*agent card* at a well-known path. The card describes the agent (name, +description) and its skills (one per workflow function). We fetch it at +registration time so the platform can list and visualize an agent it does not +run. See https://github.com/nvidia/nemo-agent-toolkit A2A server docs. +""" + +from __future__ import annotations + +import json +import logging +from collections.abc import AsyncIterator +from typing import Any +from urllib.parse import urljoin +from uuid import uuid4 + +import httpx + +logger = logging.getLogger(__name__) + +# Well-known paths that expose an A2A agent card. The spec moved from +# ``agent.json`` to ``agent-card.json``; try the current name first. +AGENT_CARD_PATHS = ("/.well-known/agent-card.json", "/.well-known/agent.json") + +# Cap the card body so a hostile/misconfigured endpoint can't stream us an +# unbounded response. Cards are a few KB in practice. +_MAX_CARD_BYTES = 512 * 1024 +_FETCH_TIMEOUT_S = 10.0 + +# Agents can take a while to answer; allow more headroom than card discovery. +_MESSAGE_TIMEOUT_S = 120.0 + +# Liveness probe is a quick "is the well-known path answering" check. +_PROBE_TIMEOUT_S = 5.0 + +# Cap message/stream bodies too — replies are larger than cards but still bounded, +# so one hostile endpoint can't OOM the shared gateway worker. +_MAX_MESSAGE_BYTES = 8 * 1024 * 1024 + + +class AgentCardError(Exception): + """Raised when an external agent's card can't be fetched or parsed.""" + + +class A2AMessageError(Exception): + """Raised when a ``message/send`` call to an external agent fails.""" + + +async def _read_capped(response: httpx.Response, cap: int) -> bytes: + """Read a streamed response body, aborting once *cap* bytes are exceeded. + + Reads incrementally (unlike ``response.content``, which buffers the whole + decoded body first) so an oversized/compressed body can't exhaust memory + before the size check runs. + """ + chunks: list[bytes] = [] + total = 0 + async for chunk in response.aiter_bytes(): + total += len(chunk) + if total > cap: + raise A2AMessageError("agent response exceeded the size limit") + chunks.append(chunk) + return b"".join(chunks) + + +def _card_url(base_url: str, path: str) -> str: + # urljoin needs a trailing slash on the base to preserve any path prefix, + # and the well-known path is absolute, so join against the origin. + return urljoin(base_url if base_url.endswith("/") else base_url + "/", path.lstrip("/")) + + +async def fetch_agent_card(base_url: str) -> dict[str, Any]: + """Fetch and parse an external agent's A2A card. + + Tries each well-known path in order. Raises :class:`AgentCardError` with a + user-facing message if the endpoint is unreachable, returns non-JSON, or + exposes no recognizable card. The returned dict is the raw card. + """ + base = base_url.strip() + if not base.startswith(("http://", "https://")): + raise AgentCardError("Endpoint must be an http(s) URL.") + + # Track the specific reason for logging, but never surface it to the caller: + # distinct "could not reach" vs "HTTP 401" vs "not JSON" messages would let a + # caller probe which internal hosts/ports exist (an SSRF oracle). + last_error = "no agent card found" + async with httpx.AsyncClient(timeout=_FETCH_TIMEOUT_S, follow_redirects=True) as client: + for path in AGENT_CARD_PATHS: + url = _card_url(base, path) + try: + async with client.stream("GET", url, headers={"Accept": "application/json"}) as resp: + if resp.status_code != 200: + last_error = f"HTTP {resp.status_code} from {path}" + continue + try: + raw = await _read_capped(resp, _MAX_CARD_BYTES) + except A2AMessageError: + last_error = "card response exceeded the size limit" + continue + except httpx.HTTPError as exc: + last_error = f"transport error ({exc.__class__.__name__})" + continue + try: + card = json.loads(raw) + except ValueError: + last_error = "card response was not valid JSON" + continue + if not isinstance(card, dict) or not (card.get("name") or card.get("skills")): + last_error = "response did not look like an A2A agent card" + continue + return card + + logger.info("Agent card fetch failed for %s: %s", _endpoint_host(base), last_error) + raise AgentCardError("Could not fetch a valid A2A agent card from the provided URL.") + + +def _endpoint_host(url: str) -> str: + """Host of *url* for logging — avoid echoing the full URL back anywhere.""" + from urllib.parse import urlparse + + try: + return urlparse(url).hostname or "unknown" + except ValueError: + return "unknown" + + +async def probe_agent_reachable(base_url: str) -> bool: + """Cheap liveness check: does the agent's well-known card path answer 200? + + Short timeout, no redirects, no body parsing — just a reachability signal for + the UI. Never raises; returns False on any failure. + """ + base = base_url.strip() + if not base.startswith(("http://", "https://")): + return False + try: + async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT_S) as client: + for path in AGENT_CARD_PATHS: + try: + resp = await client.get(_card_url(base, path)) + except httpx.HTTPError: + continue + if resp.status_code == 200: + return True + except httpx.HTTPError: + return False + return False + + +def _collect_text_parts(parts: Any) -> list[str]: + """Pull the text out of an A2A ``parts`` array, ignoring non-text parts.""" + out: list[str] = [] + if isinstance(parts, list): + for part in parts: + if isinstance(part, dict) and isinstance(part.get("text"), str) and part["text"]: + out.append(part["text"]) + return out + + +def extract_message_text(result: Any) -> str: + """Extract assistant text from an A2A ``message/send`` result. + + The result is either a Message (``parts`` directly) or a Task (text lives in + ``artifacts[].parts`` and/or ``status.message.parts``). Concatenate whatever + text parts are present; return "" if none. + """ + if not isinstance(result, dict): + return "" + texts: list[str] = [] + texts += _collect_text_parts(result.get("parts")) + artifacts = result.get("artifacts") + if isinstance(artifacts, list): + for artifact in artifacts: + if isinstance(artifact, dict): + texts += _collect_text_parts(artifact.get("parts")) + status = result.get("status") + if isinstance(status, dict) and isinstance(status.get("message"), dict): + texts += _collect_text_parts(status["message"].get("parts")) + return "\n".join(texts) + + +async def send_a2a_message(endpoint: str, text: str) -> str: + """Send *text* to an external A2A agent via ``message/send`` and return its reply. + + Speaks A2A JSON-RPC 2.0 directly (no A2A client dependency). Raises + :class:`A2AMessageError` on transport failure, a JSON-RPC error, or a + non-JSON response. + """ + payload = { + "jsonrpc": "2.0", + "id": uuid4().hex, + "method": "message/send", + "params": { + "message": { + "role": "user", + "parts": [{"kind": "text", "text": text}], + "messageId": uuid4().hex, + "kind": "message", + } + }, + } + try: + async with httpx.AsyncClient(timeout=_MESSAGE_TIMEOUT_S, follow_redirects=True) as client: + async with client.stream("POST", endpoint, json=payload, headers={"Accept": "application/json"}) as resp: + if resp.status_code != 200: + raise A2AMessageError(f"agent returned HTTP {resp.status_code}") + raw = await _read_capped(resp, _MAX_MESSAGE_BYTES) + except httpx.HTTPError as exc: + raise A2AMessageError(f"could not reach agent at {endpoint} ({exc.__class__.__name__})") from exc + + try: + data = json.loads(raw) + except ValueError as exc: + raise A2AMessageError("agent response was not valid JSON") from exc + + if isinstance(data, dict) and data.get("error"): + err = data["error"] + msg = err.get("message") if isinstance(err, dict) else str(err) + raise A2AMessageError(f"agent error: {msg}") + + return extract_message_text(data.get("result") if isinstance(data, dict) else None) + + +def extract_stream_delta(result: Any) -> str: + """Extract the text delta from a single A2A ``message/stream`` event result. + + Streaming events are Messages (``parts``), artifact updates (``artifact.parts``), + or status updates (skipped — usually just lifecycle state). Returns "" for + events carrying no text. + """ + if not isinstance(result, dict): + return "" + artifact = result.get("artifact") + if isinstance(artifact, dict): + return "".join(_collect_text_parts(artifact.get("parts"))) + if result.get("kind") == "message" or "parts" in result: + return "".join(_collect_text_parts(result.get("parts"))) + return "" + + +async def stream_a2a_message(endpoint: str, text: str) -> AsyncIterator[str]: + """Stream an external A2A agent's reply via ``message/stream``, yielding text deltas. + + Speaks A2A JSON-RPC over SSE. Agents that stream tokens (artifact-update + deltas) yield incrementally; agents that buffer (e.g. NAT react_agent) yield + a single final chunk. Raises :class:`A2AMessageError` on transport/JSON-RPC + failure before the first token; mid-stream transport drops end the iterator. + """ + payload = { + "jsonrpc": "2.0", + "id": uuid4().hex, + "method": "message/stream", + "params": { + "message": { + "role": "user", + "parts": [{"kind": "text", "text": text}], + "messageId": uuid4().hex, + "kind": "message", + } + }, + } + try: + async with httpx.AsyncClient(timeout=_MESSAGE_TIMEOUT_S, follow_redirects=True) as client: + async with client.stream("POST", endpoint, json=payload, headers={"Accept": "text/event-stream"}) as resp: + if resp.status_code != 200: + raise A2AMessageError(f"agent returned HTTP {resp.status_code}") + # Cap raw bytes, not aiter_lines() output: aiter_lines buffers a + # line internally until a newline, so a newline-less body would + # grow unbounded before any per-line check. Split lines ourselves + # from a byte-capped buffer instead. + total = 0 + buffer = "" + async for chunk in resp.aiter_bytes(): + total += len(chunk) + if total > _MAX_MESSAGE_BYTES: + raise A2AMessageError("agent stream exceeded the size limit") + buffer += chunk.decode("utf-8", errors="replace") + while "\n" in buffer: + line, buffer = buffer.split("\n", 1) + line = line.strip() + if not line or line.startswith(":"): # keepalive / comment + continue + if not line.startswith("data:"): + continue + try: + data = json.loads(line[len("data:") :].strip()) + except ValueError: + continue + if isinstance(data, dict) and data.get("error"): + err = data["error"] + msg = err.get("message") if isinstance(err, dict) else str(err) + raise A2AMessageError(f"agent error: {msg}") + delta = extract_stream_delta(data.get("result") if isinstance(data, dict) else None) + if delta: + yield delta + except httpx.HTTPError as exc: + raise A2AMessageError(f"could not reach agent at {endpoint} ({exc.__class__.__name__})") from exc diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py index 06cb89091d..bf82ac1f22 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py @@ -13,13 +13,15 @@ import logging from fastapi import APIRouter, Depends, HTTPException, Query +from nemo_agents_plugin.a2a import AgentCardError, fetch_agent_card, probe_agent_reachable from nemo_agents_plugin.api.v2._perms import AgentPerms from nemo_agents_plugin.api.v2.dependencies import get_entity_client from nemo_agents_plugin.authz import scope -from nemo_agents_plugin.entities import Agent, AgentDeployment +from nemo_agents_plugin.entities import Agent, AgentDeployment, is_external_agent from nemo_agents_plugin.schema import ( AgentFilter, AgentPage, + AgentReachability, CreateAgentRequest, ) from nemo_platform_plugin.api.filters import make_filter_obj_dep @@ -50,14 +52,35 @@ async def create_agent( body: CreateAgentRequest, entity_client: NemoEntitiesClient = Depends(get_entity_client), ) -> Agent: - """Create a new agent from a NAT workflow config.""" - agent = Agent( - name=body.name, - workspace=workspace, - description=body.description, - config=body.config, - config_format=body.config_format, - ) + """Create an agent. + + Managed (``config`` supplied): stores the NAT workflow the platform will run. + External (``url`` supplied): fetches the running agent's A2A card and stores + a pointer — the platform never runs it. ``CreateAgentRequest`` enforces that + exactly one of ``config``/``url`` is present. + """ + if body.url: + try: + card = await fetch_agent_card(body.url) + except AgentCardError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + card_description = card.get("description") if isinstance(card.get("description"), str) else "" + agent = Agent( + name=body.name, + workspace=workspace, + description=body.description or card_description or "", + source="external", + endpoint=body.url.strip(), + card=card, + ) + else: + agent = Agent( + name=body.name, + workspace=workspace, + description=body.description, + config=body.config or {}, + config_format=body.config_format, + ) try: saved = await entity_client.create(agent) except NemoEntityConflictError as exc: @@ -71,6 +94,64 @@ async def create_agent( return saved +async def _get_external_agent(name: str, workspace: str, entity_client: NemoEntitiesClient) -> Agent: + """Fetch an agent and require it to be external, else raise 404/400.""" + try: + agent = await entity_client.get(Agent, name=name, workspace=workspace) + except NemoEntityNotFoundError as exc: + raise HTTPException(status_code=404, detail=f"Agent '{name}' not found in workspace '{workspace}'.") from exc + if not is_external_agent(agent): + raise HTTPException(status_code=400, detail=f"Agent '{name}' is not an external agent.") + return agent + + +@router.post("/agents/{name}/refresh", response_model=Agent, tags=["Agents"]) +@scope.write +@path_rule( + callers=[CallerKind.PRINCIPAL], + permissions=[AgentPerms.CREATE], +) +async def refresh_external_agent( + workspace: str, + name: str, + entity_client: NemoEntitiesClient = Depends(get_entity_client), +) -> Agent: + """Re-fetch an external agent's A2A card and update the stored copy. + + The card is captured once at registration; this refreshes it (e.g. after the + agent's skills change). Returns 502 if the agent can't be reached. + """ + agent = await _get_external_agent(name, workspace, entity_client) + try: + card = await fetch_agent_card(agent.endpoint) + except AgentCardError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + + agent.card = card + try: + saved = await entity_client.update(agent) + except Exception as exc: + logger.exception("Failed to refresh external agent '%s'", name) + raise HTTPException(status_code=500, detail="Failed to refresh external agent.") from exc + return saved + + +@router.get("/agents/{name}/reachability", response_model=AgentReachability, tags=["Agents"]) +@scope.read +@path_rule( + callers=[CallerKind.PRINCIPAL], + permissions=[AgentPerms.READ], +) +async def external_agent_reachability( + workspace: str, + name: str, + entity_client: NemoEntitiesClient = Depends(get_entity_client), +) -> AgentReachability: + """Liveness probe for an external agent's endpoint (for a health badge).""" + agent = await _get_external_agent(name, workspace, entity_client) + return AgentReachability(reachable=await probe_agent_reachable(agent.endpoint)) + + @router.get("/agents", response_model=AgentPage, tags=["Agents"]) @scope.read @path_rule( diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py index d2c674ee2a..ba5f12a895 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py @@ -23,7 +23,12 @@ from nemo_agents_plugin.api.v2._perms import DeploymentPerms from nemo_agents_plugin.api.v2.dependencies import get_entity_client from nemo_agents_plugin.authz import scope -from nemo_agents_plugin.entities import Agent, AgentDeployment, is_container_deployment_mode +from nemo_agents_plugin.entities import ( + Agent, + AgentDeployment, + is_container_deployment_mode, + is_external_agent, +) from nemo_agents_plugin.schema import ( CreateDeploymentRequest, DeploymentFilter, @@ -70,6 +75,14 @@ async def create_deployment( logger.exception("Failed to look up agent '%s'", body.agent) raise HTTPException(status_code=500, detail="Failed to look up agent.") from exc + # External agents run outside NeMo Platform; there is nothing for the + # platform to deploy. Reject rather than spawn a doomed empty-config process. + if is_external_agent(agent): + raise HTTPException( + status_code=400, + detail=f"Agent '{body.agent}' is external and runs outside NeMo Platform; it cannot be deployed.", + ) + # 2. Build deployment name (auto-generate if not provided) deployment_name = body.name or f"{body.agent}-{secrets.token_hex(4)}" diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py index 99a0f485f2..741e6a86a9 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py @@ -27,16 +27,24 @@ import json import logging import os -from typing import AsyncIterator +import time +from typing import Any, AsyncIterator from urllib.parse import urljoin, urlparse, urlunparse +from uuid import uuid4 import httpx from fastapi import APIRouter, Depends, HTTPException, Request -from fastapi.responses import StreamingResponse +from fastapi.responses import JSONResponse, StreamingResponse +from nemo_agents_plugin.a2a import A2AMessageError, send_a2a_message, stream_a2a_message from nemo_agents_plugin.api.v2._perms import GatewayPerms from nemo_agents_plugin.api.v2.dependencies import get_entity_client from nemo_agents_plugin.authz import scope -from nemo_agents_plugin.entities import Agent, AgentDeployment, is_container_deployment_mode +from nemo_agents_plugin.entities import ( + Agent, + AgentDeployment, + is_container_deployment_mode, + is_external_agent, +) from nemo_platform_plugin.authz import CallerKind, path_rule from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityNotFoundError @@ -124,12 +132,24 @@ async def _serve_agent_proxy( trailing_uri: str, request: Request, entity_client: NemoEntitiesClient, -) -> StreamingResponse: - """Find the first ``running`` deployment for the named agent and forward the request to it. +) -> StreamingResponse | JSONResponse: + """Forward a request addressed by agent name to the agent behind it. - Returns ``503`` if no running deployment is found. Shared by the read/write route handlers, - which differ only in their authorization scope (``agents:read`` vs ``agents:write``). + Managed agents proxy to their first ``running`` deployment (``503`` if none). + External agents have no deployment; their chat/completions and NAT ``generate`` + traffic is bridged to A2A instead. Shared by the read/write route handlers, + which differ only in authorization scope. """ + try: + agent = await entity_client.get(Agent, name=name, workspace=workspace) + except NemoEntityNotFoundError as exc: + raise HTTPException(status_code=404, detail=f"Agent '{name}' not found in workspace '{workspace}'.") from exc + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) from exc + + if is_external_agent(agent): + return await _serve_external_agent(name, trailing_uri, request, agent) + endpoint = await _resolve_agent_endpoint(name, workspace, entity_client) return await _proxy(request, endpoint, trailing_uri, model_name=name) @@ -139,6 +159,7 @@ async def _serve_agent_proxy( methods=_PROXY_READ_METHODS, tags=["Agent Gateway"], include_in_schema=False, + response_model=None, ) @scope.read @path_rule( @@ -151,7 +172,7 @@ async def proxy_by_agent_name_read( trailing_uri: str, request: Request, entity_client: NemoEntitiesClient = Depends(get_entity_client), -) -> StreamingResponse: +) -> StreamingResponse | JSONResponse: """Read-scoped (GET/HEAD/OPTIONS) proxy to the active deployment for *agent name*.""" return await _serve_agent_proxy(workspace, name, trailing_uri, request, entity_client) @@ -161,6 +182,7 @@ async def proxy_by_agent_name_read( methods=_PROXY_WRITE_METHODS, tags=["Agent Gateway"], include_in_schema=False, + response_model=None, ) @scope.write @path_rule( @@ -173,7 +195,7 @@ async def proxy_by_agent_name_write( trailing_uri: str, request: Request, entity_client: NemoEntitiesClient = Depends(get_entity_client), -) -> StreamingResponse: +) -> StreamingResponse | JSONResponse: """Write-scoped (POST/PUT/PATCH/DELETE) proxy to the active deployment for *agent name*.""" return await _serve_agent_proxy(workspace, name, trailing_uri, request, entity_client) @@ -416,3 +438,242 @@ async def _buffered() -> AsyncIterator[bytes]: headers={k: v for k, v in response_headers.items() if k.lower() != "content-length"}, media_type=content_type, ) + + +# --------------------------------------------------------------------------- +# External agent chat — OpenAI chat/completions <-> A2A message/send bridge +# --------------------------------------------------------------------------- +# +# External agents run outside NeMo Platform and speak A2A JSON-RPC, not the +# OpenAI chat API the Studio chat playground uses. This route accepts an +# OpenAI chat/completions request, forwards the latest user turn to the agent +# via ``message/send``, and returns the reply in OpenAI shape (SSE when the +# client asked to stream). Managed agents keep using the deployment proxy above. + + +def _message_text(message: Any) -> str: + """Extract text from one OpenAI message (``content`` may be str or parts).""" + if not isinstance(message, dict): + return "" + content = message.get("content") + if isinstance(content, str): + return content + if isinstance(content, list): + return "".join(part["text"] for part in content if isinstance(part, dict) and isinstance(part.get("text"), str)) + return "" + + +def _conversation_prompt(messages: Any) -> str: + """Build a single A2A message that carries the conversation so far. + + NAT's A2A executor is message-only (no server-side ``contextId`` memory), so + multi-turn context can't be threaded via the protocol — the client (OpenAI + chat) sends the full history each turn, and we fold prior user/assistant + turns into a transcript ahead of the latest user message. A single turn is + sent verbatim so simple agents aren't handed a transcript wrapper. + """ + if not isinstance(messages, list): + return "" + turns = [ + (m.get("role"), _message_text(m)) + for m in messages + if isinstance(m, dict) and m.get("role") in ("user", "assistant") + ] + turns = [(role, text) for role, text in turns if text] + if not turns: + return "" + if len(turns) == 1: + return turns[0][1] + + *prior, (_, latest) = turns + transcript = "\n".join(f"{'User' if role == 'user' else 'Assistant'}: {text}" for role, text in prior) + return f"Continue this conversation.\n\n{transcript}\n\nUser: {latest}" + + +def _openai_completion(text: str, model: str) -> dict[str, Any]: + return { + "id": f"chatcmpl-{uuid4().hex}", + "object": "chat.completion", + "created": int(time.time()), + "model": model, + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": text}, + "finish_reason": "stop", + } + ], + } + + +def _openai_chunk(completion_id: str, created: int, model: str, delta: dict[str, Any], finish: str | None) -> bytes: + payload = { + "id": completion_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{"index": 0, "delta": delta, "finish_reason": finish}], + } + return f"data: {json.dumps(payload)}\n\n".encode() + + +async def _stream_openai_from_a2a(endpoint: str, text: str, model: str) -> AsyncIterator[bytes]: + """Stream an external agent's A2A reply as OpenAI chat.completion.chunk SSE. + + Forwards each A2A text delta as a content chunk. A mid-stream error is + surfaced as content (the HTTP status is already 200 once streaming starts), + so the user sees it in the chat rather than a silent stall. + """ + created = int(time.time()) + completion_id = f"chatcmpl-{uuid4().hex}" + role_sent = False + try: + async for delta in stream_a2a_message(endpoint, text): + payload = {"content": delta} if role_sent else {"role": "assistant", "content": delta} + role_sent = True + yield _openai_chunk(completion_id, created, model, payload, None) + if not role_sent: + yield _openai_chunk(completion_id, created, model, {"role": "assistant", "content": ""}, None) + except A2AMessageError as exc: + note = f"[external agent error: {exc}]" + payload = {"content": f"\n{note}"} if role_sent else {"role": "assistant", "content": note} + yield _openai_chunk(completion_id, created, model, payload, None) + yield _openai_chunk(completion_id, created, model, {}, "stop") + yield b"data: [DONE]\n\n" + + +@router.post( + "/agents/{name}/chat/completions", + tags=["Agent Gateway"], + include_in_schema=False, + response_model=None, +) +@scope.write +@path_rule( + callers=[CallerKind.PRINCIPAL], + permissions=[GatewayPerms.INVOKE], +) +async def external_agent_chat_completions( + workspace: str, + name: str, + request: Request, + entity_client: NemoEntitiesClient = Depends(get_entity_client), +) -> StreamingResponse | JSONResponse: + """Bridge an OpenAI chat/completions call to an external agent's A2A endpoint.""" + try: + agent = await entity_client.get(Agent, name=name, workspace=workspace) + except NemoEntityNotFoundError as exc: + raise HTTPException(status_code=404, detail=f"Agent '{name}' not found in workspace '{workspace}'.") from exc + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) from exc + + if not is_external_agent(agent): + raise HTTPException( + status_code=400, + detail=f"Agent '{name}' is managed; chat through its deployment, not this endpoint.", + ) + + endpoint = _external_a2a_endpoint(agent, name) + body = await _read_json_object(request) + return await _external_openai_chat(name, endpoint, body) + + +# --------------------------------------------------------------------------- +# External agent bridge — OpenAI chat/completions and NAT /generate <-> A2A +# --------------------------------------------------------------------------- +# +# External agents have no deployment, so requests addressed to the agent-name +# proxy (``/agents/{name}/-/...``) land in :func:`_serve_external_agent`. Two +# invocation shapes are bridged to A2A: OpenAI ``chat/completions`` (Studio, the +# CLI ``nemo agents invoke``, and the SDK) and NAT ``generate/full`` (``nat eval``). + + +def _endpoint_host(url: str) -> str: + """Host of *url* for logging — never the full URL (may carry embedded creds).""" + try: + return urlparse(url).hostname or "unknown" + except ValueError: + return "unknown" + + +def _external_a2a_endpoint(agent: Agent, name: str) -> str: + """Return the vetted URL to reach an external agent. + + Always the registered ``endpoint`` — never the card's self-reported ``url``, + which is remote-controlled content and would let a registered agent redirect + all subsequent traffic to an arbitrary (e.g. internal) host. + """ + if not agent.endpoint: + raise HTTPException(status_code=400, detail=f"External agent '{name}' has no endpoint.") + return agent.endpoint + + +async def _read_json_object(request: Request) -> dict[str, Any]: + try: + body = await request.json() + except Exception: + return {} + return body if isinstance(body, dict) else {} + + +async def _external_openai_chat(name: str, endpoint: str, body: dict[str, Any]) -> StreamingResponse | JSONResponse: + """Translate an OpenAI chat/completions body to A2A and return OpenAI shape.""" + text = _conversation_prompt(body.get("messages")) + if not text: + raise HTTPException(status_code=400, detail="Request has no user message to send.") + + model = body.get("model") or name + + # Streaming: forward A2A deltas as they arrive (message/stream), so the UI + # shows progress instead of freezing until the agent finishes. + if body.get("stream"): + return StreamingResponse(_stream_openai_from_a2a(endpoint, text, model), media_type="text/event-stream") + + # Non-streaming: single message/send, returned as one OpenAI completion. + try: + reply = await send_a2a_message(endpoint, text) + except A2AMessageError as exc: + logger.warning("External agent '%s' chat failed (%s): %s", name, _endpoint_host(endpoint), exc) + raise HTTPException(status_code=502, detail=f"External agent chat failed: {exc}") from exc + return JSONResponse(_openai_completion(reply, model)) + + +async def _generate_full_stream(text: str) -> AsyncIterator[bytes]: + # nat eval joins the ``value`` fields of every ``data:`` line; one line is enough. + yield f"data: {json.dumps({'value': text})}\n\n".encode() + + +async def _serve_external_generate_full(name: str, endpoint: str, request: Request) -> StreamingResponse: + body = await _read_json_object(request) + question = body.get("input_message") + if not isinstance(question, str) or not question: + raise HTTPException(status_code=400, detail="Request has no 'input_message' to send.") + try: + reply = await send_a2a_message(endpoint, question) + except A2AMessageError as exc: + logger.warning("External agent '%s' generate failed (%s): %s", name, _endpoint_host(endpoint), exc) + raise HTTPException(status_code=502, detail=f"External agent generate failed: {exc}") from exc + return StreamingResponse(_generate_full_stream(reply), media_type="text/event-stream") + + +async def _serve_external_agent( + name: str, trailing_uri: str, request: Request, agent: Agent +) -> StreamingResponse | JSONResponse: + """Bridge an agent-name proxy request for an external agent to A2A. + + ``chat/completions`` (Studio / CLI invoke / SDK) → OpenAI↔A2A translation; + ``generate`` (nat eval) → NAT-generate↔A2A translation. Anything else 400s — + external agents have no deployment to proxy arbitrary paths to. + """ + endpoint = _external_a2a_endpoint(agent, name) + if trailing_uri.endswith("chat/completions"): + return await _external_openai_chat(name, endpoint, await _read_json_object(request)) + if trailing_uri.startswith("generate"): + return await _serve_external_generate_full(name, endpoint, request) + raise HTTPException( + status_code=400, + detail=( + f"External agent '{name}' supports chat/completions and generate only; " + f"'{trailing_uri}' is not available (the agent runs outside NeMo Platform)." + ), + ) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/entities.py b/plugins/nemo-agents/src/nemo_agents_plugin/entities.py index 5203b315ba..6d6fb62909 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/entities.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/entities.py @@ -19,6 +19,12 @@ DeploymentStatus = Literal["pending", "starting", "running", "failed", "deleting"] +# Where an agent's runtime lives. ``managed`` agents are run by NeMo Platform +# (config compiled to a ``nat serve`` deployment). ``external`` agents run +# outside the platform; NMP only holds a pointer (``endpoint``) plus the A2A +# agent card fetched at registration time — it never spawns them. +AgentSource = Literal["managed", "external"] + # Runtime backend for an AgentDeployment. ``subprocess`` (the default) runs the # agent as a local ``nat serve`` process reachable on a loopback ``endpoint``. # ``docker``/``k8s`` run the agent as a durable container deployment via the @@ -139,7 +145,7 @@ class Agent(NemoEntity, entity_type="agent"): description: str = Field(default="", description="Human-readable description of the agent.") config: dict[str, Any] = Field( default_factory=dict, - description="Agent config dict interpreted according to config_format.", + description="Agent config dict interpreted according to config_format. Empty for external agents.", ) config_format: str = Field( default=NAT_WORKFLOW_CONFIG_FORMAT, @@ -149,6 +155,37 @@ class Agent(NemoEntity, entity_type="agent"): "`nemo-agents-spec-v1` identifies the Platform-owned agent.yaml spec format." ), ) + source: AgentSource = Field( + default="managed", + description=( + "'managed' (default) agents are deployed and run by NeMo Platform. " + "'external' agents run outside the platform; NMP holds only a pointer to them." + ), + ) + endpoint: str = Field( + default="", + description=( + "Base URL of an external agent (e.g. http://host:10000). Empty for managed " + "agents, whose runtime address lives on the AgentDeployment instead." + ), + ) + card: dict[str, Any] = Field( + default_factory=dict, + description=( + "A2A agent card fetched from an external agent at registration " + "(name, description, skills). Empty for managed agents." + ), + ) + + +def is_external_agent(agent: Agent) -> bool: + """Return True when *agent* runs outside NeMo Platform (``source == 'external'``). + + The single predicate for the managed/external branch — mirrors + :func:`is_container_deployment_mode` so consumers don't reimplement the + string comparison inline. + """ + return agent.source == "external" class AgentDeployment(NemoEntity, entity_type="agent_deployment"): diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/schema.py b/plugins/nemo-agents/src/nemo_agents_plugin/schema.py index 6955181c15..d2f1cb0c25 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/schema.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/schema.py @@ -21,6 +21,7 @@ from __future__ import annotations from typing import Any +from urllib.parse import urlparse from nemo_agents_plugin.entities import ( NAT_WORKFLOW_CONFIG_FORMAT, @@ -30,7 +31,7 @@ DeploymentStatus, ) from nemo_platform_plugin.schema import NemoFilter, NemoListResponse -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator # --------------------------------------------------------------------------- # Request bodies — plain BaseModel, named by convention @@ -38,12 +39,46 @@ class CreateAgentRequest(BaseModel): - """Request body for ``POST /v2/workspaces/{workspace}/agents``.""" + """Request body for ``POST /v2/workspaces/{workspace}/agents``. + + Creates either a **managed** agent (supply ``config`` — a NAT workflow the + platform will run) or an **external** agent (supply ``url`` — a pointer to + a NAT agent already running elsewhere, whose A2A card the platform fetches + at creation). Provide exactly one of ``config`` or ``url``. + """ name: str = Field(description="Unique agent name within the workspace.") - description: str = Field(default="", description="Human-readable description.") - config: dict[str, Any] = Field(description="Agent config dict interpreted according to config_format.") + description: str = Field( + default="", + description="Human-readable description. For external agents, falls back to the agent card's description.", + ) + config: dict[str, Any] | None = Field( + default=None, + description="NAT workflow config dict. Required for a managed agent; omit for external.", + ) config_format: str = Field(default=NAT_WORKFLOW_CONFIG_FORMAT, description="Config format identifier.") + url: str | None = Field( + default=None, + description="Base URL of a running external agent (e.g. http://host:10000). Provide instead of config.", + ) + + @model_validator(mode="after") + def _require_config_xor_url(self) -> CreateAgentRequest: + if self.url: + if self.config is not None: + raise ValueError("Provide either 'config' (managed agent) or 'url' (external agent), not both.") + parsed = urlparse(self.url.strip()) + if parsed.scheme not in ("http", "https") or not parsed.hostname: + raise ValueError("'url' must be a valid http(s) URL, e.g. http://host:10000.") + elif self.config is None: + raise ValueError("'config' is required for a managed agent, or provide 'url' to register an external one.") + return self + + +class AgentReachability(BaseModel): + """Response for ``GET /v2/workspaces/{workspace}/agents/{name}/reachability``.""" + + reachable: bool = Field(description="Whether the external agent's endpoint answered a liveness probe.") class CreateDeploymentRequest(BaseModel): diff --git a/plugins/nemo-agents/tests/unit/test_a2a.py b/plugins/nemo-agents/tests/unit/test_a2a.py new file mode 100644 index 0000000000..e1e205b6e0 --- /dev/null +++ b/plugins/nemo-agents/tests/unit/test_a2a.py @@ -0,0 +1,200 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for A2A agent-card discovery (no real network; httpx mocked).""" + +from __future__ import annotations + +import json + +import httpx +import pytest +import respx +from nemo_agents_plugin.a2a import ( + A2AMessageError, + AgentCardError, + extract_message_text, + extract_stream_delta, + fetch_agent_card, + probe_agent_reachable, + send_a2a_message, + stream_a2a_message, +) + +CARD = {"name": "Calculator Agent", "description": "does math", "skills": [{"id": "add", "name": "add"}]} + + +@pytest.mark.asyncio +@respx.mock +async def test_fetch_returns_card_from_primary_path() -> None: + respx.get("http://host:10000/.well-known/agent-card.json").mock(return_value=httpx.Response(200, json=CARD)) + card = await fetch_agent_card("http://host:10000") + assert card["name"] == "Calculator Agent" + + +@pytest.mark.asyncio +@respx.mock +async def test_falls_back_to_legacy_path() -> None: + respx.get("http://host:10000/.well-known/agent-card.json").mock(return_value=httpx.Response(404)) + respx.get("http://host:10000/.well-known/agent.json").mock(return_value=httpx.Response(200, json=CARD)) + card = await fetch_agent_card("http://host:10000") + assert card["skills"][0]["id"] == "add" + + +@pytest.mark.asyncio +async def test_rejects_non_http_url() -> None: + with pytest.raises(AgentCardError, match="http"): + await fetch_agent_card("ftp://host") + + +@pytest.mark.asyncio +@respx.mock +async def test_unreachable_raises() -> None: + respx.get("http://host:10000/.well-known/agent-card.json").mock(side_effect=httpx.ConnectError("refused")) + respx.get("http://host:10000/.well-known/agent.json").mock(side_effect=httpx.ConnectError("refused")) + with pytest.raises(AgentCardError, match="Could not fetch a valid A2A agent card"): + await fetch_agent_card("http://host:10000") + + +@pytest.mark.asyncio +@respx.mock +async def test_non_card_json_rejected() -> None: + respx.get("http://host:10000/.well-known/agent-card.json").mock( + return_value=httpx.Response(200, json={"unrelated": True}) + ) + respx.get("http://host:10000/.well-known/agent.json").mock( + return_value=httpx.Response(200, json={"unrelated": True}) + ) + with pytest.raises(AgentCardError): + await fetch_agent_card("http://host:10000") + + +@pytest.mark.asyncio +@respx.mock +async def test_probe_reachable_true_on_200() -> None: + respx.get("http://host:10000/.well-known/agent-card.json").mock(return_value=httpx.Response(200, json=CARD)) + assert await probe_agent_reachable("http://host:10000") is True + + +@pytest.mark.asyncio +@respx.mock +async def test_probe_reachable_false_on_non_200_and_unreachable() -> None: + respx.get("http://down:1/.well-known/agent-card.json").mock(return_value=httpx.Response(503)) + respx.get("http://down:1/.well-known/agent.json").mock(side_effect=httpx.ConnectError("refused")) + assert await probe_agent_reachable("http://down:1") is False + assert await probe_agent_reachable("ftp://nope") is False + + +class TestExtractMessageText: + def test_message_parts(self) -> None: + result = {"kind": "message", "parts": [{"kind": "text", "text": "hello"}]} + assert extract_message_text(result) == "hello" + + def test_task_artifacts(self) -> None: + result = { + "kind": "task", + "artifacts": [{"parts": [{"kind": "text", "text": "42"}]}], + "status": {"state": "completed"}, + } + assert extract_message_text(result) == "42" + + def test_task_status_message(self) -> None: + result = {"status": {"message": {"parts": [{"kind": "text", "text": "done"}]}}} + assert extract_message_text(result) == "done" + + def test_no_text_returns_empty(self) -> None: + assert extract_message_text({"artifacts": [{"parts": [{"kind": "file"}]}]}) == "" + assert extract_message_text(None) == "" + + +@pytest.mark.asyncio +@respx.mock +async def test_send_message_returns_reply_text() -> None: + route = respx.post("http://host:10000/").mock( + return_value=httpx.Response( + 200, + json={ + "jsonrpc": "2.0", + "id": "1", + "result": {"kind": "message", "parts": [{"kind": "text", "text": "the answer"}]}, + }, + ) + ) + reply = await send_a2a_message("http://host:10000/", "what is it?") + assert reply == "the answer" + sent = json.loads(route.calls.last.request.content) + assert sent["method"] == "message/send" + assert sent["params"]["message"]["parts"][0]["text"] == "what is it?" + + +@pytest.mark.asyncio +@respx.mock +async def test_send_message_jsonrpc_error_raises() -> None: + respx.post("http://host:10000/").mock( + return_value=httpx.Response( + 200, json={"jsonrpc": "2.0", "id": "1", "error": {"code": -32000, "message": "boom"}} + ) + ) + with pytest.raises(A2AMessageError, match="boom"): + await send_a2a_message("http://host:10000/", "hi") + + +@pytest.mark.asyncio +@respx.mock +async def test_send_message_transport_error_raises() -> None: + respx.post("http://host:10000/").mock(side_effect=httpx.ConnectError("refused")) + with pytest.raises(A2AMessageError, match="could not reach"): + await send_a2a_message("http://host:10000/", "hi") + + +class TestExtractStreamDelta: + def test_message_event(self) -> None: + assert extract_stream_delta({"kind": "message", "parts": [{"kind": "text", "text": "hi"}]}) == "hi" + + def test_artifact_update_event(self) -> None: + event = {"kind": "artifact-update", "artifact": {"parts": [{"kind": "text", "text": "tok"}]}} + assert extract_stream_delta(event) == "tok" + + def test_status_update_and_empty_yield_nothing(self) -> None: + assert extract_stream_delta({"kind": "status-update", "status": {"state": "working"}}) == "" + assert extract_stream_delta(None) == "" + + +@pytest.mark.asyncio +@respx.mock +async def test_stream_message_yields_deltas() -> None: + sse = ( + ": ping\n\n" + 'data: {"jsonrpc":"2.0","id":"1","result":{"kind":"artifact-update",' + '"artifact":{"parts":[{"kind":"text","text":"2 + 2"}]}}}\n\n' + 'data: {"jsonrpc":"2.0","id":"1","result":{"kind":"message",' + '"parts":[{"kind":"text","text":" = 4"}]}}\n\n' + ) + respx.post("http://host:10000/").mock( + return_value=httpx.Response(200, text=sse, headers={"content-type": "text/event-stream"}) + ) + deltas = [d async for d in stream_a2a_message("http://host:10000/", "what is 2+2?")] + assert deltas == ["2 + 2", " = 4"] + + +@pytest.mark.asyncio +@respx.mock +async def test_stream_jsonrpc_error_raises() -> None: + sse = 'data: {"jsonrpc":"2.0","id":"1","error":{"code":-32000,"message":"kaboom"}}\n\n' + respx.post("http://host:10000/").mock( + return_value=httpx.Response(200, text=sse, headers={"content-type": "text/event-stream"}) + ) + with pytest.raises(A2AMessageError, match="kaboom"): + _ = [d async for d in stream_a2a_message("http://host:10000/", "hi")] + + +@pytest.mark.asyncio +@respx.mock +async def test_stream_caps_newlineless_body(monkeypatch: pytest.MonkeyPatch) -> None: + # A newline-less body must trip the byte cap rather than buffer unbounded. + monkeypatch.setattr("nemo_agents_plugin.a2a._MAX_MESSAGE_BYTES", 1024) + respx.post("http://host:10000/").mock( + return_value=httpx.Response(200, text="x" * 5000, headers={"content-type": "text/event-stream"}) + ) + with pytest.raises(A2AMessageError, match="size limit"): + _ = [d async for d in stream_a2a_message("http://host:10000/", "hi")] diff --git a/plugins/nemo-agents/tests/unit/test_agents_api.py b/plugins/nemo-agents/tests/unit/test_agents_api.py index cc29f30dd4..89c4bae466 100644 --- a/plugins/nemo-agents/tests/unit/test_agents_api.py +++ b/plugins/nemo-agents/tests/unit/test_agents_api.py @@ -193,6 +193,104 @@ def test_create_preserves_workspace_from_path(self, client: TestClient, mock_ent assert resp.json()["workspace"] == "prod" +# --------------------------------------------------------------------------- +# POST /agents with `url` — create an external agent (pointer to a running one) +# --------------------------------------------------------------------------- + + +_CARD = {"name": "Ext Agent", "description": "runs elsewhere", "skills": [{"id": "s1", "name": "s1"}]} + + +class TestCreateExternalAgent: + def test_url_payload_fetches_card_and_creates_external_agent( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(agents_router_module, "fetch_agent_card", AsyncMock(return_value=_CARD)) + mock_entity_client.create = AsyncMock(side_effect=lambda a: a) + + resp = client.post( + "/apis/agents/v2/workspaces/default/agents", + json={"name": "ext", "url": "http://host:10000"}, + ) + + assert resp.status_code == 201 + body = resp.json() + assert body["source"] == "external" + assert body["endpoint"] == "http://host:10000" + assert body["card"]["name"] == "Ext Agent" + # Description falls back to the card when not supplied. + assert body["description"] == "runs elsewhere" + + def test_explicit_description_overrides_card( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(agents_router_module, "fetch_agent_card", AsyncMock(return_value=_CARD)) + mock_entity_client.create = AsyncMock(side_effect=lambda a: a) + + resp = client.post( + "/apis/agents/v2/workspaces/default/agents", + json={"name": "ext", "url": "http://host:10000", "description": "mine"}, + ) + + assert resp.status_code == 201 + assert resp.json()["description"] == "mine" + + def test_unreachable_card_returns_502( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + from nemo_agents_plugin.a2a import AgentCardError + + monkeypatch.setattr( + agents_router_module, + "fetch_agent_card", + AsyncMock(side_effect=AgentCardError("could not reach agent")), + ) + + resp = client.post( + "/apis/agents/v2/workspaces/default/agents", + json={"name": "ext", "url": "http://host:10000"}, + ) + + assert resp.status_code == 502 + assert "could not reach" in resp.json()["detail"] + mock_entity_client.create.assert_not_called() + + def test_conflict_returns_409( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(agents_router_module, "fetch_agent_card", AsyncMock(return_value=_CARD)) + mock_entity_client.create = AsyncMock(side_effect=NemoEntityConflictError("already exists")) + + resp = client.post( + "/apis/agents/v2/workspaces/default/agents", + json={"name": "ext", "url": "http://host:10000"}, + ) + + assert resp.status_code == 409 + + def test_both_config_and_url_returns_422(self, client: TestClient, mock_entity_client: AsyncMock) -> None: + resp = client.post( + "/apis/agents/v2/workspaces/default/agents", + json={"name": "ext", "url": "http://host:10000", "config": {"workflow": {}}}, + ) + assert resp.status_code == 422 + + def test_neither_config_nor_url_returns_422(self, client: TestClient, mock_entity_client: AsyncMock) -> None: + resp = client.post( + "/apis/agents/v2/workspaces/default/agents", + json={"name": "ext"}, + ) + assert resp.status_code == 422 + + @pytest.mark.parametrize("bad_url", ["http://[", "not-a-url", "ftp://host", "http://"]) + def test_malformed_url_returns_422(self, client: TestClient, mock_entity_client: AsyncMock, bad_url: str) -> None: + resp = client.post( + "/apis/agents/v2/workspaces/default/agents", + json={"name": "ext", "url": bad_url}, + ) + assert resp.status_code == 422 + + # --------------------------------------------------------------------------- # GET /agents — list (NemoListResponse envelope) # --------------------------------------------------------------------------- @@ -398,3 +496,70 @@ def test_delete_server_error_on_delete_returns_500(self, client: TestClient, moc resp = client.delete("/apis/agents/v2/workspaces/default/agents/calc") assert resp.status_code == 500 + + +def _make_external(name: str = "ext", url: str = "http://host:10000") -> Agent: + a = Agent(name=name, workspace="default", source="external", endpoint=url, card={"name": "old"}) + a._id = f"agent-{name}-id" + return a + + +class TestRefreshExternalAgent: + _URL = "/apis/agents/v2/workspaces/default/agents/ext/refresh" + + def test_refresh_updates_card( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_external()) + monkeypatch.setattr(agents_router_module, "fetch_agent_card", AsyncMock(return_value={"name": "new"})) + mock_entity_client.update = AsyncMock(side_effect=lambda a: a) + + resp = client.post(self._URL) + + assert resp.status_code == 200 + assert resp.json()["card"]["name"] == "new" + + def test_refresh_managed_returns_400(self, client: TestClient, mock_entity_client: AsyncMock) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_agent("calc")) + resp = client.post("/apis/agents/v2/workspaces/default/agents/calc/refresh") + assert resp.status_code == 400 + + def test_refresh_not_found_returns_404(self, client: TestClient, mock_entity_client: AsyncMock) -> None: + mock_entity_client.get = AsyncMock(side_effect=NemoEntityNotFoundError("nope")) + resp = client.post(self._URL) + assert resp.status_code == 404 + + def test_refresh_unreachable_returns_502( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + from nemo_agents_plugin.a2a import AgentCardError + + mock_entity_client.get = AsyncMock(return_value=_make_external()) + monkeypatch.setattr(agents_router_module, "fetch_agent_card", AsyncMock(side_effect=AgentCardError("down"))) + resp = client.post(self._URL) + assert resp.status_code == 502 + + +class TestExternalAgentReachability: + _URL = "/apis/agents/v2/workspaces/default/agents/ext/reachability" + + @pytest.mark.parametrize("reachable", [True, False]) + def test_reachability( + self, + client: TestClient, + mock_entity_client: AsyncMock, + monkeypatch: pytest.MonkeyPatch, + reachable: bool, + ) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_external()) + monkeypatch.setattr(agents_router_module, "probe_agent_reachable", AsyncMock(return_value=reachable)) + + resp = client.get(self._URL) + + assert resp.status_code == 200 + assert resp.json()["reachable"] is reachable + + def test_reachability_managed_returns_400(self, client: TestClient, mock_entity_client: AsyncMock) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_agent("calc")) + resp = client.get("/apis/agents/v2/workspaces/default/agents/calc/reachability") + assert resp.status_code == 400 diff --git a/plugins/nemo-agents/tests/unit/test_deployments_api.py b/plugins/nemo-agents/tests/unit/test_deployments_api.py new file mode 100644 index 0000000000..fc53bf6242 --- /dev/null +++ b/plugins/nemo-agents/tests/unit/test_deployments_api.py @@ -0,0 +1,63 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for deployment create route (external-agent guard + happy path). + +Uses FastAPI TestClient with a mocked EntityClient; no network or entity store. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from nemo_agents_plugin.api.v2 import deployments as deployments_router_module +from nemo_agents_plugin.api.v2.dependencies import get_entity_client +from nemo_agents_plugin.entities import Agent + + +@pytest.fixture +def mock_entity_client() -> AsyncMock: + return AsyncMock() + + +@pytest.fixture +def client(mock_entity_client: AsyncMock) -> TestClient: + app = FastAPI() + app.include_router( + deployments_router_module.router, + prefix="/apis/agents/v2/workspaces/{workspace}", + ) + app.dependency_overrides[get_entity_client] = lambda: mock_entity_client + return TestClient(app, raise_server_exceptions=False) + + +def test_deploy_external_agent_returns_400(client: TestClient, mock_entity_client: AsyncMock) -> None: + external = Agent(name="ext", workspace="default", source="external", endpoint="http://host:10000") + mock_entity_client.get = AsyncMock(return_value=external) + + resp = client.post( + "/apis/agents/v2/workspaces/default/deployments", + json={"agent": "ext"}, + ) + + assert resp.status_code == 400 + assert "external" in resp.json()["detail"].lower() + mock_entity_client.create.assert_not_called() + + +def test_deploy_managed_agent_creates_pending(client: TestClient, mock_entity_client: AsyncMock) -> None: + managed = Agent(name="calc", workspace="default", config={"workflow": {"_type": "react_agent"}}) + mock_entity_client.get = AsyncMock(return_value=managed) + mock_entity_client.create = AsyncMock(side_effect=lambda d: d) + + resp = client.post( + "/apis/agents/v2/workspaces/default/deployments", + json={"agent": "calc"}, + ) + + assert resp.status_code == 201 + assert resp.json()["status"] == "pending" + mock_entity_client.create.assert_called_once() diff --git a/plugins/nemo-agents/tests/unit/test_gateway.py b/plugins/nemo-agents/tests/unit/test_gateway.py index 5ea7ba8e70..501f323b9e 100644 --- a/plugins/nemo-agents/tests/unit/test_gateway.py +++ b/plugins/nemo-agents/tests/unit/test_gateway.py @@ -615,3 +615,211 @@ def test_no_ready_container_returns_503(self, client: TestClient, mock_entity_cl ) assert resp.status_code == 503 + + +# --------------------------------------------------------------------------- +# External agent chat — OpenAI chat/completions <-> A2A message/send bridge +# --------------------------------------------------------------------------- + + +def _make_external_agent(name: str = "ext", url: str = "http://host:10000") -> Agent: + return Agent( + name=name, + workspace="default", + source="external", + endpoint=url, + card={"url": url, "name": "Ext Agent"}, + ) + + +class TestExternalAgentChat: + _URL = "/apis/agents/v2/workspaces/default/agents/ext/chat/completions" + + def test_non_stream_returns_openai_completion( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + monkeypatch.setattr(gateway_module, "send_a2a_message", AsyncMock(return_value="42 degrees")) + + resp = client.post(self._URL, json={"messages": [{"role": "user", "content": "temp?"}]}) + + assert resp.status_code == 200 + body = resp.json() + assert body["object"] == "chat.completion" + assert body["choices"][0]["message"]["content"] == "42 degrees" + + def test_stream_forwards_a2a_deltas_as_sse( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + + async def fake_stream(endpoint: str, text: str): + for delta in ["sun", "ny ", "24C"]: + yield delta + + monkeypatch.setattr(gateway_module, "stream_a2a_message", fake_stream) + + resp = client.post( + self._URL, + json={"stream": True, "messages": [{"role": "user", "content": "weather?"}]}, + ) + + assert resp.status_code == 200 + assert resp.headers["content-type"].startswith("text/event-stream") + assert '"content": "sun"' in resp.text + assert '"content": "24C"' in resp.text + assert '"finish_reason": "stop"' in resp.text + assert "[DONE]" in resp.text + + def test_stream_mid_error_surfaced_as_content( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + from nemo_agents_plugin.a2a import A2AMessageError + + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + + async def boom(endpoint: str, text: str): + raise A2AMessageError("upstream gone") + yield # pragma: no cover - makes this an async generator + + monkeypatch.setattr(gateway_module, "stream_a2a_message", boom) + + resp = client.post( + self._URL, + json={"stream": True, "messages": [{"role": "user", "content": "hi"}]}, + ) + + assert resp.status_code == 200 + assert "external agent error" in resp.text + assert "[DONE]" in resp.text + + def test_single_turn_forwards_verbatim( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + send = AsyncMock(return_value="ok") + monkeypatch.setattr(gateway_module, "send_a2a_message", send) + + client.post(self._URL, json={"messages": [{"role": "user", "content": "just this"}]}) + + assert send.await_args is not None + assert send.await_args.args[1] == "just this" + + def test_multi_turn_folds_history_into_transcript( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + # NAT's A2A executor is message-only, so prior turns are folded into the + # message rather than threaded via contextId. + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + send = AsyncMock(return_value="ok") + monkeypatch.setattr(gateway_module, "send_a2a_message", send) + + client.post( + self._URL, + json={ + "messages": [ + {"role": "user", "content": "first"}, + {"role": "assistant", "content": "reply"}, + {"role": "user", "content": "second"}, + ] + }, + ) + + assert send.await_args is not None + sent = send.await_args.args[1] + assert "User: first" in sent + assert "Assistant: reply" in sent + assert "User: second" in sent + + def test_managed_agent_returns_400(self, client: TestClient, mock_entity_client: AsyncMock) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_agent("calc")) + + resp = client.post( + "/apis/agents/v2/workspaces/default/agents/calc/chat/completions", + json={"messages": [{"role": "user", "content": "hi"}]}, + ) + + assert resp.status_code == 400 + + def test_agent_not_found_returns_404(self, client: TestClient, mock_entity_client: AsyncMock) -> None: + mock_entity_client.get = AsyncMock(side_effect=NemoEntityNotFoundError("nope")) + + resp = client.post(self._URL, json={"messages": [{"role": "user", "content": "hi"}]}) + + assert resp.status_code == 404 + + def test_no_user_message_returns_400(self, client: TestClient, mock_entity_client: AsyncMock) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + + resp = client.post(self._URL, json={"messages": [{"role": "system", "content": "x"}]}) + + assert resp.status_code == 400 + + def test_a2a_failure_returns_502( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + from nemo_agents_plugin.a2a import A2AMessageError + + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + monkeypatch.setattr(gateway_module, "send_a2a_message", AsyncMock(side_effect=A2AMessageError("boom"))) + + resp = client.post(self._URL, json={"messages": [{"role": "user", "content": "hi"}]}) + + assert resp.status_code == 502 + + +class TestExternalAgentGenerate: + """nat eval hits .../agents/{name}/-/generate/full -> A2A bridge for external agents.""" + + _URL = "/apis/agents/v2/workspaces/default/agents/ext/-/generate/full" + + def test_generate_full_bridges_to_a2a( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + send = AsyncMock(return_value="156") + monkeypatch.setattr(gateway_module, "send_a2a_message", send) + + resp = client.post(self._URL, json={"input_message": "What is 12 * 13?"}) + + assert resp.status_code == 200 + assert resp.headers["content-type"].startswith("text/event-stream") + assert '"value": "156"' in resp.text + assert send.await_args is not None + assert send.await_args.args[1] == "What is 12 * 13?" + + def test_missing_input_message_returns_400( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + monkeypatch.setattr(gateway_module, "send_a2a_message", AsyncMock(return_value="x")) + + resp = client.post(self._URL, json={"nope": "x"}) + + assert resp.status_code == 400 + + def test_chat_completions_via_proxy_bridges_to_a2a( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + # The CLI `nemo agents invoke` and SDK POST /-/v1/chat/completions — the + # external branch must bridge that (OpenAI chat), not just generate. + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + monkeypatch.setattr(gateway_module, "send_a2a_message", AsyncMock(return_value="4")) + + resp = client.post( + "/apis/agents/v2/workspaces/default/agents/ext/-/v1/chat/completions", + json={"messages": [{"role": "user", "content": "2+2?"}]}, + ) + + assert resp.status_code == 200 + assert resp.json()["choices"][0]["message"]["content"] == "4" + + def test_unsupported_path_returns_400(self, client: TestClient, mock_entity_client: AsyncMock) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + + resp = client.post( + "/apis/agents/v2/workspaces/default/agents/ext/-/health", + json={}, + ) + + assert resp.status_code == 400 From e0983440146fe8d7540c42fea09b3ef27f6ebb38 Mon Sep 17 00:00:00 2001 From: mschwab Date: Fri, 17 Jul 2026 11:25:01 -0700 Subject: [PATCH 02/10] fix(agents): address Codex review on the A2A bridge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - probe_agent_reachable streams and checks status only (no unbounded body read) - external chat honors the card's capabilities.streaming: a streaming request to a non-streaming agent uses message/send wrapped in one SSE chunk - streaming uses an incremental UTF-8 decoder so a multi-byte char split across network chunks isn't corrupted - streaming primes the first delta so a failure before any token returns 502; a mid-stream failure is logged and ends without a normal "stop" finish - the conversation prompt preserves system messages and anchors on the last user turn (drops trailing assistant) - managed-agent proxy no longer double-fetches the Agent entity Skipped (P2): full AgentCard schema validation — would require taking a2a-sdk as a runtime dep; cards stay loosely validated (operator-trust) and incompatible ones fail at invocation with a clear error. Signed-off-by: mschwab --- .../nemo-agents/src/nemo_agents_plugin/a2a.py | 31 +-- .../src/nemo_agents_plugin/api/v2/agents.py | 4 +- .../nemo_agents_plugin/api/v2/deployments.py | 20 +- .../src/nemo_agents_plugin/api/v2/gateway.py | 243 +++++++----------- .../src/nemo_agents_plugin/entities.py | 61 +---- .../nemo-agents/tests/unit/test_gateway.py | 72 +++++- 6 files changed, 199 insertions(+), 232 deletions(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py index 023dea2277..6c04d24d0a 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py @@ -12,6 +12,7 @@ from __future__ import annotations +import codecs import json import logging from collections.abc import AsyncIterator @@ -52,12 +53,7 @@ class A2AMessageError(Exception): async def _read_capped(response: httpx.Response, cap: int) -> bytes: - """Read a streamed response body, aborting once *cap* bytes are exceeded. - - Reads incrementally (unlike ``response.content``, which buffers the whole - decoded body first) so an oversized/compressed body can't exhaust memory - before the size check runs. - """ + """Read a streamed response body incrementally, aborting once *cap* bytes are exceeded.""" chunks: list[bytes] = [] total = 0 async for chunk in response.aiter_bytes(): @@ -85,9 +81,8 @@ async def fetch_agent_card(base_url: str) -> dict[str, Any]: if not base.startswith(("http://", "https://")): raise AgentCardError("Endpoint must be an http(s) URL.") - # Track the specific reason for logging, but never surface it to the caller: - # distinct "could not reach" vs "HTTP 401" vs "not JSON" messages would let a - # caller probe which internal hosts/ports exist (an SSRF oracle). + # Log the specific reason but never surface it: distinct errors would give the + # caller an SSRF oracle for probing internal hosts/ports. last_error = "no agent card found" async with httpx.AsyncClient(timeout=_FETCH_TIMEOUT_S, follow_redirects=True) as client: for path in AGENT_CARD_PATHS: @@ -142,11 +137,13 @@ async def probe_agent_reachable(base_url: str) -> bool: async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT_S) as client: for path in AGENT_CARD_PATHS: try: - resp = await client.get(_card_url(base, path)) + # Stream so we only wait for the status line — never read the + # (potentially unbounded / slow-drip) body for a liveness check. + async with client.stream("GET", _card_url(base, path)) as resp: + if resp.status_code == 200: + return True except httpx.HTTPError: continue - if resp.status_code == 200: - return True except httpx.HTTPError: return False return False @@ -269,17 +266,17 @@ async def stream_a2a_message(endpoint: str, text: str) -> AsyncIterator[str]: async with client.stream("POST", endpoint, json=payload, headers={"Accept": "text/event-stream"}) as resp: if resp.status_code != 200: raise A2AMessageError(f"agent returned HTTP {resp.status_code}") - # Cap raw bytes, not aiter_lines() output: aiter_lines buffers a - # line internally until a newline, so a newline-less body would - # grow unbounded before any per-line check. Split lines ourselves - # from a byte-capped buffer instead. + # Cap raw bytes and split lines ourselves: aiter_lines() would buffer a + # newline-less body unbounded before any per-line check runs. total = 0 buffer = "" + # Incremental decoder keeps a partial multi-byte char across chunk boundaries. + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") async for chunk in resp.aiter_bytes(): total += len(chunk) if total > _MAX_MESSAGE_BYTES: raise A2AMessageError("agent stream exceeded the size limit") - buffer += chunk.decode("utf-8", errors="replace") + buffer += decoder.decode(chunk) while "\n" in buffer: line, buffer = buffer.split("\n", 1) line = line.strip() diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py index bf82ac1f22..6d374f3f56 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py @@ -29,9 +29,7 @@ from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityConflictError, NemoEntityNotFoundError from nemo_platform_plugin.schema import PaginationData -# Deployment statuses that block agent deletion. -# "failed" and "deleting" are excluded — they are terminal/in-cleanup and -# do not represent an actively running process the user needs to clean up first. +# Statuses that block agent deletion; "failed"/"deleting" are terminal/in-cleanup, so excluded. _BLOCKING_STATUSES = frozenset({"pending", "starting", "running"}) logger = logging.getLogger(__name__) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py index ba5f12a895..ae646a9740 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py @@ -3,15 +3,10 @@ """Deployment lifecycle routes — create/list/get/delete AgentDeployment entities. -Creating a deployment: -1. Validates the referenced Agent exists. -2. Deep-copies the agent config and injects the inference-gateway URL. -3. Creates an ``AgentDeployment`` entity with ``status: "pending"``. -4. Returns immediately — the :class:`~nemo_agents_plugin.runner.controller.AgentDeploymentController` - picks up the pending deployment on the next reconcile cycle. - -Deleting a deployment marks it ``deleting``; the controller terminates the -process and removes the entity. +Create validates the agent, injects the inference-gateway URL into a copy of its +config, and saves an ``AgentDeployment`` in ``pending``; delete marks it ``deleting``. +The :class:`~nemo_agents_plugin.runner.controller.AgentDeploymentController` reconciles +both on its next cycle. """ from __future__ import annotations @@ -63,7 +58,6 @@ async def create_deployment( The deployment starts in ``pending`` state and is picked up by the deployment controller on its next reconcile cycle. """ - # 1. Validate the referenced agent exists try: agent = await entity_client.get(Agent, name=body.agent, workspace=workspace) except NemoEntityNotFoundError as exc: @@ -75,23 +69,19 @@ async def create_deployment( logger.exception("Failed to look up agent '%s'", body.agent) raise HTTPException(status_code=500, detail="Failed to look up agent.") from exc - # External agents run outside NeMo Platform; there is nothing for the - # platform to deploy. Reject rather than spawn a doomed empty-config process. + # External agents run outside NeMo Platform — nothing to deploy. if is_external_agent(agent): raise HTTPException( status_code=400, detail=f"Agent '{body.agent}' is external and runs outside NeMo Platform; it cannot be deployed.", ) - # 2. Build deployment name (auto-generate if not provided) deployment_name = body.name or f"{body.agent}-{secrets.token_hex(4)}" - # 3. Deep-copy config and inject IGW URL, telemetry fields, and default model. resolved_config = inject_gateway_url(agent.config, workspace) resolved_config = inject_default_model(resolved_config) inject_nemo_trace_fields(resolved_config, workspace=workspace, agent_name=body.agent) - # 4. Create the entity with status "pending" deployment = AgentDeployment( name=deployment_name, workspace=workspace, diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py index 741e6a86a9..cfe1fc2e9b 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py @@ -3,23 +3,10 @@ """Agent gateway proxy routes. -Two proxy routes: - -``/v2/workspaces/{workspace}/agents/{name}/-/{trailing_uri}`` - Proxy by **agent name** — gateway resolves the active deployment and - forwards the request. This is the primary user-facing path, analogous to - how IGW routes by model name. - -``/v2/workspaces/{workspace}/deployments/{name}/-/{trailing_uri}`` - Proxy by **deployment name** — for direct targeting of a specific - deployment (e.g. A/B testing). - -The ``/-/`` separator prevents URL conflicts with the CRUD routes -(``/agents/{name}`` and ``/deployments/{name}``). This mirrors the pattern -used by the Inference Gateway. - -Streaming and SSE are supported: the response is streamed back to the client -chunk by chunk. ``text/event-stream`` responses bypass buffering. +Proxy by **agent name** (``/agents/{name}/-/{trailing_uri}``) resolves the active +deployment; proxy by **deployment name** (``/deployments/{name}/-/...``) targets one +directly. The ``/-/`` separator avoids conflicts with the CRUD routes. Responses are +streamed chunk by chunk; ``text/event-stream`` bypasses buffering. """ from __future__ import annotations @@ -52,11 +39,8 @@ router = APIRouter() -# HTTP methods forwarded through the proxy, split by authorization scope. Read-like methods -# require only agents:read; mutating methods require agents:write. This mirrors the Inference -# Gateway's proxy precedent (its GET proxy is scoped inference:read), so a read-scoped token is -# not denied on read-only proxy calls. Both groups still require the same agents.gateway.invoke -# permission. +# Forwarded methods split by scope: read-like need agents:read, mutating need agents:write. +# Both still require the agents.gateway.invoke permission. _PROXY_READ_METHODS = ["GET", "HEAD", "OPTIONS"] _PROXY_WRITE_METHODS = ["POST", "PUT", "PATCH", "DELETE"] @@ -77,37 +61,11 @@ } -# --------------------------------------------------------------------------- -# Endpoint resolution — subprocess vs. container mode -# --------------------------------------------------------------------------- -# -# STOP-GAP: these two helpers are the *only* place the gateway learns where an -# agent lives. For subprocess deployments that is the loopback -# ``AgentDeployment.endpoint`` the agents plugin bakes in at spawn. For container -# deployments (docker/k8s) the real address (k8s Service DNS, docker host:port) -# is known only to the deployments plugin and projected onto ``endpoints`` by the -# agents controller. The rest of the proxy (streaming, SSE, header stripping, the -# SSRF origin guard in ``_proxy``) is mode-agnostic and untouched. -# -# POSSIBLE FUTURE DIRECTION: one option being considered is folding agent routing -# into the Inference Gateway (so IGW also serves as the agents gateway). If that -# direction is taken, this bespoke proxy — including ``_get_deployment_endpoint`` / -# ``_is_deployment_routable`` and the container branch below — would likely retire. -# That re-architecture is not committed to here; this stop-gap stands on its own. - -# Container modes (docker/k8s) resolve their address from the projected ``endpoints`` -# rather than the loopback ``endpoint``. The agents controller maps the deployments-plugin -# Deployment.status (READY/...) onto the agents-local status, so a routable container -# deployment reads as "running" — the same value subprocess uses. - - +# These two helpers are the only place the gateway learns where an agent lives: +# subprocess deployments use the loopback ``endpoint``; container modes (docker/k8s) +# use the routable address the agents controller projects onto ``endpoints``. def _get_deployment_endpoint(dep: AgentDeployment) -> str | None: - """Return the address to proxy to for *dep*, or ``None`` if none is available yet. - - - **subprocess** → the loopback ``endpoint`` the agents plugin allocates at spawn. - - **docker/k8s** → the first http(s) URL from ``endpoints``, which the agents - controller projects from the deployments-plugin ``Deployment``. - """ + """Return the address to proxy to for *dep*, or ``None`` if none is available yet.""" if is_container_deployment_mode(dep.deployment_mode): for ep in dep.endpoints: if ep.protocol in ("http", "https") and ep.url: @@ -117,12 +75,7 @@ def _get_deployment_endpoint(dep: AgentDeployment) -> str | None: def _is_deployment_routable(dep: AgentDeployment) -> bool: - """Return ``True`` if *dep* is currently up and has a resolvable endpoint. - - A deployment is routable when its status is ``running`` (for container modes, - the controller projects the deployments-plugin READY status onto ``running``) - and an endpoint can be resolved for its mode. - """ + """Return ``True`` if *dep* is ``running`` and has a resolvable endpoint.""" return dep.status == "running" and _get_deployment_endpoint(dep) is not None @@ -283,14 +236,11 @@ async def proxy_by_deployment_name_write( async def _resolve_agent_endpoint(name: str, workspace: str, entity_client: NemoEntitiesClient) -> str: - """Find the endpoint of the first running deployment for the given agent.""" - try: - await entity_client.get(Agent, name=name, workspace=workspace) - except NemoEntityNotFoundError as exc: - raise HTTPException(status_code=404, detail=f"Agent '{name}' not found in workspace '{workspace}'.") from exc - except Exception as exc: - raise HTTPException(status_code=500, detail=str(exc)) from exc + """Find the endpoint of the first running deployment for the given agent. + The agent's existence is already validated by the caller (``_serve_agent_proxy``), + so this only lists deployments — no second entity fetch. + """ try: result = await entity_client.list(AgentDeployment, workspace=workspace) except Exception as exc: @@ -318,21 +268,10 @@ async def _proxy( ) -> StreamingResponse: """Forward *request* to ``{endpoint}/{trailing_uri}`` and stream the response. - Error handling policy: - - **4xx** from the agent: transparent pass-through (client error, agent's response). - - **5xx** from the agent: translated to **502 Bad Gateway** (upstream fault). - - **Connection failure** (httpx.RequestError): 502 Bad Gateway. - - All responses are streamed, including SSE (``text/event-stream``). - ``content-length`` is stripped from forwarded headers because chunked - transfer encoding makes the original value invalid. + Agent 5xx and connection failures become 502; 4xx pass through. SSE is + supported; ``content-length`` is stripped since chunked encoding invalidates it. """ - # The SSRF guard below is origin-relative, not a host allow-list: it only - # rejects a trailing_uri that escapes the *resolved* endpoint's origin. That - # means it works unchanged for container-mode targets — a k8s in-cluster - # Service DNS name (``..svc.cluster.local:``) or a docker - # host:port — exactly as it does for subprocess loopback, as long as the - # resolved endpoint is a well-formed ``scheme://netloc`` (checked next). + # SSRF guard: reject any trailing_uri that escapes the resolved endpoint's origin. endpoint_parsed = urlparse(endpoint) if not endpoint_parsed.scheme or not endpoint_parsed.netloc: raise HTTPException(status_code=500, detail="Deployment endpoint is misconfigured.") @@ -350,11 +289,8 @@ async def _proxy( body = await request.body() - # We need the upstream response headers before we can construct StreamingResponse - # (to forward content-type, etc.). Use a two-phase approach: - # 1. Open the stream and capture headers — this triggers the HTTP round-trip. - # 2. Prime the generator with one __anext__() call so headers are populated. - # 3. Wrap in _buffered() to re-yield the primed chunk before continuing the stream. + # Headers are needed before constructing StreamingResponse, so prime one chunk + # up front and _buffered() re-yields it before continuing the stream. response_headers: dict[str, str] = {} status_code_holder: list[int] = [200] @@ -376,9 +312,7 @@ async def _stream_with_headers() -> AsyncIterator[bytes]: for k, v in response.headers.items(): if k.lower() not in _HOP_BY_HOP: response_headers[k] = v - # Translate agent 5xx responses into 502 Bad Gateway. - # aread() consumes the full body before raising so the connection - # is cleanly closed rather than reset mid-stream. + # Agent 5xx → 502; aread() drains the body so the connection closes cleanly. if response.status_code >= 500: error_body = await response.aread() raise HTTPException( @@ -411,14 +345,8 @@ async def _buffered() -> AsyncIterator[bytes]: content_type = response_headers.get("content-type", "application/json") - # NAT's ChatResponse.from_string() defaults model to "unknown-model" when - # the agent wrapper doesn't supply one (the wrapper code lives in - # nvidia-nat-core and doesn't have access to the platform entity name). - # For non-streaming JSON responses, patch the model field to the - # agent/deployment name the client addressed. This is a gateway-level - # workaround; the proper upstream fix belongs in nvidia-nat-core's - # NemoAgentWrapperFunction.convert_to_chat_response where the LLM's - # response_metadata carries the real model name. + # NAT defaults model to "unknown-model" when the wrapper can't see the platform + # entity name; patch non-streaming JSON responses to the addressed agent/deployment. if model_name and not content_type.startswith("text/event-stream"): async for remaining in stream_gen: chunks.append(remaining) @@ -440,17 +368,8 @@ async def _buffered() -> AsyncIterator[bytes]: ) -# --------------------------------------------------------------------------- -# External agent chat — OpenAI chat/completions <-> A2A message/send bridge -# --------------------------------------------------------------------------- -# -# External agents run outside NeMo Platform and speak A2A JSON-RPC, not the -# OpenAI chat API the Studio chat playground uses. This route accepts an -# OpenAI chat/completions request, forwards the latest user turn to the agent -# via ``message/send``, and returns the reply in OpenAI shape (SSE when the -# client asked to stream). Managed agents keep using the deployment proxy above. - - +# External agents speak A2A JSON-RPC, not OpenAI chat. These helpers bridge an +# OpenAI chat/completions request to ``message/send`` and back to OpenAI shape. def _message_text(message: Any) -> str: """Extract text from one OpenAI message (``content`` may be str or parts).""" if not isinstance(message, dict): @@ -466,27 +385,28 @@ def _message_text(message: Any) -> str: def _conversation_prompt(messages: Any) -> str: """Build a single A2A message that carries the conversation so far. - NAT's A2A executor is message-only (no server-side ``contextId`` memory), so - multi-turn context can't be threaded via the protocol — the client (OpenAI - chat) sends the full history each turn, and we fold prior user/assistant - turns into a transcript ahead of the latest user message. A single turn is - sent verbatim so simple agents aren't handed a transcript wrapper. + NAT's A2A executor is message-only, so prior turns are folded into a transcript + ahead of the latest user message. A single turn is sent verbatim. """ if not isinstance(messages, list): return "" turns = [ (m.get("role"), _message_text(m)) for m in messages - if isinstance(m, dict) and m.get("role") in ("user", "assistant") + if isinstance(m, dict) and m.get("role") in ("system", "user", "assistant") ] turns = [(role, text) for role, text in turns if text] - if not turns: + # Anchor on the last user turn; everything before it becomes the transcript. + last_user = next((i for i in range(len(turns) - 1, -1, -1) if turns[i][0] == "user"), None) + if last_user is None: return "" - if len(turns) == 1: - return turns[0][1] + latest = turns[last_user][1] + prior = turns[:last_user] + if not prior: + return latest - *prior, (_, latest) = turns - transcript = "\n".join(f"{'User' if role == 'user' else 'Assistant'}: {text}" for role, text in prior) + labels = {"system": "System", "user": "User", "assistant": "Assistant"} + transcript = "\n".join(f"{labels.get(str(role), 'User')}: {text}" for role, text in prior) return f"Continue this conversation.\n\n{transcript}\n\nUser: {latest}" @@ -517,27 +437,56 @@ def _openai_chunk(completion_id: str, created: int, model: str, delta: dict[str, return f"data: {json.dumps(payload)}\n\n".encode() -async def _stream_openai_from_a2a(endpoint: str, text: str, model: str) -> AsyncIterator[bytes]: - """Stream an external agent's A2A reply as OpenAI chat.completion.chunk SSE. +def _card_supports_streaming(agent: Agent) -> bool: + """False only when the card explicitly disables streaming; default True.""" + caps = (agent.card or {}).get("capabilities") + return not (isinstance(caps, dict) and caps.get("streaming") is False) + + +async def _single_chunk_sse(text: str, model: str) -> AsyncIterator[bytes]: + """Wrap a complete (non-streamed) reply as one OpenAI SSE completion.""" + created = int(time.time()) + completion_id = f"chatcmpl-{uuid4().hex}" + yield _openai_chunk(completion_id, created, model, {"role": "assistant", "content": text}, None) + yield _openai_chunk(completion_id, created, model, {}, "stop") + yield b"data: [DONE]\n\n" + - Forwards each A2A text delta as a content chunk. A mid-stream error is - surfaced as content (the HTTP status is already 200 once streaming starts), - so the user sees it in the chat rather than a silent stall. +async def _stream_openai_from_a2a( + agen: AsyncIterator[str], + first: str, + has_first: bool, + model: str, + name: str, + endpoint: str, +) -> AsyncIterator[bytes]: + """Stream a *primed* A2A reply as OpenAI chat.completion.chunk SSE. + + Status is already 200, so a mid-stream failure ends the stream without a + normal ``stop`` finish to mark it abnormal. """ created = int(time.time()) completion_id = f"chatcmpl-{uuid4().hex}" role_sent = False + if has_first: + yield _openai_chunk(completion_id, created, model, {"role": "assistant", "content": first}, None) + role_sent = True try: - async for delta in stream_a2a_message(endpoint, text): + async for delta in agen: payload = {"content": delta} if role_sent else {"role": "assistant", "content": delta} role_sent = True yield _openai_chunk(completion_id, created, model, payload, None) - if not role_sent: - yield _openai_chunk(completion_id, created, model, {"role": "assistant", "content": ""}, None) except A2AMessageError as exc: + logger.warning( + "External agent '%s' chat stream failed mid-stream (%s): %s", name, _endpoint_host(endpoint), exc + ) note = f"[external agent error: {exc}]" payload = {"content": f"\n{note}"} if role_sent else {"role": "assistant", "content": note} yield _openai_chunk(completion_id, created, model, payload, None) + yield b"data: [DONE]\n\n" # no ``stop`` finish — signals abnormal termination + return + if not role_sent: + yield _openai_chunk(completion_id, created, model, {"role": "assistant", "content": ""}, None) yield _openai_chunk(completion_id, created, model, {}, "stop") yield b"data: [DONE]\n\n" @@ -575,19 +524,11 @@ async def external_agent_chat_completions( endpoint = _external_a2a_endpoint(agent, name) body = await _read_json_object(request) - return await _external_openai_chat(name, endpoint, body) - - -# --------------------------------------------------------------------------- -# External agent bridge — OpenAI chat/completions and NAT /generate <-> A2A -# --------------------------------------------------------------------------- -# -# External agents have no deployment, so requests addressed to the agent-name -# proxy (``/agents/{name}/-/...``) land in :func:`_serve_external_agent`. Two -# invocation shapes are bridged to A2A: OpenAI ``chat/completions`` (Studio, the -# CLI ``nemo agents invoke``, and the SDK) and NAT ``generate/full`` (``nat eval``). + return await _external_openai_chat(name, endpoint, body, _card_supports_streaming(agent)) +# External agents have no deployment, so agent-name proxy requests land in +# _serve_external_agent, which bridges OpenAI chat/completions and NAT generate to A2A. def _endpoint_host(url: str) -> str: """Host of *url* for logging — never the full URL (may carry embedded creds).""" try: @@ -616,25 +557,40 @@ async def _read_json_object(request: Request) -> dict[str, Any]: return body if isinstance(body, dict) else {} -async def _external_openai_chat(name: str, endpoint: str, body: dict[str, Any]) -> StreamingResponse | JSONResponse: +async def _external_openai_chat( + name: str, endpoint: str, body: dict[str, Any], supports_streaming: bool +) -> StreamingResponse | JSONResponse: """Translate an OpenAI chat/completions body to A2A and return OpenAI shape.""" text = _conversation_prompt(body.get("messages")) if not text: raise HTTPException(status_code=400, detail="Request has no user message to send.") model = body.get("model") or name + wants_stream = bool(body.get("stream")) - # Streaming: forward A2A deltas as they arrive (message/stream), so the UI - # shows progress instead of freezing until the agent finishes. - if body.get("stream"): - return StreamingResponse(_stream_openai_from_a2a(endpoint, text, model), media_type="text/event-stream") + if wants_stream and supports_streaming: + # Prime the first delta before returning so an early failure is a 502, not a 200. + agen = stream_a2a_message(endpoint, text) + try: + first, has_first = await anext(agen), True + except StopAsyncIteration: + first, has_first = "", False + except A2AMessageError as exc: + logger.warning("External agent '%s' chat stream failed (%s): %s", name, _endpoint_host(endpoint), exc) + raise HTTPException(status_code=502, detail=f"External agent chat failed: {exc}") from exc + return StreamingResponse( + _stream_openai_from_a2a(agen, first, has_first, model, name, endpoint), + media_type="text/event-stream", + ) - # Non-streaming: single message/send, returned as one OpenAI completion. + # Non-streaming (or streaming-disabled card): one message/send, returned as JSON or one SSE chunk. try: reply = await send_a2a_message(endpoint, text) except A2AMessageError as exc: logger.warning("External agent '%s' chat failed (%s): %s", name, _endpoint_host(endpoint), exc) raise HTTPException(status_code=502, detail=f"External agent chat failed: {exc}") from exc + if wants_stream: + return StreamingResponse(_single_chunk_sse(reply, model), media_type="text/event-stream") return JSONResponse(_openai_completion(reply, model)) @@ -667,7 +623,8 @@ async def _serve_external_agent( """ endpoint = _external_a2a_endpoint(agent, name) if trailing_uri.endswith("chat/completions"): - return await _external_openai_chat(name, endpoint, await _read_json_object(request)) + body = await _read_json_object(request) + return await _external_openai_chat(name, endpoint, body, _card_supports_streaming(agent)) if trailing_uri.startswith("generate"): return await _serve_external_generate_full(name, endpoint, request) raise HTTPException( diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/entities.py b/plugins/nemo-agents/src/nemo_agents_plugin/entities.py index 6d6fb62909..2cc159db7d 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/entities.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/entities.py @@ -43,10 +43,8 @@ def is_container_deployment_mode(mode: str) -> bool: class Endpoint(BaseModel): """A routable network endpoint for a deployment. - Mirrors ``nemo_deployments_plugin.types.Endpoint`` so container-mode - deployments can carry the address the deployments-plugin ``Deployment`` - projected without the agents plugin depending on that plugin at the - entity-schema layer. + Mirrors ``nemo_deployments_plugin.types.Endpoint`` so the agents plugin need not + depend on the deployments plugin at the entity-schema layer. """ name: str @@ -54,29 +52,9 @@ class Endpoint(BaseModel): protocol: Literal["http", "https", "grpc", "tcp"] = "http" -# --------------------------------------------------------------------------- -# Canonical spec storage convention -# --------------------------------------------------------------------------- -# -# Each agent has exactly one spec fileset, named by convention. The fileset can -# hold both the human-readable agent spec and the machine-readable agent config. -# We do **not** store these locations on the agent - they are fully derivable -# from the agent's workspace and name. Consumers should call the file-ref -# helpers below rather than reconstructing refs inline. -# -# Layout: -# - Fileset (entity ref): ``{workspace}/{agent-name}-spec`` -# - Human-readable spec: ``AGENT-SPEC.md`` (industry-standard name) -# - Machine-readable cfg: ``agent.yaml`` -# - Spec file ref: ``{workspace}/{agent-name}-spec#AGENT-SPEC.md`` -# - Config file ref: ``{workspace}/{agent-name}-spec#agent.yaml`` -# - Local cache root: ``agents/{agent-name}-spec/`` -# -# This is intentionally **not** an Optional field on the Agent. The -# relationship is 1:1 and convention-bound; carrying a stored ref would -# duplicate state with no resilience benefit (rename of either entity -# orphans both representations equally). - +# Each agent has one spec fileset (``{workspace}/{name}-spec``) holding the human spec +# and the machine config. Locations are derivable from (workspace, name), so they aren't +# stored on the agent; call the file-ref helpers below rather than rebuilding refs inline. AGENT_SPEC_FILENAME = "AGENT-SPEC.md" """Canonical filename inside the agent's spec fileset.""" @@ -107,24 +85,12 @@ def agent_spec_local_path(agent_name: str, root: str | Path = AGENT_SPEC_LOCAL_R def agent_spec_file_ref(workspace: str, agent_name: str) -> FilesetRef: - """Return the canonical file ref ``workspace/-spec#AGENT-SPEC.md``. - - Use this anywhere downstream code needs to point at an agent's spec - - do not reconstruct the path inline. If the layout ever changes (e.g. - moving to a per-agent bundle fileset holding multiple artifacts), this - is the only function that needs to update. - """ + """Return the canonical file ref ``workspace/-spec#AGENT-SPEC.md``.""" return FilesetRef(f"{workspace}/{agent_spec_fileset_name(agent_name)}#{AGENT_SPEC_FILENAME}") def agent_config_file_ref(workspace: str, agent_name: str) -> FilesetRef: - """Return the canonical file ref ``workspace/-spec#agent.yaml``. - - Use this anywhere downstream code needs to point at an agent's config - - do not reconstruct the path inline. If the layout ever changes (e.g. - moving to a per-agent bundle fileset holding multiple artifacts), this - is the only function that needs to update. - """ + """Return the canonical file ref ``workspace/-spec#agent.yaml``.""" return FilesetRef(f"{workspace}/{agent_spec_fileset_name(agent_name)}#{AGENT_CONFIG_FILENAME}") @@ -136,10 +102,8 @@ class Agent(NemoEntity, entity_type="agent"): Entity type: ``agent`` Primary lookup: by ``name`` within a ``workspace``. - The agent's spec files live at the locations returned by - :func:`agent_spec_file_ref` and :func:`agent_config_file_ref` — they - are **not** stored on the entity because the paths are fully derivable - from ``(workspace, name)``. + Spec files live at :func:`agent_spec_file_ref` / :func:`agent_config_file_ref`; + the paths aren't stored on the entity since they derive from ``(workspace, name)``. """ description: str = Field(default="", description="Human-readable description of the agent.") @@ -179,12 +143,7 @@ class Agent(NemoEntity, entity_type="agent"): def is_external_agent(agent: Agent) -> bool: - """Return True when *agent* runs outside NeMo Platform (``source == 'external'``). - - The single predicate for the managed/external branch — mirrors - :func:`is_container_deployment_mode` so consumers don't reimplement the - string comparison inline. - """ + """Return True when *agent* runs outside NeMo Platform (``source == 'external'``).""" return agent.source == "external" diff --git a/plugins/nemo-agents/tests/unit/test_gateway.py b/plugins/nemo-agents/tests/unit/test_gateway.py index 501f323b9e..e4aae9b027 100644 --- a/plugins/nemo-agents/tests/unit/test_gateway.py +++ b/plugins/nemo-agents/tests/unit/test_gateway.py @@ -622,13 +622,13 @@ def test_no_ready_container_returns_503(self, client: TestClient, mock_entity_cl # --------------------------------------------------------------------------- -def _make_external_agent(name: str = "ext", url: str = "http://host:10000") -> Agent: +def _make_external_agent(name: str = "ext", url: str = "http://host:10000", card: dict | None = None) -> Agent: return Agent( name=name, workspace="default", source="external", endpoint=url, - card={"url": url, "name": "Ext Agent"}, + card=card if card is not None else {"url": url, "name": "Ext Agent"}, ) @@ -671,9 +671,11 @@ async def fake_stream(endpoint: str, text: str): assert '"finish_reason": "stop"' in resp.text assert "[DONE]" in resp.text - def test_stream_mid_error_surfaced_as_content( + def test_stream_early_error_returns_502( self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch ) -> None: + # A failure before the first token is primed, so it becomes a real 502 — + # not a 200 whose body is an error string. from nemo_agents_plugin.a2a import A2AMessageError mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) @@ -689,8 +691,50 @@ async def boom(endpoint: str, text: str): json={"stream": True, "messages": [{"role": "user", "content": "hi"}]}, ) + assert resp.status_code == 502 + + def test_stream_mid_error_ends_without_stop( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + # A failure after the first token can't change the 200 status; it's + # surfaced as content and the stream ends WITHOUT a normal stop finish. + from nemo_agents_plugin.a2a import A2AMessageError + + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + + async def one_then_boom(endpoint: str, text: str): + yield "partial" + raise A2AMessageError("dropped") + + monkeypatch.setattr(gateway_module, "stream_a2a_message", one_then_boom) + + resp = client.post( + self._URL, + json={"stream": True, "messages": [{"role": "user", "content": "hi"}]}, + ) + assert resp.status_code == 200 + assert "partial" in resp.text assert "external agent error" in resp.text + assert '"finish_reason": "stop"' not in resp.text + + def test_streaming_disabled_card_uses_send( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + # Card disables streaming → a streaming request is served via a single + # message/send wrapped as one SSE chunk (not message/stream). + agent = _make_external_agent(card={"url": "http://host:10000", "capabilities": {"streaming": False}}) + mock_entity_client.get = AsyncMock(return_value=agent) + monkeypatch.setattr(gateway_module, "send_a2a_message", AsyncMock(return_value="full reply")) + + resp = client.post( + self._URL, + json={"stream": True, "messages": [{"role": "user", "content": "hi"}]}, + ) + + assert resp.status_code == 200 + assert resp.headers["content-type"].startswith("text/event-stream") + assert '"content": "full reply"' in resp.text assert "[DONE]" in resp.text def test_single_turn_forwards_verbatim( @@ -731,6 +775,28 @@ def test_multi_turn_folds_history_into_transcript( assert "Assistant: reply" in sent assert "User: second" in sent + def test_system_message_preserved_and_anchored_on_last_user( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + send = AsyncMock(return_value="ok") + monkeypatch.setattr(gateway_module, "send_a2a_message", send) + + client.post( + self._URL, + json={ + "messages": [ + {"role": "system", "content": "be terse"}, + {"role": "user", "content": "hello"}, + ] + }, + ) + + assert send.await_args is not None + sent = send.await_args.args[1] + assert "System: be terse" in sent + assert sent.rstrip().endswith("User: hello") + def test_managed_agent_returns_400(self, client: TestClient, mock_entity_client: AsyncMock) -> None: mock_entity_client.get = AsyncMock(return_value=_make_agent("calc")) From a310800c8b465577f0f2690b8ea2ffa2c9e8610e Mon Sep 17 00:00:00 2001 From: mschwab Date: Fri, 17 Jul 2026 12:26:39 -0700 Subject: [PATCH 03/10] =?UTF-8?q?fix(agents):=20address=20CodeRabbit=20rev?= =?UTF-8?q?iew=20=E2=80=94=20disable=20A2A=20redirects,=20sanitize=20gatew?= =?UTF-8?q?ay=20errors?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - a2a.py: follow_redirects=False on all httpx sinks so a registered external agent URL can't 302 the backend into loopback/metadata endpoints (SSRF). - gateway.py: entity-store lookup failures no longer leak str(exc); log and return a fixed detail, matching the agents.py CRUD routes. - Regression tests for both. Signed-off-by: mschwab --- .../nemo-agents/src/nemo_agents_plugin/a2a.py | 6 ++--- .../src/nemo_agents_plugin/api/v2/gateway.py | 12 ++++++--- plugins/nemo-agents/tests/unit/test_a2a.py | 27 +++++++++++++++++++ .../nemo-agents/tests/unit/test_gateway.py | 12 +++++++++ 4 files changed, 50 insertions(+), 7 deletions(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py index 6c04d24d0a..c8a69595be 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py @@ -84,7 +84,7 @@ async def fetch_agent_card(base_url: str) -> dict[str, Any]: # Log the specific reason but never surface it: distinct errors would give the # caller an SSRF oracle for probing internal hosts/ports. last_error = "no agent card found" - async with httpx.AsyncClient(timeout=_FETCH_TIMEOUT_S, follow_redirects=True) as client: + async with httpx.AsyncClient(timeout=_FETCH_TIMEOUT_S, follow_redirects=False) as client: for path in AGENT_CARD_PATHS: url = _card_url(base, path) try: @@ -202,7 +202,7 @@ async def send_a2a_message(endpoint: str, text: str) -> str: }, } try: - async with httpx.AsyncClient(timeout=_MESSAGE_TIMEOUT_S, follow_redirects=True) as client: + async with httpx.AsyncClient(timeout=_MESSAGE_TIMEOUT_S, follow_redirects=False) as client: async with client.stream("POST", endpoint, json=payload, headers={"Accept": "application/json"}) as resp: if resp.status_code != 200: raise A2AMessageError(f"agent returned HTTP {resp.status_code}") @@ -262,7 +262,7 @@ async def stream_a2a_message(endpoint: str, text: str) -> AsyncIterator[str]: }, } try: - async with httpx.AsyncClient(timeout=_MESSAGE_TIMEOUT_S, follow_redirects=True) as client: + async with httpx.AsyncClient(timeout=_MESSAGE_TIMEOUT_S, follow_redirects=False) as client: async with client.stream("POST", endpoint, json=payload, headers={"Accept": "text/event-stream"}) as resp: if resp.status_code != 200: raise A2AMessageError(f"agent returned HTTP {resp.status_code}") diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py index cfe1fc2e9b..50106a3458 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py @@ -98,7 +98,8 @@ async def _serve_agent_proxy( except NemoEntityNotFoundError as exc: raise HTTPException(status_code=404, detail=f"Agent '{name}' not found in workspace '{workspace}'.") from exc except Exception as exc: - raise HTTPException(status_code=500, detail=str(exc)) from exc + logger.exception("Failed to look up agent '%s'", name) + raise HTTPException(status_code=500, detail="Failed to look up agent.") from exc if is_external_agent(agent): return await _serve_external_agent(name, trailing_uri, request, agent) @@ -172,7 +173,8 @@ async def _serve_deployment_proxy( status_code=404, detail=f"Deployment '{name}' not found in workspace '{workspace}'." ) from exc except Exception as exc: - raise HTTPException(status_code=500, detail=str(exc)) from exc + logger.exception("Failed to look up deployment '%s'", name) + raise HTTPException(status_code=500, detail="Failed to look up deployment.") from exc if not _is_deployment_routable(dep): raise HTTPException( @@ -244,7 +246,8 @@ async def _resolve_agent_endpoint(name: str, workspace: str, entity_client: Nemo try: result = await entity_client.list(AgentDeployment, workspace=workspace) except Exception as exc: - raise HTTPException(status_code=500, detail=str(exc)) from exc + logger.exception("Failed to list deployments for agent '%s'", name) + raise HTTPException(status_code=500, detail="Failed to list deployments.") from exc running = [d for d in result.data if d.agent == name and _is_deployment_routable(d)] if not running: @@ -514,7 +517,8 @@ async def external_agent_chat_completions( except NemoEntityNotFoundError as exc: raise HTTPException(status_code=404, detail=f"Agent '{name}' not found in workspace '{workspace}'.") from exc except Exception as exc: - raise HTTPException(status_code=500, detail=str(exc)) from exc + logger.exception("Failed to look up agent '%s'", name) + raise HTTPException(status_code=500, detail="Failed to look up agent.") from exc if not is_external_agent(agent): raise HTTPException( diff --git a/plugins/nemo-agents/tests/unit/test_a2a.py b/plugins/nemo-agents/tests/unit/test_a2a.py index e1e205b6e0..4a70473b22 100644 --- a/plugins/nemo-agents/tests/unit/test_a2a.py +++ b/plugins/nemo-agents/tests/unit/test_a2a.py @@ -69,6 +69,33 @@ async def test_non_card_json_rejected() -> None: await fetch_agent_card("http://host:10000") +@pytest.mark.asyncio +@respx.mock +async def test_fetch_does_not_follow_redirects() -> None: + respx.get("http://host:10000/.well-known/agent-card.json").mock( + return_value=httpx.Response(302, headers={"location": "http://169.254.169.254/"}) + ) + respx.get("http://host:10000/.well-known/agent.json").mock( + return_value=httpx.Response(302, headers={"location": "http://169.254.169.254/"}) + ) + internal = respx.get("http://169.254.169.254/").mock(return_value=httpx.Response(200, json=CARD)) + with pytest.raises(AgentCardError): + await fetch_agent_card("http://host:10000") + assert not internal.called + + +@pytest.mark.asyncio +@respx.mock +async def test_send_message_does_not_follow_redirects() -> None: + respx.post("http://host:10000/").mock( + return_value=httpx.Response(302, headers={"location": "http://169.254.169.254/"}) + ) + internal = respx.post("http://169.254.169.254/").mock(return_value=httpx.Response(200, json={"result": {}})) + with pytest.raises(A2AMessageError): + await send_a2a_message("http://host:10000/", "hi") + assert not internal.called + + @pytest.mark.asyncio @respx.mock async def test_probe_reachable_true_on_200() -> None: diff --git a/plugins/nemo-agents/tests/unit/test_gateway.py b/plugins/nemo-agents/tests/unit/test_gateway.py index e4aae9b027..190e7e7bc5 100644 --- a/plugins/nemo-agents/tests/unit/test_gateway.py +++ b/plugins/nemo-agents/tests/unit/test_gateway.py @@ -208,6 +208,18 @@ def test_503_from_agent_becomes_502(self, client: TestClient, mock_entity_client assert resp.status_code == 502 + def test_entity_lookup_failure_is_sanitized(self, client: TestClient, mock_entity_client: AsyncMock) -> None: + mock_entity_client.get = AsyncMock(side_effect=RuntimeError("db dsn postgres://secret@host")) + + resp = client.post( + "/apis/agents/v2/workspaces/default/deployments/calc-dep/-/v1/chat/completions", + json={}, + ) + + assert resp.status_code == 500 + assert "secret" not in resp.text + assert "Failed to look up deployment." in resp.text + def test_4xx_from_agent_passed_through(self, client: TestClient, mock_entity_client: AsyncMock) -> None: """4xx client errors from the agent are transparent pass-through.""" dep = _make_deployment(status="running", endpoint="http://localhost:9001") From 09069100aa9bc71eca840743283554c760c91da3 Mon Sep 17 00:00:00 2001 From: mschwab Date: Fri, 17 Jul 2026 13:00:13 -0700 Subject: [PATCH 04/10] chore(agents): regenerate SDK/OpenAPI after docstring trims Signed-off-by: mschwab --- plugins/nemo-agents/openapi/openapi.yaml | 19 ++++++++----------- .../nemo-platform/.github/workflows/ci.yml | 6 +++--- 2 files changed, 11 insertions(+), 14 deletions(-) diff --git a/plugins/nemo-agents/openapi/openapi.yaml b/plugins/nemo-agents/openapi/openapi.yaml index ec3eb473e2..c88908b1fc 100644 --- a/plugins/nemo-agents/openapi/openapi.yaml +++ b/plugins/nemo-agents/openapi/openapi.yaml @@ -2355,8 +2355,8 @@ components: additionalProperties: true type: object title: Config - description: Agent config dict interpreted according to config_format. - Empty for external agents. + description: Agent config dict interpreted according to config_format. Empty + for external agents. config_format: type: string title: Config Format @@ -2434,9 +2434,9 @@ components: title: Agent description: "An agent definition \u2014 stores agent config and metadata.\n\ \nEntity type: ``agent``\nPrimary lookup: by ``name`` within a ``workspace``.\n\ - \nThe agent's spec files live at the locations returned by\n:func:`agent_spec_file_ref`\ - \ and :func:`agent_config_file_ref` \u2014 they\nare **not** stored on the\ - \ entity because the paths are fully derivable\nfrom ``(workspace, name)``." + \nSpec files live at :func:`agent_spec_file_ref` / :func:`agent_config_file_ref`;\n\ + the paths aren't stored on the entity since they derive from ``(workspace,\ + \ name)``." AgentDeployment: properties: name: @@ -2884,13 +2884,10 @@ components: description: 'A routable network endpoint for a deployment. - Mirrors ``nemo_deployments_plugin.types.Endpoint`` so container-mode + Mirrors ``nemo_deployments_plugin.types.Endpoint`` so the agents plugin need + not - deployments can carry the address the deployments-plugin ``Deployment`` - - projected without the agents plugin depending on that plugin at the - - entity-schema layer.' + depend on the deployments plugin at the entity-schema layer.' EvaluateAgentSpec: properties: agent: diff --git a/sdk/python/nemo-platform/.github/workflows/ci.yml b/sdk/python/nemo-platform/.github/workflows/ci.yml index 558389ad06..7c53993236 100644 --- a/sdk/python/nemo-platform/.github/workflows/ci.yml +++ b/sdk/python/nemo-platform/.github/workflows/ci.yml @@ -18,7 +18,7 @@ jobs: lint: timeout-minutes: 10 name: lint - runs-on: ${{ github.repository == 'stainless-sdks/nemo-platform-python' && 'depot-ubuntu-24.04' || 'ubuntu-latest' }} + runs-on: ${{ startsWith(github.repository, 'stainless-sdks/') && 'depot-ubuntu-24.04' || 'ubuntu-latest' }} if: (github.event_name == 'push' || github.event.pull_request.head.repo.fork) && (github.event_name != 'push' || github.event.head_commit.message != 'codegen metadata') steps: - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 @@ -41,7 +41,7 @@ jobs: permissions: contents: read id-token: write - runs-on: ${{ github.repository == 'stainless-sdks/nemo-platform-python' && 'depot-ubuntu-24.04' || 'ubuntu-latest' }} + runs-on: ${{ startsWith(github.repository, 'stainless-sdks/') && 'depot-ubuntu-24.04' || 'ubuntu-latest' }} steps: - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 @@ -78,7 +78,7 @@ jobs: test: timeout-minutes: 10 name: test - runs-on: ${{ github.repository == 'stainless-sdks/nemo-platform-python' && 'depot-ubuntu-24.04' || 'ubuntu-latest' }} + runs-on: ${{ startsWith(github.repository, 'stainless-sdks/') && 'depot-ubuntu-24.04' || 'ubuntu-latest' }} if: github.event_name == 'push' || github.event.pull_request.head.repo.fork steps: - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 From 980762f78a61bf4a6e65e2724500930d1bce987c Mon Sep 17 00:00:00 2001 From: mschwab Date: Fri, 17 Jul 2026 13:12:11 -0700 Subject: [PATCH 05/10] =?UTF-8?q?fix(agents):=20address=20CodeQL=20?= =?UTF-8?q?=E2=80=94=20SSRF=20egress=20guard=20+=20log-injection=20scrubbi?= =?UTF-8?q?ng?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - a2a.py: block external-agent URLs that resolve to private/loopback/link-local/ reserved addresses (e.g. 169.254.169.254 metadata) before any request, in fetch/send/stream/probe. Gated by NEMO_AGENTS_ALLOW_PRIVATE_AGENT_HOSTS for local dev. (py/full-ssrf #4269) - log_utils.scrub(): strip CR/LF from user-controlled values before logging; applied across gateway/agents/a2a logger calls. (py/log-injection #4270-4279) - Tests for the guard (blocks private, allows public) and the scrubber. Signed-off-by: mschwab --- .../nemo-agents/src/nemo_agents_plugin/a2a.py | 59 +++++++++++++++++-- .../src/nemo_agents_plugin/api/v2/agents.py | 13 ++-- .../src/nemo_agents_plugin/api/v2/gateway.py | 24 +++++--- .../src/nemo_agents_plugin/log_utils.py | 11 ++++ plugins/nemo-agents/tests/unit/test_a2a.py | 51 ++++++++++++++++ .../nemo-agents/tests/unit/test_log_utils.py | 17 ++++++ 6 files changed, 157 insertions(+), 18 deletions(-) create mode 100644 plugins/nemo-agents/src/nemo_agents_plugin/log_utils.py create mode 100644 plugins/nemo-agents/tests/unit/test_log_utils.py diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py index c8a69595be..272004c3ad 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py @@ -12,18 +12,29 @@ from __future__ import annotations +import asyncio import codecs +import ipaddress import json import logging +import os +import socket from collections.abc import AsyncIterator from typing import Any -from urllib.parse import urljoin +from urllib.parse import urljoin, urlparse from uuid import uuid4 import httpx +from nemo_agents_plugin.log_utils import scrub logger = logging.getLogger(__name__) +# External-agent URLs are user-supplied, so the backend fetching them is an SSRF sink. +# By default reject any host that resolves to a private/loopback/link-local/reserved +# address (e.g. cloud metadata at 169.254.169.254); set this env var to allow them, +# which is needed for local dev where agents run on localhost. +_ALLOW_PRIVATE_HOSTS_ENV = "NEMO_AGENTS_ALLOW_PRIVATE_AGENT_HOSTS" + # Well-known paths that expose an A2A agent card. The spec moved from # ``agent.json`` to ``agent-card.json``; try the current name first. AGENT_CARD_PATHS = ("/.well-known/agent-card.json", "/.well-known/agent.json") @@ -52,6 +63,38 @@ class A2AMessageError(Exception): """Raised when a ``message/send`` call to an external agent fails.""" +class AgentHostNotAllowed(A2AMessageError): + """Raised when an agent URL resolves to a disallowed (e.g. private/loopback) address.""" + + +def _address_disallowed(ip_text: str) -> bool: + try: + ip = ipaddress.ip_address(ip_text) + except ValueError: + return True + return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast or ip.is_unspecified + + +async def _ensure_host_allowed(url: str) -> None: + """Reject *url* whose host resolves to a private/loopback/link-local address. + + The SSRF guard for external-agent URLs. Bypassed by ``_ALLOW_PRIVATE_HOSTS_ENV`` + for local dev. Never surfaces the resolved address to the caller. + """ + if os.environ.get(_ALLOW_PRIVATE_HOSTS_ENV): + return + host = urlparse(url).hostname + if not host: + raise AgentHostNotAllowed("agent endpoint has no host") + try: + infos = await asyncio.to_thread(socket.getaddrinfo, host, None, type=socket.SOCK_STREAM) + except socket.gaierror as exc: + raise AgentHostNotAllowed(f"could not resolve agent host ({exc.__class__.__name__})") from exc + if any(_address_disallowed(str(info[4][0])) for info in infos): + logger.warning("Blocked external-agent host %s: resolves to a disallowed address", _endpoint_host(url)) + raise AgentHostNotAllowed("agent host resolves to a disallowed address") + + async def _read_capped(response: httpx.Response, cap: int) -> bytes: """Read a streamed response body incrementally, aborting once *cap* bytes are exceeded.""" chunks: list[bytes] = [] @@ -80,6 +123,10 @@ async def fetch_agent_card(base_url: str) -> dict[str, Any]: base = base_url.strip() if not base.startswith(("http://", "https://")): raise AgentCardError("Endpoint must be an http(s) URL.") + try: + await _ensure_host_allowed(base) + except AgentHostNotAllowed as exc: + raise AgentCardError("Could not fetch a valid A2A agent card from the provided URL.") from exc # Log the specific reason but never surface it: distinct errors would give the # caller an SSRF oracle for probing internal hosts/ports. @@ -110,14 +157,12 @@ async def fetch_agent_card(base_url: str) -> dict[str, Any]: continue return card - logger.info("Agent card fetch failed for %s: %s", _endpoint_host(base), last_error) + logger.info("Agent card fetch failed for %s: %s", _endpoint_host(base), scrub(last_error)) raise AgentCardError("Could not fetch a valid A2A agent card from the provided URL.") def _endpoint_host(url: str) -> str: """Host of *url* for logging — avoid echoing the full URL back anywhere.""" - from urllib.parse import urlparse - try: return urlparse(url).hostname or "unknown" except ValueError: @@ -133,6 +178,10 @@ async def probe_agent_reachable(base_url: str) -> bool: base = base_url.strip() if not base.startswith(("http://", "https://")): return False + try: + await _ensure_host_allowed(base) + except AgentHostNotAllowed: + return False try: async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT_S) as client: for path in AGENT_CARD_PATHS: @@ -188,6 +237,7 @@ async def send_a2a_message(endpoint: str, text: str) -> str: :class:`A2AMessageError` on transport failure, a JSON-RPC error, or a non-JSON response. """ + await _ensure_host_allowed(endpoint) payload = { "jsonrpc": "2.0", "id": uuid4().hex, @@ -248,6 +298,7 @@ async def stream_a2a_message(endpoint: str, text: str) -> AsyncIterator[str]: a single final chunk. Raises :class:`A2AMessageError` on transport/JSON-RPC failure before the first token; mid-stream transport drops end the iterator. """ + await _ensure_host_allowed(endpoint) payload = { "jsonrpc": "2.0", "id": uuid4().hex, diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py index 6d374f3f56..da923f5533 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py @@ -18,6 +18,7 @@ from nemo_agents_plugin.api.v2.dependencies import get_entity_client from nemo_agents_plugin.authz import scope from nemo_agents_plugin.entities import Agent, AgentDeployment, is_external_agent +from nemo_agents_plugin.log_utils import scrub from nemo_agents_plugin.schema import ( AgentFilter, AgentPage, @@ -87,7 +88,7 @@ async def create_agent( detail=f"Agent '{body.name}' already exists in workspace '{workspace}'.", ) from exc except Exception as exc: - logger.exception("Failed to create agent '%s'", body.name) + logger.exception("Failed to create agent '%s'", scrub(body.name)) raise HTTPException(status_code=500, detail="Failed to create agent.") from exc return saved @@ -129,7 +130,7 @@ async def refresh_external_agent( try: saved = await entity_client.update(agent) except Exception as exc: - logger.exception("Failed to refresh external agent '%s'", name) + logger.exception("Failed to refresh external agent '%s'", scrub(name)) raise HTTPException(status_code=500, detail="Failed to refresh external agent.") from exc return saved @@ -176,7 +177,7 @@ async def list_agents( filter_obj=filter_dict or None, ) except Exception as exc: - logger.exception("Failed to list agents in workspace '%s'", workspace) + logger.exception("Failed to list agents in workspace '%s'", scrub(workspace)) raise HTTPException(status_code=500, detail="Failed to list agents.") from exc pagination = PaginationData.model_validate(result.pagination.model_dump()) if result.pagination else None @@ -208,7 +209,7 @@ async def get_agent( detail=f"Agent '{name}' not found in workspace '{workspace}'.", ) from exc except Exception as exc: - logger.exception("Failed to get agent '%s'", name) + logger.exception("Failed to get agent '%s'", scrub(name)) raise HTTPException(status_code=500, detail="Failed to get agent.") from exc return agent @@ -234,7 +235,7 @@ async def delete_agent( try: result = await entity_client.list(AgentDeployment, workspace=workspace) except Exception as exc: - logger.exception("Failed to list deployments before deleting agent '%s'", name) + logger.exception("Failed to list deployments before deleting agent '%s'", scrub(name)) raise HTTPException(status_code=500, detail="Failed to check deployments.") from exc blocking = [d for d in result.data if d.agent == name and d.status in _BLOCKING_STATUSES] @@ -256,5 +257,5 @@ async def delete_agent( detail=f"Agent '{name}' not found in workspace '{workspace}'.", ) from exc except Exception as exc: - logger.exception("Failed to delete agent '%s'", name) + logger.exception("Failed to delete agent '%s'", scrub(name)) raise HTTPException(status_code=500, detail="Failed to delete agent.") from exc diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py index 50106a3458..e8bdbee075 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py @@ -32,6 +32,7 @@ is_container_deployment_mode, is_external_agent, ) +from nemo_agents_plugin.log_utils import scrub from nemo_platform_plugin.authz import CallerKind, path_rule from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityNotFoundError @@ -98,7 +99,7 @@ async def _serve_agent_proxy( except NemoEntityNotFoundError as exc: raise HTTPException(status_code=404, detail=f"Agent '{name}' not found in workspace '{workspace}'.") from exc except Exception as exc: - logger.exception("Failed to look up agent '%s'", name) + logger.exception("Failed to look up agent '%s'", scrub(name)) raise HTTPException(status_code=500, detail="Failed to look up agent.") from exc if is_external_agent(agent): @@ -173,7 +174,7 @@ async def _serve_deployment_proxy( status_code=404, detail=f"Deployment '{name}' not found in workspace '{workspace}'." ) from exc except Exception as exc: - logger.exception("Failed to look up deployment '%s'", name) + logger.exception("Failed to look up deployment '%s'", scrub(name)) raise HTTPException(status_code=500, detail="Failed to look up deployment.") from exc if not _is_deployment_routable(dep): @@ -246,7 +247,7 @@ async def _resolve_agent_endpoint(name: str, workspace: str, entity_client: Nemo try: result = await entity_client.list(AgentDeployment, workspace=workspace) except Exception as exc: - logger.exception("Failed to list deployments for agent '%s'", name) + logger.exception("Failed to list deployments for agent '%s'", scrub(name)) raise HTTPException(status_code=500, detail="Failed to list deployments.") from exc running = [d for d in result.data if d.agent == name and _is_deployment_routable(d)] @@ -481,7 +482,10 @@ async def _stream_openai_from_a2a( yield _openai_chunk(completion_id, created, model, payload, None) except A2AMessageError as exc: logger.warning( - "External agent '%s' chat stream failed mid-stream (%s): %s", name, _endpoint_host(endpoint), exc + "External agent '%s' chat stream failed mid-stream (%s): %s", + scrub(name), + _endpoint_host(endpoint), + scrub(exc), ) note = f"[external agent error: {exc}]" payload = {"content": f"\n{note}"} if role_sent else {"role": "assistant", "content": note} @@ -517,7 +521,7 @@ async def external_agent_chat_completions( except NemoEntityNotFoundError as exc: raise HTTPException(status_code=404, detail=f"Agent '{name}' not found in workspace '{workspace}'.") from exc except Exception as exc: - logger.exception("Failed to look up agent '%s'", name) + logger.exception("Failed to look up agent '%s'", scrub(name)) raise HTTPException(status_code=500, detail="Failed to look up agent.") from exc if not is_external_agent(agent): @@ -580,7 +584,9 @@ async def _external_openai_chat( except StopAsyncIteration: first, has_first = "", False except A2AMessageError as exc: - logger.warning("External agent '%s' chat stream failed (%s): %s", name, _endpoint_host(endpoint), exc) + logger.warning( + "External agent '%s' chat stream failed (%s): %s", scrub(name), _endpoint_host(endpoint), scrub(exc) + ) raise HTTPException(status_code=502, detail=f"External agent chat failed: {exc}") from exc return StreamingResponse( _stream_openai_from_a2a(agen, first, has_first, model, name, endpoint), @@ -591,7 +597,7 @@ async def _external_openai_chat( try: reply = await send_a2a_message(endpoint, text) except A2AMessageError as exc: - logger.warning("External agent '%s' chat failed (%s): %s", name, _endpoint_host(endpoint), exc) + logger.warning("External agent '%s' chat failed (%s): %s", scrub(name), _endpoint_host(endpoint), scrub(exc)) raise HTTPException(status_code=502, detail=f"External agent chat failed: {exc}") from exc if wants_stream: return StreamingResponse(_single_chunk_sse(reply, model), media_type="text/event-stream") @@ -611,7 +617,9 @@ async def _serve_external_generate_full(name: str, endpoint: str, request: Reque try: reply = await send_a2a_message(endpoint, question) except A2AMessageError as exc: - logger.warning("External agent '%s' generate failed (%s): %s", name, _endpoint_host(endpoint), exc) + logger.warning( + "External agent '%s' generate failed (%s): %s", scrub(name), _endpoint_host(endpoint), scrub(exc) + ) raise HTTPException(status_code=502, detail=f"External agent generate failed: {exc}") from exc return StreamingResponse(_generate_full_stream(reply), media_type="text/event-stream") diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/log_utils.py b/plugins/nemo-agents/src/nemo_agents_plugin/log_utils.py new file mode 100644 index 0000000000..5a130c5031 --- /dev/null +++ b/plugins/nemo-agents/src/nemo_agents_plugin/log_utils.py @@ -0,0 +1,11 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Logging helpers.""" + +from __future__ import annotations + + +def scrub(value: object) -> str: + """Strip CR/LF from *value* so user-controlled text can't forge log lines.""" + return str(value).replace("\r", "").replace("\n", "") diff --git a/plugins/nemo-agents/tests/unit/test_a2a.py b/plugins/nemo-agents/tests/unit/test_a2a.py index 4a70473b22..23e8fd99a3 100644 --- a/plugins/nemo-agents/tests/unit/test_a2a.py +++ b/plugins/nemo-agents/tests/unit/test_a2a.py @@ -11,6 +11,7 @@ import pytest import respx from nemo_agents_plugin.a2a import ( + _ALLOW_PRIVATE_HOSTS_ENV, A2AMessageError, AgentCardError, extract_message_text, @@ -24,6 +25,12 @@ CARD = {"name": "Calculator Agent", "description": "does math", "skills": [{"id": "add", "name": "add"}]} +@pytest.fixture(autouse=True) +def _allow_private_hosts(monkeypatch: pytest.MonkeyPatch) -> None: + """Bypass the SSRF egress guard so behavioral tests can use mock hosts.""" + monkeypatch.setenv(_ALLOW_PRIVATE_HOSTS_ENV, "1") + + @pytest.mark.asyncio @respx.mock async def test_fetch_returns_card_from_primary_path() -> None: @@ -225,3 +232,47 @@ async def test_stream_caps_newlineless_body(monkeypatch: pytest.MonkeyPatch) -> ) with pytest.raises(A2AMessageError, match="size limit"): _ = [d async for d in stream_a2a_message("http://host:10000/", "hi")] + + +def _resolve_to(ip: str): + def _fake(host, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 + return [(2, 1, 6, "", (ip, 0))] + + return _fake + + +class TestSsrfGuard: + @pytest.mark.asyncio + @respx.mock + async def test_fetch_blocks_private_host(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(_ALLOW_PRIVATE_HOSTS_ENV, raising=False) + monkeypatch.setattr("socket.getaddrinfo", _resolve_to("169.254.169.254")) + route = respx.get("http://evil/.well-known/agent-card.json").mock(return_value=httpx.Response(200, json=CARD)) + with pytest.raises(AgentCardError): + await fetch_agent_card("http://evil") + assert not route.called + + @pytest.mark.asyncio + @respx.mock + async def test_send_blocks_loopback(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(_ALLOW_PRIVATE_HOSTS_ENV, raising=False) + monkeypatch.setattr("socket.getaddrinfo", _resolve_to("127.0.0.1")) + route = respx.post("http://evil/").mock(return_value=httpx.Response(200, json={"result": {}})) + with pytest.raises(A2AMessageError): + await send_a2a_message("http://evil/", "hi") + assert not route.called + + @pytest.mark.asyncio + async def test_probe_blocks_private_host(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(_ALLOW_PRIVATE_HOSTS_ENV, raising=False) + monkeypatch.setattr("socket.getaddrinfo", _resolve_to("10.0.0.5")) + assert await probe_agent_reachable("http://evil") is False + + @pytest.mark.asyncio + @respx.mock + async def test_public_host_allowed(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(_ALLOW_PRIVATE_HOSTS_ENV, raising=False) + monkeypatch.setattr("socket.getaddrinfo", _resolve_to("93.184.216.34")) + respx.get("http://public/.well-known/agent-card.json").mock(return_value=httpx.Response(200, json=CARD)) + card = await fetch_agent_card("http://public") + assert card["name"] == "Calculator Agent" diff --git a/plugins/nemo-agents/tests/unit/test_log_utils.py b/plugins/nemo-agents/tests/unit/test_log_utils.py new file mode 100644 index 0000000000..59f33598d5 --- /dev/null +++ b/plugins/nemo-agents/tests/unit/test_log_utils.py @@ -0,0 +1,17 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for log-value scrubbing.""" + +from __future__ import annotations + +from nemo_agents_plugin.log_utils import scrub + + +def test_scrub_strips_crlf() -> None: + assert scrub("ok\r\nINFO forged log line") == "okINFO forged log line" + + +def test_scrub_stringifies_non_str() -> None: + assert scrub(42) == "42" + assert scrub(RuntimeError("boom\nfake")) == "boomfake" From a606d5fc95080760e54f1edd0886900851bbada0 Mon Sep 17 00:00:00 2001 From: mschwab Date: Fri, 17 Jul 2026 13:24:54 -0700 Subject: [PATCH 06/10] fix(agents): scrub host in _endpoint_host to clear residual log-injection _endpoint_host output feeds logger calls whose URL arg is user-controlled; sanitize at the source so CodeQL clears py/log-injection #4280/#4281. Signed-off-by: mschwab --- plugins/nemo-agents/src/nemo_agents_plugin/a2a.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py index 272004c3ad..cf7df04540 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py @@ -164,7 +164,7 @@ async def fetch_agent_card(base_url: str) -> dict[str, Any]: def _endpoint_host(url: str) -> str: """Host of *url* for logging — avoid echoing the full URL back anywhere.""" try: - return urlparse(url).hostname or "unknown" + return scrub(urlparse(url).hostname or "unknown") except ValueError: return "unknown" From 11e13f2a689fefc1ec99db51cb2da55a4d4279c6 Mon Sep 17 00:00:00 2001 From: mschwab Date: Fri, 17 Jul 2026 14:42:04 -0700 Subject: [PATCH 07/10] fix(agents): encode config/url XOR in CreateAgentRequest OpenAPI schema Runtime enforces exactly-one-of config/url via model_validator, but the generated OpenAPI schema allowed both/neither. Add json_schema_extra oneOf so generated client contracts match runtime behavior. Signed-off-by: mschwab --- plugins/nemo-agents/openapi/openapi.yaml | 5 +++++ plugins/nemo-agents/src/nemo_agents_plugin/schema.py | 4 +++- plugins/nemo-agents/tests/unit/test_entities.py | 8 +++++++- 3 files changed, 15 insertions(+), 2 deletions(-) diff --git a/plugins/nemo-agents/openapi/openapi.yaml b/plugins/nemo-agents/openapi/openapi.yaml index c88908b1fc..60bcfa2189 100644 --- a/plugins/nemo-agents/openapi/openapi.yaml +++ b/plugins/nemo-agents/openapi/openapi.yaml @@ -2755,6 +2755,11 @@ components: - -updated_at title: AnalyzeJobsSortField CreateAgentRequest: + oneOf: + - required: + - config + - required: + - url properties: name: type: string diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/schema.py b/plugins/nemo-agents/src/nemo_agents_plugin/schema.py index d2f1cb0c25..8ad49908a4 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/schema.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/schema.py @@ -31,7 +31,7 @@ DeploymentStatus, ) from nemo_platform_plugin.schema import NemoFilter, NemoListResponse -from pydantic import BaseModel, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, model_validator # --------------------------------------------------------------------------- # Request bodies — plain BaseModel, named by convention @@ -47,6 +47,8 @@ class CreateAgentRequest(BaseModel): at creation). Provide exactly one of ``config`` or ``url``. """ + model_config = ConfigDict(json_schema_extra={"oneOf": [{"required": ["config"]}, {"required": ["url"]}]}) + name: str = Field(description="Unique agent name within the workspace.") description: str = Field( default="", diff --git a/plugins/nemo-agents/tests/unit/test_entities.py b/plugins/nemo-agents/tests/unit/test_entities.py index 7eef5bf9e2..078aecce7a 100644 --- a/plugins/nemo-agents/tests/unit/test_entities.py +++ b/plugins/nemo-agents/tests/unit/test_entities.py @@ -125,7 +125,7 @@ def test_status_transitions(self) -> None: def test_invalid_status_rejected(self) -> None: with pytest.raises(ValidationError): - AgentDeployment(name="dep", workspace="default", status="unknown") + AgentDeployment.model_validate({"name": "dep", "workspace": "default", "status": "unknown"}) def test_data_fields_include_deployment_fields(self) -> None: d = AgentDeployment( @@ -188,6 +188,12 @@ def test_custom_format(self) -> None: req = CreateAgentRequest(name="x", config={}, config_format="custom-v2") assert req.config_format == "custom-v2" + def test_schema_encodes_config_xor_url(self) -> None: + assert CreateAgentRequest.model_json_schema()["oneOf"] == [ + {"required": ["config"]}, + {"required": ["url"]}, + ] + # --------------------------------------------------------------------------- # Canonical spec-location helpers From d88167be4c464c1ea1ddd3f0a1749141547d4dbc Mon Sep 17 00:00:00 2001 From: mschwab Date: Wed, 22 Jul 2026 10:11:33 -0700 Subject: [PATCH 08/10] fix(agents): require POST to invoke external agents The agent-name proxy read route (GET/HEAD/OPTIONS) reaches _serve_external_agent, so a bodied GET could invoke an external agent under read scope. Guard the bridge with a 405 unless the method is POST. Signed-off-by: mschwab --- .../src/nemo_agents_plugin/api/v2/gateway.py | 8 ++++++++ .../nemo-agents/tests/unit/test_gateway.py | 19 +++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py index e8bdbee075..e8f6d7a7e5 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py @@ -633,6 +633,14 @@ async def _serve_external_agent( ``generate`` (nat eval) → NAT-generate↔A2A translation. Anything else 400s — external agents have no deployment to proxy arbitrary paths to. """ + # This bridge is reachable through the read-scoped GET/HEAD/OPTIONS proxy route, + # so a bodied GET must not be able to invoke the agent under read scope. + if request.method != "POST": + raise HTTPException( + status_code=405, + detail="External agent invocation requires POST.", + headers={"Allow": "POST"}, + ) endpoint = _external_a2a_endpoint(agent, name) if trailing_uri.endswith("chat/completions"): body = await _read_json_object(request) diff --git a/plugins/nemo-agents/tests/unit/test_gateway.py b/plugins/nemo-agents/tests/unit/test_gateway.py index 190e7e7bc5..8566d49ed1 100644 --- a/plugins/nemo-agents/tests/unit/test_gateway.py +++ b/plugins/nemo-agents/tests/unit/test_gateway.py @@ -901,3 +901,22 @@ def test_unsupported_path_returns_400(self, client: TestClient, mock_entity_clie ) assert resp.status_code == 400 + + def test_read_scoped_get_cannot_invoke( + self, client: TestClient, mock_entity_client: AsyncMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + # The GET/HEAD/OPTIONS proxy route reaches this bridge; a bodied GET must + # not be able to invoke the agent under read scope. + mock_entity_client.get = AsyncMock(return_value=_make_external_agent()) + send = AsyncMock(return_value="4") + monkeypatch.setattr(gateway_module, "send_a2a_message", send) + + resp = client.request( + "GET", + "/apis/agents/v2/workspaces/default/agents/ext/-/v1/chat/completions", + json={"messages": [{"role": "user", "content": "2+2?"}]}, + ) + + assert resp.status_code == 405 + assert resp.headers["allow"] == "POST" + send.assert_not_awaited() From 53a94375b20dfd13d6978913ab2ecde9593dcdcf Mon Sep 17 00:00:00 2001 From: mschwab Date: Wed, 22 Jul 2026 10:46:34 -0700 Subject: [PATCH 09/10] style(agents): any()-over-list address check, match/case message content Address review nits: rewrite _address_disallowed as any() over a list of flags, and dispatch OpenAI message content via match/case. Signed-off-by: mschwab --- plugins/nemo-agents/src/nemo_agents_plugin/a2a.py | 11 ++++++++++- .../src/nemo_agents_plugin/api/v2/gateway.py | 15 +++++++++------ 2 files changed, 19 insertions(+), 7 deletions(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py index cf7df04540..4e16ee8e9f 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py @@ -72,7 +72,16 @@ def _address_disallowed(ip_text: str) -> bool: ip = ipaddress.ip_address(ip_text) except ValueError: return True - return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast or ip.is_unspecified + return any( + [ + ip.is_private, + ip.is_loopback, + ip.is_link_local, + ip.is_reserved, + ip.is_multicast, + ip.is_unspecified, + ] + ) async def _ensure_host_allowed(url: str) -> None: diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py index e8f6d7a7e5..94e967c412 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/gateway.py @@ -378,12 +378,15 @@ def _message_text(message: Any) -> str: """Extract text from one OpenAI message (``content`` may be str or parts).""" if not isinstance(message, dict): return "" - content = message.get("content") - if isinstance(content, str): - return content - if isinstance(content, list): - return "".join(part["text"] for part in content if isinstance(part, dict) and isinstance(part.get("text"), str)) - return "" + match message.get("content"): + case str() as content: + return content + case list() as content: + return "".join( + part["text"] for part in content if isinstance(part, dict) and isinstance(part.get("text"), str) + ) + case _: + return "" def _conversation_prompt(messages: Any) -> str: From 459338b967336544fc0ddeac1f2c17ee3a698772 Mon Sep 17 00:00:00 2001 From: mschwab Date: Wed, 22 Jul 2026 10:55:42 -0700 Subject: [PATCH 10/10] fix(agents): drop endpoint from A2A transport errors The A2AMessageError from transport failures flows into gateway HTTP/SSE responses; an external agent endpoint can embed credentials. Drop the endpoint from the message, keeping only the exception class name. Signed-off-by: mschwab --- .../nemo-agents/src/nemo_agents_plugin/a2a.py | 4 ++-- plugins/nemo-agents/tests/unit/test_a2a.py | 21 ++++++++++++++++--- 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py index 4e16ee8e9f..e7f235b0c9 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/a2a.py @@ -267,7 +267,7 @@ async def send_a2a_message(endpoint: str, text: str) -> str: raise A2AMessageError(f"agent returned HTTP {resp.status_code}") raw = await _read_capped(resp, _MAX_MESSAGE_BYTES) except httpx.HTTPError as exc: - raise A2AMessageError(f"could not reach agent at {endpoint} ({exc.__class__.__name__})") from exc + raise A2AMessageError(f"could not reach agent ({exc.__class__.__name__})") from exc try: data = json.loads(raw) @@ -356,4 +356,4 @@ async def stream_a2a_message(endpoint: str, text: str) -> AsyncIterator[str]: if delta: yield delta except httpx.HTTPError as exc: - raise A2AMessageError(f"could not reach agent at {endpoint} ({exc.__class__.__name__})") from exc + raise A2AMessageError(f"could not reach agent ({exc.__class__.__name__})") from exc diff --git a/plugins/nemo-agents/tests/unit/test_a2a.py b/plugins/nemo-agents/tests/unit/test_a2a.py index 23e8fd99a3..9092da1909 100644 --- a/plugins/nemo-agents/tests/unit/test_a2a.py +++ b/plugins/nemo-agents/tests/unit/test_a2a.py @@ -176,9 +176,13 @@ async def test_send_message_jsonrpc_error_raises() -> None: @pytest.mark.asyncio @respx.mock async def test_send_message_transport_error_raises() -> None: - respx.post("http://host:10000/").mock(side_effect=httpx.ConnectError("refused")) - with pytest.raises(A2AMessageError, match="could not reach"): - await send_a2a_message("http://host:10000/", "hi") + endpoint = "http://user:s3cret@host:10000/" + respx.post(endpoint).mock(side_effect=httpx.ConnectError("refused")) + with pytest.raises(A2AMessageError, match="could not reach") as excinfo: + await send_a2a_message(endpoint, "hi") + # The endpoint may carry embedded credentials; it must not leak into the error. + assert "s3cret" not in str(excinfo.value) + assert "host:10000" not in str(excinfo.value) class TestExtractStreamDelta: @@ -222,6 +226,17 @@ async def test_stream_jsonrpc_error_raises() -> None: _ = [d async for d in stream_a2a_message("http://host:10000/", "hi")] +@pytest.mark.asyncio +@respx.mock +async def test_stream_transport_error_does_not_leak_endpoint() -> None: + endpoint = "http://user:s3cret@host:10000/" + respx.post(endpoint).mock(side_effect=httpx.ConnectError("refused")) + with pytest.raises(A2AMessageError, match="could not reach") as excinfo: + _ = [d async for d in stream_a2a_message(endpoint, "hi")] + assert "s3cret" not in str(excinfo.value) + assert "host:10000" not in str(excinfo.value) + + @pytest.mark.asyncio @respx.mock async def test_stream_caps_newlineless_body(monkeypatch: pytest.MonkeyPatch) -> None: