implemnt plugin model for ResultSink

Signed-off-by: Peter Jausovec <[email protected]>

Author: peterj

examples/README.md

@@ -119,6 +119,12 @@ The zero-code and SDK examples implement the same toy agent (dice rolling + prim
 |---------|-------------|
 | [kubernetes/](./kubernetes/) | Deploy agentevals with kagent on Kubernetes using native OTLP gRPC ingestion (or optionally an OTel Collector). Includes a walkthrough for comparing two kagent agents (different models) and evaluating them with tool trajectory and response match scores. |
 
+## Custom result sinks
+
+Plugins can deliver run results (partial metrics, final summary, errors) to arbitrary backends alongside the database. Install a package that declares `[project.entry-points."agentevals.sinks"]`, restart agentevals, then reference the plugin’s `kind` in `spec.sinks` on `POST /api/runs`.
+
+See [custom_sink/README.md](./custom_sink/README.md) for a minimal setuptools plugin and configuration examples.
+
 ## Advanced: GenAI Semantic Convention Patterns
 
 > [!TIP]

examples/custom_sink/README.md

@@ -0,0 +1,80 @@
+# Custom result sink plugin
+
+This folder is a tiny installable Python package that registers a result **sink** with agentevals via setuptools **entry points**. The worker fans out partial/final/error events to every configured sink in addition to the database.
+
+## What gets implemented
+
+- **`DemoNdjsonSink`** — subclasses `ResultSink` from `agentevals.run.sinks` and appends one JSON object per line to `path` from the run spec (same pattern as the built-in `file` sink, with a `"demo": true` marker on each line).
+- **`create_demo_sink(spec)`** — factory callable; must accept the full sink dict from the run spec and return a `ResultSink` (see return type in code).
+
+The entry point **name** (`demo_ndjson` in `pyproject.toml`) is the **`kind`** string clients put under `spec.sinks`.
+
+## Install (local dev)
+
+From the agentevals repo root, install the framework first, then this example:
+
+```bash
+uv pip install -e .
+uv pip install -e examples/custom_sink
+```
+
+Restart the agentevals process so `importlib.metadata` picks up the new distribution.
+
+PyPI-style usage is the same: depend on `agentevals-example-custom-sink` next to `agentevals-cli`, install both into the server environment, restart.
+
+## Configure runs
+
+Async runs are submitted with **`POST /api/runs`**. Put your sink in **`spec.sinks`** (requires Postgres storage — see main docs).
+
+Example body (use **absolute** `path` on the host where the agentevals process runs when possible). **`path` must be a file path** (e.g. `/tmp/demo.ndjson`). If `path` is an **existing directory** (including `"."` for the process working directory), output goes to `<path>/agentevals-demo-sink.ndjson`, or `<path>/<filename>` if you add an optional `"filename"` field next to `path` in the sink dict.
+
+The `inline` object must contain real trace data (Jaeger JSON or OTLP), not an empty object.
+
+```json
+{
+  "spec": {
+    "approach": "trace_replay",
+    "target": {
+      "kind": "inline",
+      "traceFormat": "jaeger-json",
+      "inline": {
+        "data": [
+          {
+            "traceID": "61646461646164646164616461646164",
+            "spans": [
+              {
+                "traceID": "61646461646164646164616461646164",
+                "spanID": "6164616461646164",
+                "operationName": "demo-op",
+                "startTime": 1000000,
+                "duration": 100000,
+                "tags": [],
+                "logs": [],
+                "references": [],
+                "processID": "p1"
+              }
+            ],
+            "processes": { "p1": { "serviceName": "demo" } }
+          }
+        ]
+      }
+    },
+    "sinks": [{ "kind": "demo_ndjson", "path": "/tmp/agentevals-demo.ndjson" }]
+  }
+}
+```
+
+You can list several sinks; they run in parallel. Built-in kinds are `stdout`, `file`, and `http_webhook`.
+
+## Publishing your own sink
+
+1. Implement `ResultSink` from `agentevals.run.sinks` (subclass the protocol, or provide the three async methods).
+2. Expose a factory `def create_*(spec: dict) -> ResultSink`.
+3. Add the following to your `pyproject.toml`:
+
+```toml
+[project.entry-points."agentevals.sinks"]
+your_kind = "your_package.module:your_factory"
+```
+
+4. Install the package into the **same environment** as `agentevals serve`, restart, and reference `"kind": "your_kind"` in `spec.sinks`.

examples/custom_sink/agentevals_example_custom_sink/__init__.py

@@ -0,0 +1 @@
+"""Example result sink plugin for agentevals (see README)."""

examples/custom_sink/agentevals_example_custom_sink/sink.py

@@ -0,0 +1,71 @@
+"""Minimal NDJSON sink registered via setuptools entry points."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+from agentevals.run.sinks import ResultSink
+from agentevals.storage.models import Result
+
+
+def _result_payload(r: Result) -> dict:
+    return r.model_dump(mode="json", by_alias=True)
+
+
+_DEFAULT_FILENAME = "agentevals-demo-sink.ndjson"
+
+
+def _resolve_output_file(spec: dict[str, Any]) -> Path:
+    """If ``path`` is an existing directory (including ``.``), write NDJSON inside it."""
+    p = Path(spec["path"]).expanduser()
+    if p.exists() and p.is_dir():
+        name = spec.get("filename") or _DEFAULT_FILENAME
+        return p / name
+    return p
+
+
+class DemoNdjsonSink(ResultSink):
+    """Concrete :class:`~agentevals.run.sinks.ResultSink`; append-only JSON lines with a ``demo`` marker."""
+
+    def __init__(self, path: Path) -> None:
+        self._path = path
+        self._lock = asyncio.Lock()
+
+    async def _write(self, payload: dict) -> None:
+        async with self._lock:
+            self._path.parent.mkdir(parents=True, exist_ok=True)
+            with self._path.open("a") as f:  # noqa: ASYNC230
+                f.write(json.dumps(payload) + "\n")
+
+    async def emit_partial(self, run_id: UUID, results: list[Result], attempt: int) -> None:
+        for r in results:
+            await self._write(
+                {
+                    "phase": "partial",
+                    "run_id": str(run_id),
+                    "attempt": attempt,
+                    "demo": True,
+                    "result": _result_payload(r),
+                }
+            )
+
+    async def emit_final(self, run_id: UUID, summary: dict, attempt: int) -> None:
+        await self._write(
+            {"phase": "final", "run_id": str(run_id), "attempt": attempt, "demo": True, "summary": summary}
+        )
+
+    async def emit_error(self, run_id: UUID, error: str, attempt: int) -> None:
+        await self._write({"phase": "error", "run_id": str(run_id), "attempt": attempt, "demo": True, "error": error})
+
+
+def create_demo_sink(spec: dict[str, Any]) -> ResultSink:
+    """Entry-point factory: returns a :class:`ResultSink`; ``kind`` must be ``demo_ndjson`` (see pyproject).
+
+    ``path`` should normally be a **file** path. If it points at an existing directory (e.g. ``.`` or ``/tmp``),
+    lines are appended to ``<path>/agentevals-demo-sink.ndjson``, or ``<path>/<filename>`` if ``filename`` is set.
+    """
+    return DemoNdjsonSink(_resolve_output_file(spec))

examples/custom_sink/pyproject.toml

@@ -0,0 +1,19 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "agentevals-example-custom-sink"
+version = "0.1.0"
+description = "Example setuptools plugin that registers an agentevals result sink"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "agentevals-cli>=0.7.0",
+]
+
+[project.entry-points."agentevals.sinks"]
+demo_ndjson = "agentevals_example_custom_sink.sink:create_demo_sink"
+
+[tool.hatch.build.targets.wheel]
+packages = ["agentevals_example_custom_sink"]

src/agentevals/api/app.py

@@ -17,6 +17,7 @@
 from agentevals import __version__
 
 from ..run.service import RunService
+from ..run.sinks import log_registered_sinks
 from ..run.worker import AsyncRunWorker
 from ..storage import StorageSettings, build_repos
 from ..storage.postgres.migrator import Migrator
@@ -83,6 +84,7 @@ async def lifespan(app: FastAPI):
             worker = AsyncRunWorker(runs=repos.runs, results=repos.results, settings=storage_settings)
             await worker.start()
             app.state.run_worker = worker
+            log_registered_sinks()
 
         yield
 

src/agentevals/run/__init__.py

@@ -2,7 +2,7 @@
 
 Contents:
 - :mod:`fetcher` resolves a run spec's ``target`` into a list of traces.
-- :mod:`sinks` fan-out result delivery (stdout, file, http_webhook).
+- :mod:`sinks` fan-out result delivery (built-ins plus setuptools plugins / :func:`~agentevals.run.sinks.register_sink_factory`).
 - :mod:`service` is the synchronous control surface used by HTTP handlers.
 - :mod:`worker` is the in-process loop that claims runs and drives the
   existing :func:`agentevals.runner.run_evaluation_from_traces` pipeline.

src/agentevals/run/sinks.py

@@ -3,6 +3,14 @@
 The :class:`agentevals.storage.repos.ResultRepository` is always written;
 sinks are an additional delivery channel. Sink failures are logged with
 ``run_id`` / ``result_id`` but do not fail the run.
+
+**Plugins:** third-party packages declare setuptools entry points in group
+``agentevals.sinks`` (entry **name** = ``kind`` string; **value** = ``module:factory``
+callable ``factory(spec: dict) -> ResultSink``). Built-in kinds
+(``stdout``, ``file``, ``http_webhook``) are not overridden by entry points;
+hosts may replace any kind via :func:`register_sink_factory` (highest precedence).
+
+Tests may call :func:`clear_sink_plugin_registry` to drop programmatic registrations.
 """
 
 from __future__ import annotations
@@ -12,8 +20,10 @@
 import logging
 import os
 import sys
+from collections.abc import Callable
+from importlib.metadata import entry_points
 from pathlib import Path
-from typing import Any, Protocol
+from typing import Any, Protocol, cast
 from uuid import UUID
 
 import httpx
@@ -22,13 +32,20 @@
 
 logger = logging.getLogger(__name__)
 
+SINK_ENTRY_POINT_GROUP = "agentevals.sinks"
+
 
 class ResultSink(Protocol):
     async def emit_partial(self, run_id: UUID, results: list[Result], attempt: int) -> None: ...
     async def emit_final(self, run_id: UUID, summary: dict, attempt: int) -> None: ...
     async def emit_error(self, run_id: UUID, error: str, attempt: int) -> None: ...
 
 
+SinkFactory = Callable[[dict[str, Any]], ResultSink]
+
+_PLUGIN_FACTORIES: dict[str, SinkFactory] = {}
+
+
 def _result_payload(r: Result) -> dict:
     return r.model_dump(mode="json", by_alias=True)
 
@@ -187,33 +204,18 @@ async def _guard(coro: Any, phase: str) -> None:
             logger.exception("sink delivery failed in phase=%s", phase)
 
 
-def build_sinks(specs: list[dict]) -> SinkFanout:
-    """Construct a fan-out from the run spec's ``sinks`` array.
+def register_sink_factory(kind: str, factory: SinkFactory) -> None:
+    """Register or replace the factory for ``kind`` (overrides built-ins and entry points).
 
-    Each spec is a dict with ``kind`` plus kind-specific args. Unknown kinds
-    are skipped with a warning so a future kind added by a host doesn't
-    break older agentevals replicas mid-rollout.
+    Call during process startup before run workers consume specs. The factory receives
+    the full sink spec dict (including ``kind``) and returns a :class:`ResultSink`.
     """
-    sinks: list[ResultSink] = []
-    for spec in specs:
-        kind = spec.get("kind")
-        if kind == "stdout":
-            sinks.append(StdoutSink())
-        elif kind == "file":
-            sinks.append(FileSink(spec["path"]))
-        elif kind == "http_webhook":
-            sinks.append(
-                HttpWebhookSink(
-                    url=spec["url"],
-                    headers=spec.get("headers"),
-                    headers_from_env=spec.get("headers_from_env") or _extract_env_headers(spec.get("auth")),
-                    timeout_s=float(spec.get("timeout_s", 10.0)),
-                    max_attempts=int(spec.get("max_attempts", 5)),
-                )
-            )
-        else:
-            logger.warning("unknown sink kind '%s'; skipping", kind)
-    return SinkFanout(sinks)
+    _PLUGIN_FACTORIES[kind] = factory
+
+
+def clear_sink_plugin_registry() -> None:
+    """Drop all registrations from :func:`register_sink_factory` (for tests)."""
+    _PLUGIN_FACTORIES.clear()
 
 
 def _extract_env_headers(auth: Any) -> dict[str, str]:
@@ -228,3 +230,85 @@ def _extract_env_headers(auth: Any) -> dict[str, str]:
         if isinstance(value, dict) and "from_env" in value:
             result[header_name] = value["from_env"]
     return result
+
+
+def _http_webhook_from_spec(spec: dict[str, Any]) -> HttpWebhookSink:
+    return HttpWebhookSink(
+        url=spec["url"],
+        headers=spec.get("headers"),
+        headers_from_env=spec.get("headers_from_env") or _extract_env_headers(spec.get("auth")),
+        timeout_s=float(spec.get("timeout_s", 10.0)),
+        max_attempts=int(spec.get("max_attempts", 5)),
+    )
+
+
+def _builtin_factories() -> dict[str, SinkFactory]:
+    return {
+        "stdout": lambda _spec: StdoutSink(),
+        "file": lambda spec: FileSink(spec["path"]),
+        "http_webhook": _http_webhook_from_spec,
+    }
+
+
+def _merge_sink_factories() -> dict[str, SinkFactory]:
+    """Built-ins, then entry points (no built-in shadowing), then programmatic overrides."""
+    merged: dict[str, SinkFactory] = dict(_builtin_factories())
+    eps = entry_points(group=SINK_ENTRY_POINT_GROUP)
+    for ep in eps:
+        if ep.name in merged:
+            logger.debug("skipping sink entry point %r; built-in kind takes precedence", ep.name)
+            continue
+        try:
+            loaded = ep.load()
+            if not callable(loaded):
+                logger.warning("sink entry point %r is not callable; skipping", ep.name)
+                continue
+            merged[ep.name] = cast(SinkFactory, loaded)
+        except Exception:
+            logger.exception("failed to load sink entry point %r", ep.name)
+    merged.update(_PLUGIN_FACTORIES)
+    return merged
+
+
+def registered_sink_kinds() -> tuple[str, ...]:
+    """Sorted sink ``kind`` strings that would resolve if :func:`build_sinks` ran now.
+
+    Includes built-ins, successfully loaded setuptools entry points for group
+    :data:`SINK_ENTRY_POINT_GROUP`, and registrations from
+    :func:`register_sink_factory`. The tuple reflects current process state and
+    can change if the programmatic registry is mutated after startup.
+    """
+    return tuple(sorted(_merge_sink_factories().keys()))
+
+
+def log_registered_sinks() -> None:
+    """Emit one INFO line listing available sink kinds (for operator diagnostics)."""
+    kinds = registered_sink_kinds()
+    logger.info("Result sinks available (%d kinds): %s", len(kinds), ", ".join(kinds))
+
+
+def build_sinks(specs: list[dict]) -> SinkFanout:
+    """Construct a fan-out from the run spec's ``sinks`` array.
+
+    Each spec is a dict with ``kind`` plus kind-specific args. Unknown kinds
+    are skipped with a warning so a future kind added by a host doesn't
+    break older agentevals replicas mid-rollout.
+
+    Factory lookup starts from built-ins, adds setuptools entry points (group
+    ``agentevals.sinks``) for ``kind`` names not already built-in, then applies
+    :func:`register_sink_factory` registrations, which override any prior factory
+    for the same ``kind``. See :func:`_merge_sink_factories`.
+    """
+    factories = _merge_sink_factories()
+    sinks: list[ResultSink] = []
+    for spec in specs:
+        kind = spec.get("kind")
+        factory = factories.get(kind) if kind is not None else None
+        if factory is None:
+            logger.warning("unknown sink kind '%s'; skipping", kind)
+            continue
+        try:
+            sinks.append(factory(spec))
+        except Exception:
+            logger.exception("sink factory failed for kind=%s", kind)
+    return SinkFanout(sinks)