---
# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

title: "Evaluate a Deployed Agent over HTTP"
description: "Point NeMo Evaluator at an agent reachable over HTTP by describing it as a GenericAgent target — the request to send and where the answer is in the response — then score it exactly like the quickstart."
---

The [quickstart](/documentation/evaluate-models/agent-eval/quickstart) evaluated an in-process agent.
This guide evaluates an agent that runs **behind an HTTP endpoint** — a service you (or someone else)
deployed. Everything else is the same: the same tasks, the same metrics, the same `run()`. Only the
**target** changes.

You describe an HTTP agent as a `GenericAgent`: its URL, the JSON request to send, and where the
answer lives in the response. The evaluator then calls that endpoint over real HTTP for each task.

<Note>

This guide assumes you've done the [quickstart](/documentation/evaluate-models/agent-eval/quickstart)
(install, tasks, metrics, `run()`). It reuses that quickstart's `KeywordMatchMetric` and tasks.

</Note>

## 1. Describe the agent

A `GenericAgent` has three parts you configure:

- **`url`** — where the agent is reachable.
- **`body`** — the JSON request to POST, as a Jinja template rendered against the task's `inputs`.
  Reference them directly: the task's instruction is `{{ instruction }}`. The payload is entirely
  yours — the evaluator sends exactly what `body` produces.
- **`response_path`** — a JSONPath into the response that selects the agent's answer.

```python
from nemo_evaluator_sdk.enums import AgentFormat
from nemo_evaluator_sdk.values import GenericAgent

agent = GenericAgent(
    name="my-http-agent",
    url="http://127.0.0.1:8000/agent",
    format=AgentFormat.GENERIC,
    body={"message": "{{ instruction }}"},  # -> {"message": "What is the capital of France?"}
    response_path="$.answer",               # reads {"answer": "..."} from the response
    # api_key_secret="MY_AGENT_TOKEN",      # optional: env var (local run) or platform secret (job), sent as a bearer token
)
```

Set `body` and `response_path` to match **your** agent's request and response shape. Uncomment
`api_key_secret` for an agent that needs a token — for a local `run()` it names an environment
variable in your process, and for a submitted job it names a platform secret in the workspace; either
way its value is sent as a bearer token on each request. The local stand-in below needs no auth, so
this walkthrough leaves it off.

## 2. Run it against a local stand-in

To try this end to end without deploying anything, stand up a tiny local HTTP server that plays the
role of the agent. In practice you'd skip this and point `url` at your real endpoint.

```python
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class _AgentHandler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        payload = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))) or b"{}")
        text = str(payload.get("message", "")).lower()
        answer = "Paris" if "france" in text else "Tokyo" if "japan" in text else "unsure"
        body = json.dumps({"answer": answer}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format: str, *args: object) -> None:
        return  # keep the demo output quiet
```

## 3. Run the evaluation

With an agent target and no inference function supplied, `run()` generates answers by making real HTTP
calls to the agent's `url`.

```python
import asyncio

from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig

async def run_evaluation() -> None:
    result = await AgentEvaluator().run(
        tasks=make_tasks(),                 # from the quickstart
        target=agent,
        config=AgentEvalRunConfig(parallelism=2),
    )

    for aggregate in result.summary.scores.scores:
        print(f"{aggregate.name}: {aggregate.mean}")

asyncio.run(run_evaluation())
```

## Point at your real agent

To evaluate your own deployed agent, change three things and drop the local server:

1. **`url`** → your agent's endpoint.
2. **`body`** → the request your agent expects (reference the task's inputs, e.g. `{{ instruction }}`).
3. **`response_path`** → the JSONPath to the answer in your agent's response.

Add `api_key_secret="MY_AGENT_TOKEN"` if it needs auth (an env var for a local run, a platform secret for a job). If your agent is a **NeMo Agent Toolkit**
workflow, use `NemoAgentToolkitAgent` instead of `GenericAgent` — it targets a NAT endpoint's fixed
protocol, so you don't hand-write a `body` template.

## Streaming SSE agents

Set `stream=True` when the endpoint returns JSON Server-Sent Events. The evaluator applies
`response_path` to every `data:` payload and keeps the last matching value as the candidate output.
The raw evidence retains `event:` lines, but only payload fields such as `data:` become parsed stream
frames. An application-owned stream translator can turn those frames into an ATIF trajectory without
changing the shared HTTP transport.

The following example demonstrates how to evaluate a Hermes agent that streams SSE responses with
`GenericAgent`.

To set up Hermes locally:

- Follow the [Hermes quickstart](https://hermes-agent.nousresearch.com/docs/getting-started/quickstart).
- After the agent is running and tested with `hermes --tui`, [set up the local gateway](https://hermes-agent.nousresearch.com/docs/user-guide/features/api-server#1-enable-the-api-server) to interact with the agent over its API.
  1. Set `API_SERVER_ENABLED=true` and `API_SERVER_KEY=<key>` in `~/.hermes/.env`.
  1. Start the API server with `hermes gateway`.
  1. Export the same key as `HERMES_TOKEN` for the evaluator client.

The [complete Hermes example](https://github.com/NVIDIA-NeMo/nemo-platform/blob/main/packages/nemo_evaluator_sdk/examples/hermes/example.py)
defines `HermesStreamTranslator`, which implements the [AgentStreamTranslator](https://github.com/NVIDIA-NeMo/nemo-platform/blob/main/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_stream_translation.py)
protocol, and scores whether the deployed agent returns the requested phrase `streaming works`:

```python
from functools import partial

from nemo_evaluator_sdk import GenericAgent, SecretRef
from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
from nemo_evaluator_sdk.agent_inference import make_agent_inference_fn

agent = GenericAgent(
    name="hermes-agent",
    url="http://127.0.0.1:8642/v1/responses",
    api_key_secret=SecretRef("HERMES_TOKEN"),
    body={
        "model": "hermes-agent",
        "input": "{{ instruction }}",
        "stream": True,
        "store": False,
    },
    response_path="$..text",
    stream=True,
)

evaluator = AgentEvaluator(
    agent_inference_fn_factory=partial(
        make_agent_inference_fn,
        # Copy HermesStreamTranslator from the complete example linked above.
        stream_translator=HermesStreamTranslator(),
        capture_evidence=True,
    ),
    default_headers={"Accept": "text/event-stream"},
)
```

The SDK sends the bearer token from the `HERMES_TOKEN` environment variable, preserves the raw stream as evidence, and
passes the extracted output text to the task's metrics. The Hermes translator reads the terminal
`response.completed` payload, validates completion, and records user/agent steps plus token usage in
ATIF-v1.7.

## Full script

```python
import asyncio
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask
from nemo_evaluator_sdk.enums import AgentFormat
from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult
from nemo_evaluator_sdk.values import GenericAgent

class KeywordMatchMetric:
    @property
    def type(self) -> str:
        return "keyword_match"

    def output_spec(self) -> list[MetricOutputSpec]:
        return [MetricOutputSpec.continuous_score("score")]

    async def compute_scores(self, input: MetricInput) -> MetricResult:
        expected = str(input.row.data.get("reference", {}).get("expected", "")).lower()
        answer = (input.candidate.output_text or "").lower()
        return MetricResult(outputs=[MetricOutput(name="score", value=1.0 if expected and expected in answer else 0.0)])

def make_tasks() -> list[AgentEvalTask]:
    return [
        AgentEvalTask(id="capital-france", intent="Answer the geography question.",
                      inputs={"instruction": "What is the capital of France?"},
                      reference={"expected": "Paris"}, metrics=[KeywordMatchMetric()]),
        AgentEvalTask(id="capital-japan", intent="Answer the geography question.",
                      inputs={"instruction": "What is the capital of Japan?"},
                      reference={"expected": "Tokyo"}, metrics=[KeywordMatchMetric()]),
    ]

# A stand-in "deployed agent": a tiny local HTTP server. Replace with your real agent's URL.
class _AgentHandler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        payload = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))) or b"{}")
        text = str(payload.get("message", "")).lower()
        answer = "Paris" if "france" in text else "Tokyo" if "japan" in text else "unsure"
        body = json.dumps({"answer": answer}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format: str, *args: object) -> None:
        return

async def main() -> None:
    httpd = ThreadingHTTPServer(("127.0.0.1", 0), _AgentHandler)
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    host, port = httpd.server_address[0], httpd.server_address[1]

    agent = GenericAgent(
        name="my-http-agent",
        url=f"http://{host}:{port}/agent",
        format=AgentFormat.GENERIC,
        body={"message": "{{ instruction }}"},
        response_path="$.answer",
    )
    try:
        result = await AgentEvaluator().run(
            tasks=make_tasks(),
            target=agent,
            config=AgentEvalRunConfig(parallelism=2),
        )
    finally:
        httpd.shutdown()

    for aggregate in result.summary.scores.scores:
        print(f"{aggregate.name}: {aggregate.mean}")
    for trial in result.trials:
        output_text = trial.output.output_text if trial.output else None
        print(f"  {trial.task_id}: {output_text!r}")

asyncio.run(main())
```

Expected output — the answers came back over HTTP and both contain the expected keyword:

```
keyword_match.score: 1.0
  capital-france: 'Paris'
  capital-japan: 'Tokyo'
```

## Next steps

<Cards>
  <Card title="Agent Evaluation (concepts)" href="/documentation/evaluate-models/agent-eval" />
  <Card title="Agent Evaluation Quickstart" href="/documentation/evaluate-models/agent-eval/quickstart" />
</Cards>

- Score the agent's **trajectory** (its tool use), not just the final answer.
- Give a task **multiple metrics** and combine them into a named **view**.
