feat: add EvaluationClient with run() for on-demand session evaluation (aws#300)

* feat: add EvaluationClient with run() for on-demand session evaluation

EvaluationClient collects spans from CloudWatch and calls the evaluate
API with level-aware batching (SESSION/TRACE/TOOL_CALL). Accepts
evaluator_ids, session_id, and agent_id or log_group_name. Auto-derives
log group from agent_id, caches evaluator level lookups, and batches
evaluate requests at max 10 target IDs per request.

* refactor: simplify EvaluationClient and let exceptions propagate

Remove try/except around evaluate() so errors propagate to the caller
instead of being silently swallowed. Simplify _extract_trace_ids with
dict.fromkeys(), inline _batch() into list comprehensions, and remove
the evaluator_result_count tracking variable.

* style: fix trailing newlines from ruff format

Author: jariy17

src/bedrock_agentcore/evaluation/__init__.py

@@ -1,5 +1,6 @@
-"""AgentCore Evaluation integration for Strands."""
+"""AgentCore Evaluation: EvaluationClient and Strands integration."""
 
+from bedrock_agentcore.evaluation.client import EvaluationClient
 from bedrock_agentcore.evaluation.integrations.strands_agents_evals.evaluator import (
     StrandsEvalsAgentCoreEvaluator,
     create_strands_evaluator,
@@ -12,8 +13,9 @@
 )
 
 __all__ = [
-    "create_strands_evaluator",
+    "EvaluationClient",
     "StrandsEvalsAgentCoreEvaluator",
     "convert_strands_to_adot",
+    "create_strands_evaluator",
     "fetch_spans_from_cloudwatch",
 ]

src/bedrock_agentcore/evaluation/_agent_span_collector/__init__.py

@@ -0,0 +1,8 @@
+"""Agent span collector: collects OpenTelemetry spans for evaluation."""
+
+from .agent_span_collector import AgentSpanCollector, CloudWatchAgentSpanCollector
+
+__all__ = [
+    "AgentSpanCollector",
+    "CloudWatchAgentSpanCollector",
+]

src/bedrock_agentcore/evaluation/_agent_span_collector/agent_span_collector.py

@@ -0,0 +1,127 @@
+"""Span collector abstraction for the evaluation runner."""
+
+import logging
+import time
+from abc import ABC, abstractmethod
+from datetime import datetime, timedelta
+from typing import List
+
+from bedrock_agentcore._utils.endpoints import DEFAULT_REGION
+from bedrock_agentcore.evaluation.utils.cloudwatch_span_helper import CloudWatchSpanHelper
+
+logger = logging.getLogger(__name__)
+
+AWS_SPANS_LOG_GROUP = "aws/spans"
+
+
+class AgentSpanCollector(ABC):
+    """Abstract base class for collecting spans after agent invocation."""
+
+    @abstractmethod
+    def collect(self, session_id: str, start_time: datetime, end_time: datetime) -> List[dict]:
+        """Collect spans for a given session.
+
+        Args:
+            session_id: The session ID to collect spans for.
+            start_time: The start time of the session invocation.
+            end_time: The end time of the session invocation.
+
+        Returns:
+            List of span dictionaries.
+        """
+
+
+class CloudWatchAgentSpanCollector(AgentSpanCollector):
+    """Collects spans from CloudWatch using precise attributes.session.id filtering."""
+
+    def __init__(
+        self,
+        log_group_name: str,
+        region: str = DEFAULT_REGION,
+        max_wait_seconds: int = 300,
+        poll_interval_seconds: int = 30,
+    ):
+        """Initialize the CloudWatch span collector.
+
+        Args:
+            log_group_name: CloudWatch log group name for event logs.
+            region: AWS region for CloudWatch client.
+            max_wait_seconds: Maximum time to poll for spans before giving up (default 300s).
+            poll_interval_seconds: Time between poll attempts (default 30s).
+        """
+        self.log_group_name = log_group_name
+        self.region = region
+        self.max_wait_seconds = max_wait_seconds
+        self.poll_interval_seconds = poll_interval_seconds
+        self._helper = CloudWatchSpanHelper(region=region)
+
+    def collect(self, session_id: str, start_time: datetime, end_time: datetime) -> List[dict]:
+        """Collect spans from CloudWatch, polling until spans appear or timeout.
+
+        Args:
+            session_id: The session ID to collect spans for.
+            start_time: The start time of the session invocation.
+            end_time: The end time of the session invocation.
+
+        Returns:
+            List of ADOT span dictionaries.
+        """
+        # Widen the query window so spans ingested shortly after the
+        # invocation ended are not excluded.  CloudWatch Logs Insights
+        # treats endTime as exclusive and ingestion can lag by seconds,
+        # so a 60-second buffer avoids missing spans on every retry.
+        query_end_time = end_time + timedelta(seconds=60)
+        logger.debug(
+            "Collecting spans for session_id=%s, log_group=%s, time_range=[%s, %s]",
+            session_id,
+            self.log_group_name,
+            start_time,
+            query_end_time,
+        )
+        deadline = time.monotonic() + self.max_wait_seconds
+
+        while True:
+            spans = self._fetch_spans(session_id, start_time, query_end_time)
+            logger.debug("fetch_spans returned %d span(s)", len(spans))
+
+            if spans:
+                logger.info("Collected %d span(s) for session %s", len(spans), session_id)
+                return spans
+
+            if time.monotonic() + self.poll_interval_seconds > deadline:
+                logger.warning(
+                    "Span collection timed out after %ds for session %s (0 spans found)",
+                    self.max_wait_seconds,
+                    session_id,
+                )
+                return spans
+
+            logger.info("No spans found yet, retrying in %ds...", self.poll_interval_seconds)
+            time.sleep(self.poll_interval_seconds)
+
+    def _fetch_spans(self, session_id: str, start_time: datetime, end_time: datetime) -> List[dict]:
+        """Fetch spans from both aws/spans and the configured log group.
+
+        Queries both log groups with a precise attributes.session.id filter,
+        combines results, and returns only valid ADOT span documents.
+        """
+        query_string = (
+            f"fields @timestamp, @message"
+            f'\n| filter attributes.session.id = "{session_id}"'
+            f"\n| filter ispresent(scope.name)"
+            f"\n| filter ispresent(traceId)"
+            f"\n| filter ispresent(spanId)"
+            f"\n| sort @timestamp asc"
+        )
+
+        aws_spans = self._helper.query_log_group(
+            AWS_SPANS_LOG_GROUP, session_id, start_time, end_time, query_string=query_string
+        )
+        event_spans = self._helper.query_log_group(
+            self.log_group_name, session_id, start_time, end_time, query_string=query_string
+        )
+
+        all_data = aws_spans + event_spans
+
+        logger.info("Fetched %d span items from CloudWatch", len(all_data))
+        return all_data

src/bedrock_agentcore/evaluation/client.py

@@ -0,0 +1,234 @@
+"""EvaluationClient for collecting spans and running evaluations."""
+
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Optional
+
+import boto3
+from botocore.config import Config
+
+from bedrock_agentcore._utils.user_agent import build_user_agent_suffix
+from bedrock_agentcore.evaluation._agent_span_collector import CloudWatchAgentSpanCollector
+
+logger = logging.getLogger(__name__)
+
+MAX_TARGET_IDS_PER_REQUEST = 10
+QUERY_TIMEOUT_SECONDS = 60
+POLL_INTERVAL_SECONDS = 2
+
+
+class EvaluationClient:
+    """Client for evaluating agent sessions.
+
+    Collects spans from CloudWatch and calls the evaluation API with
+    level-aware batching.
+
+    Example::
+
+        client = EvaluationClient(region_name="us-west-2")
+
+        # Using agent_id (log group derived automatically)
+        results = client.run(
+            evaluator_ids=["accuracy", "toxicity"],
+            session_id="sess-123",
+            agent_id="my-agent",
+        )
+
+        # Using log_group_name directly
+        results = client.run(
+            evaluator_ids=["accuracy", "toxicity"],
+            session_id="sess-123",
+            log_group_name="/custom/my-log-group",
+        )
+
+        for r in results:
+            print(f"{r['evaluatorId']}: {r.get('value')} - {r.get('explanation')}")
+    """
+
+    def __init__(
+        self,
+        region_name: Optional[str] = None,
+        integration_source: Optional[str] = None,
+    ):
+        """Initialize the EvaluationClient.
+
+        Args:
+            region_name: AWS region. Falls back to boto3 session region or us-west-2.
+            integration_source: Optional integration framework identifier for telemetry.
+        """
+        self.region_name = region_name or boto3.Session().region_name or "us-west-2"
+        self.integration_source = integration_source
+
+        user_agent_extra = build_user_agent_suffix(integration_source)
+        client_config = Config(user_agent_extra=user_agent_extra)
+
+        self._dp_client = boto3.client(
+            "bedrock-agentcore",
+            region_name=self.region_name,
+            config=client_config,
+        )
+        self._cp_client = boto3.client(
+            "bedrock-agentcore-control",
+            region_name=self.region_name,
+            config=client_config,
+        )
+        self._evaluator_level_cache: Dict[str, str] = {}
+
+        logger.info("Initialized EvaluationClient in region %s", self.region_name)
+
+    def run(
+        self,
+        evaluator_ids: List[str],
+        session_id: str,
+        agent_id: Optional[str] = None,
+        look_back_time: timedelta = timedelta(days=7),
+        log_group_name: Optional[str] = None,
+    ) -> List[Dict[str, Any]]:
+        """Evaluate an agent session end-to-end.
+
+        1. Collects spans from CloudWatch.
+        2. For each evaluator, looks up its level (SESSION/TRACE/TOOL_CALL).
+        3. Builds the appropriate evaluationTarget based on level.
+        4. Calls evaluate() with auto-batching (max 10 target IDs per request).
+        5. Returns combined evaluationResults from all evaluators.
+
+        Either ``agent_id`` or ``log_group_name`` must be provided.
+        When only ``agent_id`` is given, the log group name is derived as
+        ``/aws/bedrock-agentcore/runtimes/{agent_id}-DEFAULT``.
+
+        Args:
+            evaluator_ids: List of evaluator IDs (built-in or custom ARNs).
+            session_id: The session ID to evaluate.
+            agent_id: The agent ID. Used to derive the log group when
+                ``log_group_name`` is not provided.
+            look_back_time: How far back to search for spans (default: 7 days).
+            log_group_name: CloudWatch log group name. If provided, ``agent_id``
+                is not required.
+
+        Returns:
+            List of evaluation result dicts from all evaluators.
+
+        Raises:
+            ValueError: If neither ``agent_id`` nor ``log_group_name`` is provided.
+        """
+        if not agent_id and not log_group_name:
+            raise ValueError("Provide either agent_id or log_group_name.")
+
+        if not log_group_name:
+            log_group_name = f"/aws/bedrock-agentcore/runtimes/{agent_id}-DEFAULT"
+            logger.debug("Derived log_group_name=%s from agent_id=%s", log_group_name, agent_id)
+
+        end_time = datetime.now(timezone.utc)
+        start_time = end_time - look_back_time
+
+        logger.info(
+            "Running evaluation for session=%s, log_group=%s, time_range=[%s, %s]",
+            session_id,
+            log_group_name,
+            start_time,
+            end_time,
+        )
+
+        # Step 1: Collect spans
+        collector = CloudWatchAgentSpanCollector(
+            log_group_name=log_group_name,
+            region=self.region_name,
+            max_wait_seconds=QUERY_TIMEOUT_SECONDS,
+            poll_interval_seconds=POLL_INTERVAL_SECONDS,
+        )
+        spans = collector.collect(
+            session_id=session_id,
+            start_time=start_time,
+            end_time=end_time,
+        )
+
+        if not spans:
+            logger.warning("No spans found for session %s", session_id)
+            return []
+
+        base_input = {"evaluationInput": {"sessionSpans": spans}}
+
+        # Steps 2-4: For each evaluator, look up level, build targets, call API
+        all_results = []
+        for evaluator_id in evaluator_ids:
+            level = self._get_evaluator_level(evaluator_id)
+            logger.info("Evaluating with %s (level=%s)", evaluator_id, level)
+            requests = self._build_requests_for_level(evaluator_id, level, base_input, spans)
+            if len(requests) > 1:
+                logger.debug("Split into %d batched request(s) for evaluator %s", len(requests), evaluator_id)
+            for request in requests:
+                response = self._dp_client.evaluate(evaluatorId=evaluator_id, **request)
+                all_results.extend(response.get("evaluationResults", []))
+
+        logger.info(
+            "Evaluation complete: %d result(s) from %d evaluator(s)",
+            len(all_results),
+            len(evaluator_ids),
+        )
+        return all_results
+
+    def _get_evaluator_level(self, evaluator_id: str) -> str:
+        """Look up evaluator level with caching. Falls back to SESSION."""
+        if evaluator_id not in self._evaluator_level_cache:
+            try:
+                response = self._cp_client.get_evaluator(evaluatorId=evaluator_id)
+                self._evaluator_level_cache[evaluator_id] = response["level"]
+            except Exception as e:
+                logger.warning(
+                    "Failed to get level for %s, defaulting to SESSION: %s",
+                    evaluator_id,
+                    e,
+                )
+                self._evaluator_level_cache[evaluator_id] = "SESSION"
+        return self._evaluator_level_cache[evaluator_id]
+
+    def _build_requests_for_level(
+        self,
+        evaluator_id: str,
+        level: str,
+        base_input: dict,
+        spans: list,
+    ) -> List[dict]:
+        """Build one or more evaluate request payloads based on evaluator level."""
+        if level == "SESSION":
+            return [base_input]
+
+        if level == "TRACE":
+            trace_ids = self._extract_trace_ids(spans)
+            logger.debug("Extracted %d unique trace ID(s) for evaluator %s", len(trace_ids), evaluator_id)
+            if not trace_ids:
+                raise ValueError(f"No trace IDs found for trace-level evaluator {evaluator_id}")
+            return [
+                {**base_input, "evaluationTarget": {"traceIds": trace_ids[i : i + MAX_TARGET_IDS_PER_REQUEST]}}
+                for i in range(0, len(trace_ids), MAX_TARGET_IDS_PER_REQUEST)
+            ]
+
+        if level == "TOOL_CALL":
+            tool_span_ids = self._extract_tool_span_ids(spans)
+            logger.debug("Extracted %d tool span ID(s) for evaluator %s", len(tool_span_ids), evaluator_id)
+            if not tool_span_ids:
+                raise ValueError(f"No tool span IDs found for tool-level evaluator {evaluator_id}")
+            return [
+                {**base_input, "evaluationTarget": {"spanIds": tool_span_ids[i : i + MAX_TARGET_IDS_PER_REQUEST]}}
+                for i in range(0, len(tool_span_ids), MAX_TARGET_IDS_PER_REQUEST)
+            ]
+
+        raise ValueError(f"Unknown evaluator level: {level}")
+
+    @staticmethod
+    def _extract_trace_ids(spans: list) -> List[str]:
+        """Extract unique trace IDs from spans, ordered by appearance."""
+        return list(dict.fromkeys(span.get("traceId") for span in spans if span.get("traceId")))
+
+    @staticmethod
+    def _extract_tool_span_ids(spans: list) -> List[str]:
+        """Extract span IDs for tool execution spans."""
+        tool_span_ids: List[str] = []
+        for span in spans:
+            name = span.get("name", "")
+            kind = span.get("kind")
+            if kind == "SPAN_KIND_INTERNAL" and name.startswith("Tool:"):
+                span_id = span.get("spanId")
+                if span_id:
+                    tool_span_ids.append(span_id)
+        return tool_span_ids