Evaluate a Deployed Agent over HTTP

View as Markdown

The 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.

This guide assumes you’ve done the quickstart (install, tasks, metrics, run()). It reuses that quickstart’s KeywordMatchMetric and tasks.

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.
1from nemo_evaluator_sdk.enums import AgentFormat
2from nemo_evaluator_sdk.values import GenericAgent
3
4agent = GenericAgent(
5 name="my-http-agent",
6 url="http://127.0.0.1:8000/agent",
7 format=AgentFormat.GENERIC,
8 body={"message": "{{ instruction }}"}, # -> {"message": "What is the capital of France?"}
9 response_path="$.answer", # reads {"answer": "..."} from the response
10 # api_key_secret="MY_AGENT_TOKEN", # optional: env var (local run) or platform secret (job), sent as a bearer token
11)

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.

1import json
2import threading
3from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
4
5
6class _AgentHandler(BaseHTTPRequestHandler):
7 def do_POST(self) -> None:
8 payload = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))) or b"{}")
9 text = str(payload.get("message", "")).lower()
10 answer = "Paris" if "france" in text else "Tokyo" if "japan" in text else "unsure"
11 body = json.dumps({"answer": answer}).encode()
12 self.send_response(200)
13 self.send_header("Content-Type", "application/json")
14 self.end_headers()
15 self.wfile.write(body)
16
17 def log_message(self, *args: object) -> None:
18 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.

1from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
2from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig
3
4result = await AgentEvaluator().run(
5 tasks=make_tasks(), # from the quickstart
6 target=agent,
7 config=AgentEvalRunConfig(parallelism=2),
8)
9
10for aggregate in result.summary.scores.scores:
11 print(f"{aggregate.name}: {aggregate.mean}")

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.

Full script

1import asyncio
2import json
3import threading
4from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
5
6from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
7from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask
8from nemo_evaluator_sdk.enums import AgentFormat
9from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult
10from nemo_evaluator_sdk.values import GenericAgent
11
12
13class KeywordMatchMetric:
14 @property
15 def type(self) -> str:
16 return "keyword_match"
17
18 def output_spec(self) -> list[MetricOutputSpec]:
19 return [MetricOutputSpec.continuous_score("score")]
20
21 async def compute_scores(self, input: MetricInput) -> MetricResult:
22 expected = str(input.row.data.get("reference", {}).get("expected", "")).lower()
23 answer = (input.candidate.output_text or "").lower()
24 return MetricResult(outputs=[MetricOutput(name="score", value=1.0 if expected and expected in answer else 0.0)])
25
26
27def make_tasks() -> list[AgentEvalTask]:
28 return [
29 AgentEvalTask(id="capital-france", intent="Answer the geography question.",
30 inputs={"instruction": "What is the capital of France?"},
31 reference={"expected": "Paris"}, metrics=[KeywordMatchMetric()]),
32 AgentEvalTask(id="capital-japan", intent="Answer the geography question.",
33 inputs={"instruction": "What is the capital of Japan?"},
34 reference={"expected": "Tokyo"}, metrics=[KeywordMatchMetric()]),
35 ]
36
37
38# A stand-in "deployed agent": a tiny local HTTP server. Replace with your real agent's URL.
39class _AgentHandler(BaseHTTPRequestHandler):
40 def do_POST(self) -> None:
41 payload = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))) or b"{}")
42 text = str(payload.get("message", "")).lower()
43 answer = "Paris" if "france" in text else "Tokyo" if "japan" in text else "unsure"
44 body = json.dumps({"answer": answer}).encode()
45 self.send_response(200)
46 self.send_header("Content-Type", "application/json")
47 self.end_headers()
48 self.wfile.write(body)
49
50 def log_message(self, *args: object) -> None:
51 return
52
53
54async def main() -> None:
55 httpd = ThreadingHTTPServer(("127.0.0.1", 0), _AgentHandler)
56 threading.Thread(target=httpd.serve_forever, daemon=True).start()
57 host, port = httpd.server_address
58
59 agent = GenericAgent(
60 name="my-http-agent",
61 url=f"http://{host}:{port}/agent",
62 format=AgentFormat.GENERIC,
63 body={"message": "{{ instruction }}"},
64 response_path="$.answer",
65 )
66 try:
67 result = await AgentEvaluator().run(
68 tasks=make_tasks(),
69 target=agent,
70 config=AgentEvalRunConfig(parallelism=2),
71 )
72 finally:
73 httpd.shutdown()
74
75 for aggregate in result.summary.scores.scores:
76 print(f"{aggregate.name}: {aggregate.mean}")
77 for trial in result.trials:
78 print(f" {trial.task_id}: {trial.output.output_text!r}")
79
80
81asyncio.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

  • 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.