fix: resolve mypy redundant-cast errors while keeping pyright happy

Use cast(list[Any], x) with type: ignore[redundant-cast] comments to
satisfy both mypy (which considers casting Any redundant) and pyright
strict mode (which needs explicit casts to narrow Unknown types).

Also fix evaluator decorator check_name type annotation to be
explicitly str, resolving mypy str|Any|None mismatch.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

python/packages/azure-ai/agent_framework_azure_ai/__init__.py

@@ -11,6 +11,11 @@ from ._embedding_client import (
     AzureAIInferenceEmbeddingSettings,
     RawAzureAIInferenceEmbeddingClient,
 )
+from ._foundry_evals import (
+    FoundryEvals,
+    evaluate_foundry_target,
+    evaluate_traces,
+)
 from ._foundry_memory_provider import FoundryMemoryProvider
 from ._project_provider import AzureAIProjectAgentProvider
 from ._shared import AzureAISettings
@@ -31,8 +36,11 @@ __all__ = [
     "AzureAIProjectAgentOptions",
     "AzureAIProjectAgentProvider",
     "AzureAISettings",
+    "FoundryEvals",
     "FoundryMemoryProvider",
     "RawAzureAIClient",
     "RawAzureAIInferenceEmbeddingClient",
     "__version__",
+    "evaluate_foundry_target",
+    "evaluate_traces",
 ]

python/packages/azure-ai/agent_framework_azure_ai/_foundry_evals.py

@@ -0,0 +1,838 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Microsoft Foundry Evals integration for Microsoft Agent Framework.
+
+Provides ``FoundryEvals``, an ``Evaluator`` implementation backed by Azure AI
+Foundry's built-in evaluators. See docs/decisions/0018-foundry-evals-integration.md
+for the design rationale.
+
+Typical usage::
+
+    from agent_framework import evaluate_agent
+    from agent_framework_azure_ai import FoundryEvals
+
+    evals = FoundryEvals(project_client=project_client, model_deployment="gpt-4o")
+    results = await evaluate_agent(
+        agent=my_agent,
+        queries=["What's the weather in Seattle?"],
+        evaluators=evals,
+    )
+    assert results.all_passed
+    print(results.report_url)
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import TYPE_CHECKING, Any, Sequence, cast
+
+from agent_framework._evaluation import (
+    ConversationSplit,
+    ConversationSplitter,
+    EvalItem,
+    EvalItemResult,
+    EvalResults,
+    EvalScoreResult,
+)
+
+if TYPE_CHECKING:
+    from azure.ai.projects.aio import AIProjectClient
+    from openai import AsyncOpenAI
+
+logger = logging.getLogger(__name__)
+
+# Agent evaluators that accept query/response as conversation arrays.
+# Maintained manually — check https://learn.microsoft.com/en-us/azure/ai-studio/how-to/develop/evaluate-sdk
+# for the latest evaluator list. These are the evaluators that need conversation-format input.
+_AGENT_EVALUATORS: set[str] = {
+    "builtin.intent_resolution",
+    "builtin.task_adherence",
+    "builtin.task_completion",
+    "builtin.task_navigation_efficiency",
+    "builtin.tool_call_accuracy",
+    "builtin.tool_selection",
+    "builtin.tool_input_accuracy",
+    "builtin.tool_output_utilization",
+    "builtin.tool_call_success",
+}
+
+# Evaluators that additionally require tool_definitions.
+_TOOL_EVALUATORS: set[str] = {
+    "builtin.tool_call_accuracy",
+    "builtin.tool_selection",
+    "builtin.tool_input_accuracy",
+    "builtin.tool_output_utilization",
+    "builtin.tool_call_success",
+}
+
+_BUILTIN_EVALUATORS: dict[str, str] = {
+    # Agent behavior
+    "intent_resolution": "builtin.intent_resolution",
+    "task_adherence": "builtin.task_adherence",
+    "task_completion": "builtin.task_completion",
+    "task_navigation_efficiency": "builtin.task_navigation_efficiency",
+    # Tool usage
+    "tool_call_accuracy": "builtin.tool_call_accuracy",
+    "tool_selection": "builtin.tool_selection",
+    "tool_input_accuracy": "builtin.tool_input_accuracy",
+    "tool_output_utilization": "builtin.tool_output_utilization",
+    "tool_call_success": "builtin.tool_call_success",
+    # Quality
+    "coherence": "builtin.coherence",
+    "fluency": "builtin.fluency",
+    "relevance": "builtin.relevance",
+    "groundedness": "builtin.groundedness",
+    "response_completeness": "builtin.response_completeness",
+    "similarity": "builtin.similarity",
+    # Safety
+    "violence": "builtin.violence",
+    "sexual": "builtin.sexual",
+    "self_harm": "builtin.self_harm",
+    "hate_unfairness": "builtin.hate_unfairness",
+}
+
+# Default evaluator sets used when evaluators=None
+_DEFAULT_EVALUATORS: list[str] = [
+    "relevance",
+    "coherence",
+    "task_adherence",
+]
+
+_DEFAULT_TOOL_EVALUATORS: list[str] = [
+    "tool_call_accuracy",
+]
+
+
+def _resolve_evaluator(name: str) -> str:
+    """Resolve a short evaluator name to its fully-qualified ``builtin.*`` form.
+
+    Args:
+        name: Short name (e.g. ``"relevance"``) or fully-qualified name
+            (e.g. ``"builtin.relevance"``).
+
+    Returns:
+        The fully-qualified evaluator name.
+
+    Raises:
+        ValueError: If the name is not recognized.
+    """
+    if name.startswith("builtin."):
+        return name
+    resolved = _BUILTIN_EVALUATORS.get(name)
+    if resolved is None:
+        raise ValueError(f"Unknown evaluator '{name}'. Available: {sorted(_BUILTIN_EVALUATORS)}")
+    return resolved
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+
+def _build_testing_criteria(
+    evaluators: Sequence[str],
+    model_deployment: str,
+    *,
+    include_data_mapping: bool = False,
+) -> list[dict[str, Any]]:
+    """Build ``testing_criteria`` for ``evals.create()``.
+
+    Args:
+        evaluators: Evaluator names.
+        model_deployment: Model deployment for the LLM judge.
+        include_data_mapping: Whether to include field-level data mapping
+            (required for the JSONL data source, not needed for response-based).
+    """
+    criteria: list[dict[str, Any]] = []
+    for name in evaluators:
+        qualified = _resolve_evaluator(name)
+        short = name if not name.startswith("builtin.") else name.split(".")[-1]
+
+        entry: dict[str, Any] = {
+            "type": "azure_ai_evaluator",
+            "name": short,
+            "evaluator_name": qualified,
+            "initialization_parameters": {"deployment_name": model_deployment},
+        }
+
+        if include_data_mapping:
+            if qualified in _AGENT_EVALUATORS:
+                # Agent evaluators: query/response as conversation arrays
+                mapping: dict[str, str] = {
+                    "query": "{{item.query_messages}}",
+                    "response": "{{item.response_messages}}",
+                }
+            else:
+                # Quality evaluators: query/response as strings
+                mapping = {
+                    "query": "{{item.query}}",
+                    "response": "{{item.response}}",
+                }
+            if qualified == "builtin.groundedness":
+                mapping["context"] = "{{item.context}}"
+            if qualified in _TOOL_EVALUATORS:
+                mapping["tool_definitions"] = "{{item.tool_definitions}}"
+            entry["data_mapping"] = mapping
+
+        criteria.append(entry)
+    return criteria
+
+
+def _build_item_schema(*, has_context: bool = False, has_tools: bool = False) -> dict[str, Any]:
+    """Build the ``item_schema`` for custom JSONL eval definitions."""
+    properties: dict[str, Any] = {
+        "query": {"type": "string"},
+        "response": {"type": "string"},
+        "query_messages": {"type": "array"},
+        "response_messages": {"type": "array"},
+    }
+    if has_context:
+        properties["context"] = {"type": "string"}
+    if has_tools:
+        properties["tool_definitions"] = {"type": "array"}
+    return {
+        "type": "object",
+        "properties": properties,
+        "required": ["query", "response"],
+    }
+
+
+def _resolve_default_evaluators(
+    evaluators: Sequence[str] | None,
+    items: Sequence[EvalItem | dict[str, Any]] | None = None,
+) -> list[str]:
+    """Resolve evaluators, applying defaults when ``None``.
+
+    Defaults to relevance + coherence + task_adherence. Automatically adds
+    tool_call_accuracy when items contain tools.
+    """
+    if evaluators is not None:
+        return list(evaluators)
+
+    result = list(_DEFAULT_EVALUATORS)
+    if items is not None:
+        has_tools = any((item.tools if isinstance(item, EvalItem) else item.get("tool_definitions")) for item in items)
+        if has_tools:
+            result.extend(_DEFAULT_TOOL_EVALUATORS)
+    return result
+
+
+def _filter_tool_evaluators(
+    evaluators: list[str],
+    items: Sequence[EvalItem | dict[str, Any]],
+) -> list[str]:
+    """Remove tool evaluators if no items have tool definitions."""
+    has_tools = any((item.tools if isinstance(item, EvalItem) else item.get("tool_definitions")) for item in items)
+    if has_tools:
+        return evaluators
+    filtered = [e for e in evaluators if _resolve_evaluator(e) not in _TOOL_EVALUATORS]
+    return filtered if filtered else list(_DEFAULT_EVALUATORS)
+
+
+async def _ensure_async_result(func: Any, *args: Any, **kwargs: Any) -> Any:
+    """Invoke a sync or async client method transparently.
+
+    If ``func`` returns a coroutine (async client), awaits it directly.
+    Otherwise returns the already-resolved result.
+    """
+    import inspect
+
+    result = func(*args, **kwargs)
+    if inspect.isawaitable(result):
+        return await result
+    return result
+
+
+async def _poll_eval_run(
+    client: AsyncOpenAI,
+    eval_id: str,
+    run_id: str,
+    poll_interval: float = 5.0,
+    timeout: float = 600.0,
+    provider: str = "Microsoft Foundry",
+    *,
+    fetch_output_items: bool = True,
+) -> EvalResults:
+    """Poll an eval run until completion or timeout."""
+    loop = asyncio.get_event_loop()
+    deadline = loop.time() + timeout
+    while True:
+        run = await _ensure_async_result(client.evals.runs.retrieve, run_id=run_id, eval_id=eval_id)
+        if run.status in ("completed", "failed", "canceled"):
+            error_msg = None
+            if run.status == "failed":
+                error_msg = (
+                    getattr(run, "error", None)
+                    or getattr(run, "error_message", None)
+                    or getattr(run, "failure_reason", None)
+                )
+                if error_msg and not isinstance(error_msg, str):
+                    error_msg = str(error_msg)
+
+            items: list[EvalItemResult] = []
+            if fetch_output_items and run.status == "completed":
+                items = await _fetch_output_items(client, eval_id, run_id)
+
+            return EvalResults(
+                provider=provider,
+                eval_id=eval_id,
+                run_id=run_id,
+                status=run.status,
+                result_counts=_extract_result_counts(run),
+                report_url=getattr(run, "report_url", None),
+                error=error_msg,
+                per_evaluator=_extract_per_evaluator(run),
+                items=items,
+            )
+        remaining = deadline - loop.time()
+        if remaining <= 0:
+            return EvalResults(provider=provider, eval_id=eval_id, run_id=run_id, status="timeout")
+        logger.debug("Eval run %s status: %s (%.0fs remaining)", run_id, run.status, remaining)
+        await asyncio.sleep(min(poll_interval, remaining))
+
+
+def _extract_result_counts(run: Any) -> dict[str, int] | None:
+    """Safely extract result_counts from an eval run object."""
+    counts = getattr(run, "result_counts", None)
+    if counts is None:
+        return None
+    if isinstance(counts, dict):
+        return cast(dict[str, int], counts)
+    try:
+        attrs = cast(dict[str, Any], vars(counts))
+        return {str(k): v for k, v in attrs.items() if isinstance(v, int)}
+    except TypeError:
+        return None
+
+
+def _extract_per_evaluator(run: Any) -> dict[str, dict[str, int]]:
+    """Safely extract per-evaluator result breakdowns from an eval run."""
+    per_eval: dict[str, dict[str, int]] = {}
+    per_testing_criteria = getattr(run, "per_testing_criteria_results", None)
+    if per_testing_criteria is None:
+        return per_eval
+    try:
+        items = cast(list[Any], per_testing_criteria) if isinstance(per_testing_criteria, list) else []  # type: ignore[redundant-cast]
+        for item in items:
+            name: str = str(getattr(item, "name", None) or getattr(item, "testing_criteria", "unknown"))
+            counts = _extract_result_counts(item)
+            if name and counts:
+                per_eval[name] = counts
+    except (TypeError, AttributeError):
+        pass
+    return per_eval
+
+
+async def _fetch_output_items(
+    client: AsyncOpenAI,
+    eval_id: str,
+    run_id: str,
+) -> list[EvalItemResult]:
+    """Fetch per-item results from the output_items API.
+
+    Converts the provider-specific ``OutputItemListResponse`` objects into
+    provider-agnostic ``EvalItemResult`` instances with per-evaluator scores,
+    error categorization, and token usage.
+    """
+    items: list[EvalItemResult] = []
+    try:
+        output_items_page = await _ensure_async_result(
+            client.evals.runs.output_items.list,
+            run_id=run_id,
+            eval_id=eval_id,
+        )
+
+        for oi in output_items_page:
+            item_id = getattr(oi, "id", "") or ""
+            status = getattr(oi, "status", "unknown") or "unknown"
+
+            # Extract per-evaluator scores
+            scores: list[EvalScoreResult] = []
+            for r in getattr(oi, "results", []) or []:
+                scores.append(
+                    EvalScoreResult(
+                        name=getattr(r, "name", "unknown"),
+                        score=getattr(r, "score", 0.0),
+                        passed=getattr(r, "passed", None),
+                        sample=getattr(r, "sample", None),
+                    )
+                )
+
+            # Extract error info from sample
+            error_code: str | None = None
+            error_message: str | None = None
+            token_usage: dict[str, int] | None = None
+            input_text: str | None = None
+            output_text: str | None = None
+            response_id: str | None = None
+
+            sample = getattr(oi, "sample", None)
+            if sample is not None:
+                error = getattr(sample, "error", None)
+                if error is not None:
+                    code = getattr(error, "code", None)
+                    msg = getattr(error, "message", None)
+                    if code or msg:
+                        error_code = code or None
+                        error_message = msg or None
+
+                usage = getattr(sample, "usage", None)
+                if usage is not None:
+                    total = getattr(usage, "total_tokens", 0)
+                    if total:
+                        token_usage = {
+                            "prompt_tokens": getattr(usage, "prompt_tokens", 0),
+                            "completion_tokens": getattr(usage, "completion_tokens", 0),
+                            "total_tokens": total,
+                            "cached_tokens": getattr(usage, "cached_tokens", 0),
+                        }
+
+                # Extract input/output text
+                sample_input = getattr(sample, "input", None)
+                if sample_input:
+                    parts = [getattr(si, "content", "") for si in sample_input if getattr(si, "role", "") == "user"]
+                    if parts:
+                        input_text = " ".join(parts)
+
+                sample_output = getattr(sample, "output", None)
+                if sample_output:
+                    parts = [
+                        getattr(so, "content", "") or ""
+                        for so in sample_output
+                        if getattr(so, "role", "") == "assistant"
+                    ]
+                    if parts:
+                        output_text = " ".join(parts)
+
+            # Extract response_id from datasource_item
+            ds_item = getattr(oi, "datasource_item", None)
+            if ds_item and isinstance(ds_item, dict):
+                ds_dict = cast(dict[str, Any], ds_item)
+                resp_id_val = ds_dict.get("resp_id") or ds_dict.get("response_id")
+                response_id = str(resp_id_val) if resp_id_val else None
+
+            items.append(
+                EvalItemResult(
+                    item_id=item_id,
+                    status=status,
+                    scores=scores,
+                    error_code=error_code,
+                    error_message=error_message,
+                    response_id=response_id,
+                    input_text=input_text,
+                    output_text=output_text,
+                    token_usage=token_usage,
+                )
+            )
+    except Exception:
+        logger.debug("Could not fetch output_items for run %s", run_id, exc_info=True)
+
+    return items
+
+
+def _resolve_openai_client(
+    openai_client: AsyncOpenAI | None = None,
+    project_client: AIProjectClient | None = None,
+) -> AsyncOpenAI:
+    """Resolve an OpenAI client from explicit client or project_client."""
+    if openai_client is not None:
+        return openai_client
+    if project_client is not None:
+        return project_client.get_openai_client()
+    raise ValueError("Provide either 'openai_client' or 'project_client'.")
+
+
+# ---------------------------------------------------------------------------
+# FoundryEvals — Evaluator implementation for Microsoft Foundry
+# ---------------------------------------------------------------------------
+
+
+class FoundryEvals:
+    """Evaluation provider backed by Microsoft Foundry.
+
+    Implements the ``Evaluator`` protocol so it can be passed to the
+    provider-agnostic ``evaluate_agent()`` and
+    ``evaluate_workflow()`` functions from ``agent_framework``.
+
+    Also provides constants for built-in evaluator names for IDE
+    autocomplete and typo prevention::
+
+        from agent_framework_azure_ai import FoundryEvals
+
+        evaluators = [FoundryEvals.RELEVANCE, FoundryEvals.TOOL_CALL_ACCURACY]
+
+    The simplest usage::
+
+        from agent_framework import evaluate_agent
+        from agent_framework_azure_ai import FoundryEvals
+
+        evals = FoundryEvals(project_client=client, model_deployment="gpt-4o")
+        results = await evaluate_agent(agent=agent, queries=queries, evaluators=evals)
+
+    **Evaluator selection:**
+
+    By default, runs ``relevance``, ``coherence``, and ``task_adherence``.
+    Automatically adds ``tool_call_accuracy`` when items contain tool
+    definitions. Override with ``evaluators=``.
+
+    **Responses API optimization:**
+
+    When all items have a ``response_id`` and no tool evaluators are needed,
+    uses Foundry's server-side response retrieval path (no data upload).
+
+    Args:
+        project_client: An ``AIProjectClient`` instance (sync or async).
+            Provide this or *openai_client*.
+        openai_client: An ``AsyncOpenAI`` client with evals API.
+        model_deployment: Model deployment name for the evaluator LLM judge.
+        evaluators: Evaluator names (e.g. ``["relevance", "tool_call_accuracy"]``).
+            When ``None`` (default), uses smart defaults based on item data.
+        conversation_split: How to split multi-turn conversations into
+            query/response halves.  Defaults to ``LAST_TURN``.  Pass a
+            ``ConversationSplit`` enum value or a custom callable — see
+            ``ConversationSplitter``.
+        poll_interval: Seconds between status polls (default 5.0).
+        timeout: Maximum seconds to wait for completion (default 600.0).
+    """
+
+    # ---------------------------------------------------------------------------
+    # Built-in evaluator name constants
+    # ---------------------------------------------------------------------------
+
+    # Agent behavior
+    INTENT_RESOLUTION: str = "intent_resolution"
+    TASK_ADHERENCE: str = "task_adherence"
+    TASK_COMPLETION: str = "task_completion"
+    TASK_NAVIGATION_EFFICIENCY: str = "task_navigation_efficiency"
+
+    # Tool usage
+    TOOL_CALL_ACCURACY: str = "tool_call_accuracy"
+    TOOL_SELECTION: str = "tool_selection"
+    TOOL_INPUT_ACCURACY: str = "tool_input_accuracy"
+    TOOL_OUTPUT_UTILIZATION: str = "tool_output_utilization"
+    TOOL_CALL_SUCCESS: str = "tool_call_success"
+
+    # Quality
+    COHERENCE: str = "coherence"
+    FLUENCY: str = "fluency"
+    RELEVANCE: str = "relevance"
+    GROUNDEDNESS: str = "groundedness"
+    RESPONSE_COMPLETENESS: str = "response_completeness"
+    SIMILARITY: str = "similarity"
+
+    # Safety
+    VIOLENCE: str = "violence"
+    SEXUAL: str = "sexual"
+    SELF_HARM: str = "self_harm"
+    HATE_UNFAIRNESS: str = "hate_unfairness"
+
+    def __init__(
+        self,
+        *,
+        project_client: AIProjectClient | None = None,
+        openai_client: AsyncOpenAI | None = None,
+        model_deployment: str,
+        evaluators: Sequence[str] | None = None,
+        conversation_split: ConversationSplitter = ConversationSplit.LAST_TURN,
+        poll_interval: float = 5.0,
+        timeout: float = 600.0,
+    ):
+        self.name = "Microsoft Foundry"
+        self._client = _resolve_openai_client(openai_client, project_client)
+        self._model_deployment = model_deployment
+        self._evaluators = list(evaluators) if evaluators is not None else None
+        self._conversation_split = conversation_split
+        self._poll_interval = poll_interval
+        self._timeout = timeout
+
+    async def evaluate(
+        self,
+        items: Sequence[EvalItem],
+        *,
+        eval_name: str = "Agent Framework Eval",
+    ) -> EvalResults:
+        """Evaluate items using Foundry evaluators.
+
+        Implements the ``Evaluator`` protocol. Automatically selects the
+        optimal data path (Responses API vs JSONL dataset) and filters
+        tool evaluators for items without tool definitions.
+
+        Args:
+            items: Eval data items from ``AgentEvalConverter.to_eval_item()``.
+            eval_name: Display name for the evaluation run.
+
+        Returns:
+            ``EvalResults`` with status, counts, and portal link.
+        """
+        # Resolve evaluators with auto-detection
+        resolved = _resolve_default_evaluators(self._evaluators, items=items)
+        # Filter tool evaluators if items don't have tools
+        resolved = _filter_tool_evaluators(resolved, items)
+
+        # Standard JSONL dataset path
+        return await self._evaluate_via_dataset(items, resolved, eval_name)
+
+    # -- Internal evaluation paths --
+
+    async def _evaluate_via_responses(
+        self,
+        response_ids: Sequence[str],
+        evaluators: list[str],
+        eval_name: str,
+    ) -> EvalResults:
+        """Evaluate using Foundry's Responses API retrieval path."""
+        eval_obj = await _ensure_async_result(
+            self._client.evals.create,
+            name=eval_name,
+            data_source_config={"type": "azure_ai_source", "scenario": "responses"},
+            testing_criteria=_build_testing_criteria(evaluators, self._model_deployment),
+        )
+
+        data_source = {
+            "type": "azure_ai_responses",
+            "item_generation_params": {
+                "type": "response_retrieval",
+                "data_mapping": {"response_id": "{{item.resp_id}}"},
+                "source": {
+                    "type": "file_content",
+                    "content": [{"item": {"resp_id": rid}} for rid in response_ids],
+                },
+            },
+        }
+
+        run = await _ensure_async_result(
+            self._client.evals.runs.create,
+            eval_id=eval_obj.id,
+            name=f"{eval_name} Run",
+            data_source=data_source,
+        )
+
+        return await _poll_eval_run(
+            self._client,
+            eval_obj.id,
+            run.id,
+            self._poll_interval,
+            self._timeout,
+            provider=self.name,
+        )
+
+    async def _evaluate_via_dataset(
+        self,
+        items: Sequence[EvalItem],
+        evaluators: list[str],
+        eval_name: str,
+    ) -> EvalResults:
+        """Evaluate using JSONL dataset upload path."""
+        dicts = [item.to_eval_data(split=item.split_strategy or self._conversation_split) for item in items]
+        has_context = any("context" in d for d in dicts)
+        has_tools = any("tool_definitions" in d for d in dicts)
+
+        eval_obj = await _ensure_async_result(
+            self._client.evals.create,
+            name=eval_name,
+            data_source_config={
+                "type": "custom",
+                "item_schema": _build_item_schema(has_context=has_context, has_tools=has_tools),
+                "include_sample_schema": True,
+            },
+            testing_criteria=_build_testing_criteria(
+                evaluators,
+                self._model_deployment,
+                include_data_mapping=True,
+            ),
+        )
+
+        data_source = {
+            "type": "jsonl",
+            "source": {
+                "type": "file_content",
+                "content": [{"item": d} for d in dicts],
+            },
+        }
+
+        run = await _ensure_async_result(
+            self._client.evals.runs.create,
+            eval_id=eval_obj.id,
+            name=f"{eval_name} Run",
+            data_source=data_source,
+        )
+
+        return await _poll_eval_run(
+            self._client,
+            eval_obj.id,
+            run.id,
+            self._poll_interval,
+            self._timeout,
+            provider=self.name,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Foundry-specific functions (not part of the Evaluator protocol)
+# ---------------------------------------------------------------------------
+
+
+async def evaluate_traces(
+    *,
+    evaluators: Sequence[str] | None = None,
+    openai_client: AsyncOpenAI | None = None,
+    project_client: AIProjectClient | None = None,
+    model_deployment: str,
+    response_ids: Sequence[str] | None = None,
+    trace_ids: Sequence[str] | None = None,
+    agent_id: str | None = None,
+    lookback_hours: int = 24,
+    eval_name: str = "Agent Framework Trace Eval",
+    poll_interval: float = 5.0,
+    timeout: float = 600.0,
+) -> EvalResults:
+    """Evaluate agent behavior from OTel traces or response IDs.
+
+    Foundry-specific function — works with any agent that emits OTel traces
+    to App Insights. Provide *response_ids* for specific responses,
+    *trace_ids* for specific traces, or *agent_id* with *lookback_hours*
+    to evaluate recent activity.
+
+    Args:
+        evaluators: Evaluator names (e.g. ``[FoundryEvals.RELEVANCE]``).
+            Defaults to relevance, coherence, and task_adherence.
+        openai_client: ``AsyncOpenAI`` client. Provide this or *project_client*.
+        project_client: An ``AIProjectClient`` instance.
+        model_deployment: Model deployment name for the evaluator LLM judge.
+        response_ids: Evaluate specific Responses API responses.
+        trace_ids: Evaluate specific OTel trace IDs from App Insights.
+        agent_id: Filter traces by agent ID (used with *lookback_hours*).
+        lookback_hours: Hours of trace history to evaluate (default 24).
+        eval_name: Display name for the evaluation.
+        poll_interval: Seconds between status polls.
+        timeout: Maximum seconds to wait for completion.
+
+    Returns:
+        ``EvalResults`` with status, result counts, and portal link.
+
+    Example::
+
+        results = await evaluate_traces(
+            response_ids=[response.response_id],
+            evaluators=[FoundryEvals.RELEVANCE],
+            project_client=project_client,
+            model_deployment="gpt-4o",
+        )
+    """
+    client = _resolve_openai_client(openai_client, project_client)
+    resolved_evaluators = _resolve_default_evaluators(evaluators)
+
+    if response_ids:
+        foundry = FoundryEvals(
+            openai_client=client,
+            model_deployment=model_deployment,
+            evaluators=resolved_evaluators,
+            poll_interval=poll_interval,
+            timeout=timeout,
+        )
+        return await foundry._evaluate_via_responses(  # pyright: ignore[reportPrivateUsage]
+            response_ids,
+            resolved_evaluators,
+            eval_name,
+        )
+
+    if not trace_ids and not agent_id:
+        raise ValueError("Provide at least one of: response_ids, trace_ids, or agent_id")
+
+    trace_source: dict[str, Any] = {
+        "type": "azure_ai_traces",
+        "lookback_hours": lookback_hours,
+    }
+    if trace_ids:
+        trace_source["trace_ids"] = list(trace_ids)
+    if agent_id:
+        trace_source["agent_id"] = agent_id
+
+    eval_obj = await _ensure_async_result(
+        client.evals.create,
+        name=eval_name,
+        data_source_config={"type": "azure_ai_source", "scenario": "traces"},
+        testing_criteria=_build_testing_criteria(resolved_evaluators, model_deployment),
+    )
+
+    run = await _ensure_async_result(
+        client.evals.runs.create,
+        eval_id=eval_obj.id,
+        name=f"{eval_name} Run",
+        data_source=trace_source,
+    )
+
+    return await _poll_eval_run(client, eval_obj.id, run.id, poll_interval, timeout)
+
+
+async def evaluate_foundry_target(
+    *,
+    target: dict[str, Any],
+    test_queries: Sequence[str],
+    evaluators: Sequence[str] | None = None,
+    openai_client: AsyncOpenAI | None = None,
+    project_client: AIProjectClient | None = None,
+    model_deployment: str,
+    eval_name: str = "Agent Framework Target Eval",
+    poll_interval: float = 5.0,
+    timeout: float = 600.0,
+) -> EvalResults:
+    """Evaluate a Foundry-registered agent or model deployment.
+
+    Foundry invokes the target, captures the output, and evaluates it. Use
+    this for scheduled evals, red teaming, and CI/CD quality gates.
+
+    Args:
+        target: Target configuration dict.
+        test_queries: Queries for Foundry to send to the target.
+        evaluators: Evaluator names.
+        openai_client: ``AsyncOpenAI`` client. Provide this or *project_client*.
+        project_client: An ``AIProjectClient`` instance.
+        model_deployment: Model deployment name for the evaluator LLM judge.
+        eval_name: Display name for the evaluation.
+        poll_interval: Seconds between status polls.
+        timeout: Maximum seconds to wait for completion.
+
+    Returns:
+        ``EvalResults`` with status, result counts, and portal link.
+
+    Example::
+
+        results = await evaluate_foundry_target(
+            target={"type": "azure_ai_agent", "name": "my-agent"},
+            test_queries=["Book a flight to Paris"],
+            project_client=project_client,
+            model_deployment="gpt-4o",
+        )
+    """
+    client = _resolve_openai_client(openai_client, project_client)
+    resolved_evaluators = _resolve_default_evaluators(evaluators)
+
+    eval_obj = await _ensure_async_result(
+        client.evals.create,
+        name=eval_name,
+        data_source_config={
+            "type": "azure_ai_source",
+            "scenario": "target_completions",
+        },
+        testing_criteria=_build_testing_criteria(resolved_evaluators, model_deployment),
+    )
+
+    data_source: dict[str, Any] = {
+        "type": "azure_ai_target_completions",
+        "target": target,
+        "source": {
+            "type": "file_content",
+            "content": [{"item": {"query": q}} for q in test_queries],
+        },
+    }
+
+    run = await _ensure_async_result(
+        client.evals.runs.create,
+        eval_id=eval_obj.id,
+        name=f"{eval_name} Run",
+        data_source=data_source,
+    )
+
+    return await _poll_eval_run(client, eval_obj.id, run.id, poll_interval, timeout)

python/packages/core/agent_framework/__init__.py

@@ -57,6 +57,27 @@ from ._compaction import (
     included_messages,
     included_token_count,
 )
+from ._evaluation import (
+    AgentEvalConverter,
+    CheckResult,
+    ConversationSplit,
+    ConversationSplitter,
+    EvalItem,
+    EvalItemResult,
+    EvalResults,
+    EvalScoreResult,
+    Evaluator,
+    ExpectedToolCall,
+    LocalEvaluator,
+    evaluate_agent,
+    evaluate_response,
+    evaluate_workflow,
+    evaluator,
+    keyword_check,
+    tool_call_args_match,
+    tool_called_check,
+    tool_calls_present,
+)
 from ._mcp import MCPStdioTool, MCPStreamableHTTPTool, MCPWebsocketTool
 from ._middleware import (
     AgentContext,
@@ -242,6 +263,7 @@ __all__ = [
     "USER_AGENT_TELEMETRY_DISABLED_ENV_VAR",
     "Agent",
     "AgentContext",
+    "AgentEvalConverter",
     "AgentExecutor",
     "AgentExecutorRequest",
     "AgentExecutorResponse",
@@ -268,11 +290,14 @@ __all__ = [
     "ChatOptions",
     "ChatResponse",
     "ChatResponseUpdate",
+    "CheckResult",
     "CheckpointStorage",
     "CompactionProvider",
     "CompactionStrategy",
     "Content",
     "ContinuationToken",
+    "ConversationSplit",
+    "ConversationSplitter",
     "Default",
     "Edge",
     "EdgeCondition",
@@ -281,7 +306,13 @@ __all__ = [
     "EmbeddingGenerationOptions",
     "EmbeddingInputT",
     "EmbeddingT",
+    "EvalItem",
+    "EvalItemResult",
+    "EvalResults",
+    "EvalScoreResult",
+    "Evaluator",
     "Executor",
+    "ExpectedToolCall",
     "FanInEdgeGroup",
     "FanOutEdgeGroup",
     "FileCheckpointStorage",
@@ -300,6 +331,7 @@ __all__ = [
     "InMemoryCheckpointStorage",
     "InMemoryHistoryProvider",
     "InProcRunnerContext",
+    "LocalEvaluator",
     "MCPStdioTool",
     "MCPStreamableHTTPTool",
     "MCPWebsocketTool",
@@ -379,11 +411,16 @@ __all__ = [
     "chat_middleware",
     "create_edge_runner",
     "detect_media_type_from_base64",
+    "evaluate_agent",
+    "evaluate_response",
+    "evaluate_workflow",
+    "evaluator",
     "executor",
     "function_middleware",
     "handler",
     "included_messages",
     "included_token_count",
+    "keyword_check",
     "load_settings",
     "map_chat_to_agent_update",
     "merge_chat_options",
@@ -396,6 +433,9 @@ __all__ = [
     "resolve_agent_id",
     "response_handler",
     "tool",
+    "tool_call_args_match",
+    "tool_called_check",
+    "tool_calls_present",
     "validate_chat_options",
     "validate_tool_mode",
     "validate_tools",

python/packages/core/agent_framework/_agents.py

@@ -639,7 +639,7 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]):  # type: ignore[misc]
                 client=client,
                 name="reasoning-agent",
                 instructions="You are a reasoning assistant.",
-                options={
+                default_options={
                     "temperature": 0.7,
                     "max_tokens": 500,
                     "reasoning_effort": "high",  # OpenAI-specific, IDE will autocomplete!
@@ -697,6 +697,12 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]):  # type: ignore[misc]
                 If both this and a tokenizer on the underlying client are set, this one is used.
             kwargs: Any additional keyword arguments. Will be stored as ``additional_properties``.
         """
+        # Accept 'options' as an alias for 'default_options' so that
+        # Agent(options={"store": False}) works as expected instead of
+        # silently dropping the options into additional_properties.
+        if "options" in kwargs and default_options is None:
+            default_options = kwargs.pop("options")
+
         opts = dict(default_options) if default_options else {}
 
         if not isinstance(client, FunctionInvocationLayer) and isinstance(client, BaseChatClient):

python/packages/core/agent_framework/_workflows/_agent_executor.py

@@ -287,9 +287,12 @@ class AgentExecutor(Executor):
             self._pending_responses_to_agent = pending_responses_payload
 
     def reset(self) -> None:
-        """Reset the internal cache of the executor."""
-        logger.debug("AgentExecutor %s: Resetting cache", self.id)
+        """Reset the internal cache and service session state of the executor for a new run."""
+        logger.debug("AgentExecutor %s: Resetting cache and service session", self.id)
         self._cache.clear()
+        # Clear service_session_id to prevent stale previous_response_id
+        # from leaking between workflow runs (e.g. in evaluate_workflow loops).
+        self._session.service_session_id = None
 
     async def _run_agent_and_emit(
         self,

python/packages/core/agent_framework/_workflows/_workflow.py

@@ -345,6 +345,10 @@ class Workflow(DictConvertible):
                     self._runner.reset_iteration_count()
                     self._runner.context.reset_for_new_run()
                     self._state.clear()
+                    # Reset all executors (clears cached messages, sessions, etc.)
+                    for executor in self.executors.values():
+                        if hasattr(executor, "reset"):
+                            executor.reset()
 
                 # Store run kwargs in State so executors can access them.
                 # Only overwrite when new kwargs are explicitly provided or state was

python/packages/core/tests/core/test_local_eval.py

@@ -0,0 +1,749 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for evaluator checks and LocalEvaluator."""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+
+from agent_framework._evaluation import (
+    CheckResult,
+    EvalItem,
+    ExpectedToolCall,
+    LocalEvaluator,
+    evaluator,
+    keyword_check,
+    tool_call_args_match,
+    tool_calls_present,
+)
+from agent_framework._types import Content, Message
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_item(
+    query: str = "What's the weather in Paris?",
+    response: str = "It's sunny and 75°F",
+    expected_output: str | None = None,
+    conversation: list | None = None,
+    tools: list | None = None,
+    context: str | None = None,
+) -> EvalItem:
+    if conversation is None:
+        conversation = [Message("user", [query]), Message("assistant", [response])]
+    return EvalItem(
+        conversation=conversation,
+        expected_output=expected_output,
+        tools=tools,
+        context=context,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Tier 1: (query, response) -> result
+# ---------------------------------------------------------------------------
+
+
+class TestTier1SimpleChecks:
+    @pytest.mark.asyncio
+    async def test_bool_return_true(self):
+        @evaluator
+        def has_temperature(query: str, response: str) -> bool:
+            return "°F" in response
+
+        result = await has_temperature(_make_item())
+        assert result.passed is True
+        assert result.check_name == "has_temperature"
+
+    @pytest.mark.asyncio
+    async def test_bool_return_false(self):
+        @evaluator
+        def has_celsius(query: str, response: str) -> bool:
+            return "°C" in response
+
+        result = await has_celsius(_make_item())
+        assert result.passed is False
+
+    @pytest.mark.asyncio
+    async def test_float_return_passing(self):
+        @evaluator
+        def length_score(response: str) -> float:
+            return min(len(response) / 10, 1.0)
+
+        result = await length_score(_make_item())
+        assert result.passed is True
+        assert "score=" in result.reason
+
+    @pytest.mark.asyncio
+    async def test_float_return_failing(self):
+        @evaluator
+        def always_low(response: str) -> float:
+            return 0.1
+
+        result = await always_low(_make_item())
+        assert result.passed is False
+
+    @pytest.mark.asyncio
+    async def test_response_only(self):
+        """Function with only 'response' param should work."""
+
+        @evaluator
+        def is_short(response: str) -> bool:
+            return len(response) < 1000
+
+        result = await is_short(_make_item())
+        assert result.passed is True
+
+    @pytest.mark.asyncio
+    async def test_query_only(self):
+        """Function with only 'query' param should work."""
+
+        @evaluator
+        def is_question(query: str) -> bool:
+            return "?" in query
+
+        result = await is_question(_make_item())
+        assert result.passed is True
+
+
+# ---------------------------------------------------------------------------
+# Tier 2: (query, response, expected_output) -> result
+# ---------------------------------------------------------------------------
+
+
+class TestTier2GroundTruth:
+    @pytest.mark.asyncio
+    async def test_exact_match(self):
+        @evaluator
+        def exact_match(response: str, expected_output: str) -> bool:
+            return response.strip() == expected_output.strip()
+
+        item = _make_item(response="42", expected_output="42")
+        assert (await exact_match(item)).passed is True
+
+        item2 = _make_item(response="43", expected_output="42")
+        assert (await exact_match(item2)).passed is False
+
+    @pytest.mark.asyncio
+    async def test_expected_output_defaults_to_empty(self):
+        """When expected_output is None on the item, it should be passed as ''."""
+
+        @evaluator
+        def check_expected(expected_output: str) -> bool:
+            return expected_output == ""
+
+        result = await check_expected(_make_item(expected_output=None))
+        assert result.passed is True
+
+    @pytest.mark.asyncio
+    async def test_similarity_score(self):
+        @evaluator
+        def word_overlap(response: str, expected_output: str) -> float:
+            r_words = set(response.lower().split())
+            e_words = set(expected_output.lower().split())
+            if not e_words:
+                return 1.0
+            return len(r_words & e_words) / len(e_words)
+
+        item = _make_item(response="sunny warm day", expected_output="warm sunny afternoon")
+        result = await word_overlap(item)
+        assert result.passed is True  # 2/3 overlap ≥ 0.5
+
+
+# ---------------------------------------------------------------------------
+# Tier 3: full context (conversation, tools, context)
+# ---------------------------------------------------------------------------
+
+
+class TestTier3FullContext:
+    @pytest.mark.asyncio
+    async def test_conversation_access(self):
+        @evaluator
+        def multi_turn(query: str, response: str, *, conversation: list) -> bool:
+            return len(conversation) >= 2
+
+        item = _make_item(conversation=[Message("user", []), Message("assistant", [])])
+        assert (await multi_turn(item)).passed is True
+
+        item2 = _make_item(conversation=[Message("user", [])])
+        assert (await multi_turn(item2)).passed is False
+
+    @pytest.mark.asyncio
+    async def test_tools_access(self):
+        @evaluator
+        def has_tools(tools: list) -> bool:
+            return len(tools) > 0
+
+        mock_tool = type(
+            "MockTool",
+            (),
+            {"name": "get_weather", "description": "Get weather", "parameters": lambda self: {}},
+        )()
+        item = _make_item(tools=[mock_tool])
+        assert (await has_tools(item)).passed is True
+
+    @pytest.mark.asyncio
+    async def test_context_access(self):
+        @evaluator
+        def grounded(response: str, context: str) -> bool:
+            if not context:
+                return True
+            return any(word in response.lower() for word in context.lower().split())
+
+        item = _make_item(response="It's sunny", context="sunny warm")
+        assert (await grounded(item)).passed is True
+
+    @pytest.mark.asyncio
+    async def test_all_params(self):
+        @evaluator
+        def full_check(
+            query: str,
+            response: str,
+            expected_output: str,
+            conversation: list,
+            tools: list,
+            context: str,
+        ) -> bool:
+            return all([query, response, expected_output is not None, isinstance(conversation, list)])
+
+        item = _make_item(expected_output="foo", context="bar")
+        assert (await full_check(item)).passed is True
+
+
+# ---------------------------------------------------------------------------
+# Return type coercion
+# ---------------------------------------------------------------------------
+
+
+class TestReturnTypeCoercion:
+    @pytest.mark.asyncio
+    async def test_dict_with_score(self):
+        @evaluator
+        def scored(response: str) -> dict:
+            return {"score": 0.9, "reason": "good answer"}
+
+        result = await scored(_make_item())
+        assert result.passed is True
+        assert result.reason == "good answer"
+
+    @pytest.mark.asyncio
+    async def test_dict_with_score_below_threshold(self):
+        @evaluator
+        def low_scored(response: str) -> dict:
+            return {"score": 0.3}
+
+        result = await low_scored(_make_item())
+        assert result.passed is False
+
+    @pytest.mark.asyncio
+    async def test_dict_with_custom_threshold(self):
+        @evaluator
+        def custom_threshold(response: str) -> dict:
+            return {"score": 0.3, "threshold": 0.2}
+
+        result = await custom_threshold(_make_item())
+        assert result.passed is True
+
+    @pytest.mark.asyncio
+    async def test_dict_with_passed(self):
+        @evaluator
+        def explicit_pass(response: str) -> dict:
+            return {"passed": True, "reason": "all good"}
+
+        result = await explicit_pass(_make_item())
+        assert result.passed is True
+        assert result.reason == "all good"
+
+    @pytest.mark.asyncio
+    async def test_check_result_passthrough(self):
+        @evaluator
+        def returns_check_result(response: str) -> CheckResult:
+            return CheckResult(True, "direct result", "custom")
+
+        result = await returns_check_result(_make_item())
+        assert result.passed is True
+        assert result.reason == "direct result"
+        assert result.check_name == "custom"
+
+    @pytest.mark.asyncio
+    async def test_unsupported_return_type(self):
+        @evaluator
+        def bad_return(response: str) -> str:
+            return "oops"
+
+        with pytest.raises(TypeError, match="unsupported type"):
+            await bad_return(_make_item())
+
+    @pytest.mark.asyncio
+    async def test_int_return(self):
+        @evaluator
+        def int_score(response: str) -> int:
+            return 1
+
+        result = await int_score(_make_item())
+        assert result.passed is True
+
+
+# ---------------------------------------------------------------------------
+# Decorator variants
+# ---------------------------------------------------------------------------
+
+
+class TestDecoratorVariants:
+    @pytest.mark.asyncio
+    async def test_decorator_no_parens(self):
+        @evaluator
+        def my_check(response: str) -> bool:
+            return True
+
+        assert (await my_check(_make_item())).passed is True
+
+    @pytest.mark.asyncio
+    async def test_decorator_with_name(self):
+        @evaluator(name="custom_name")
+        def my_check(response: str) -> bool:
+            return True
+
+        assert my_check.__name__ == "custom_name"
+        result = await my_check(_make_item())
+        assert result.check_name == "custom_name"
+
+    @pytest.mark.asyncio
+    async def test_direct_call(self):
+        def raw_fn(query: str, response: str) -> bool:
+            return len(response) > 0
+
+        check = evaluator(raw_fn, name="direct")
+        result = await check(_make_item())
+        assert result.passed is True
+        assert result.check_name == "direct"
+
+
+# ---------------------------------------------------------------------------
+# Error handling
+# ---------------------------------------------------------------------------
+
+
+class TestErrorHandling:
+    @pytest.mark.asyncio
+    async def test_unknown_required_param_raises(self):
+        @evaluator
+        def bad_params(query: str, unknown_param: str) -> bool:
+            return True
+
+        with pytest.raises(TypeError, match="unknown required parameter"):
+            await bad_params(_make_item())
+
+    @pytest.mark.asyncio
+    async def test_unknown_optional_param_ok(self):
+        @evaluator
+        def optional_unknown(query: str, foo: str = "default") -> bool:
+            return foo == "default"
+
+        result = await optional_unknown(_make_item())
+        assert result.passed is True
+
+    @pytest.mark.asyncio
+    async def test_async_function_works_with_evaluator(self):
+        """Using an async function with @evaluator should work."""
+
+        @evaluator
+        async def async_fn(response: str) -> bool:
+            return True
+
+        result = async_fn(_make_item())
+        # Should return an awaitable
+        assert inspect.isawaitable(result)
+        check_result = await result
+        assert check_result.passed is True
+
+
+# ---------------------------------------------------------------------------
+# Integration with LocalEvaluator
+# ---------------------------------------------------------------------------
+
+
+class TestLocalEvaluatorIntegration:
+    @pytest.mark.asyncio
+    async def test_mixed_checks(self):
+        """Function evaluators mix with built-in checks in LocalEvaluator."""
+
+        @evaluator
+        def length_ok(response: str) -> bool:
+            return len(response) > 5
+
+        local = LocalEvaluator(
+            keyword_check("sunny"),
+            length_ok,
+        )
+        items = [_make_item()]
+        results = await local.evaluate(items, eval_name="mixed test")
+
+        assert results.status == "completed"
+        assert results.result_counts["passed"] == 1
+        assert results.result_counts["failed"] == 0
+
+    @pytest.mark.asyncio
+    async def test_evaluator_failure_counted(self):
+        @evaluator
+        def always_fail(response: str) -> bool:
+            return False
+
+        local = LocalEvaluator(always_fail)
+        results = await local.evaluate([_make_item()])
+
+        assert results.result_counts["failed"] == 1
+
+    @pytest.mark.asyncio
+    async def test_multiple_evaluators(self):
+        @evaluator
+        def check_a(response: str) -> float:
+            return 0.9
+
+        @evaluator
+        def check_b(query: str, response: str, expected_output: str) -> bool:
+            return True
+
+        @evaluator(name="check_c")
+        def check_c(response: str, conversation: list) -> dict:
+            return {"score": 0.8, "reason": "looks good"}
+
+        local = LocalEvaluator(check_a, check_b, check_c)
+        results = await local.evaluate([_make_item(expected_output="test")])
+
+        assert results.result_counts["passed"] == 1
+        assert "check_a" in results.per_evaluator
+        assert "check_b" in results.per_evaluator
+        assert "check_c" in results.per_evaluator
+
+
+# ---------------------------------------------------------------------------
+# Async evaluator (via @evaluator which handles async automatically)
+# ---------------------------------------------------------------------------
+
+
+class TestAsyncFunctionEvaluator:
+    @pytest.mark.asyncio
+    async def test_async_evaluator_in_local(self):
+        @evaluator
+        async def async_check(query: str, response: str) -> bool:
+            return len(response) > 0
+
+        local = LocalEvaluator(async_check)
+        results = await local.evaluate([_make_item()])
+        assert results.result_counts["passed"] == 1
+
+    @pytest.mark.asyncio
+    async def test_async_with_name(self):
+        @evaluator(name="named_async")
+        async def my_async(response: str) -> float:
+            return 0.75
+
+        result = await my_async(_make_item())
+        assert result.passed is True
+        assert result.check_name == "named_async"
+
+
+# ---------------------------------------------------------------------------
+# Auto-wrapping bare checks in evaluate_agent
+# ---------------------------------------------------------------------------
+
+
+class TestAutoWrapEvalChecks:
+    @pytest.mark.asyncio
+    async def test_bare_check_in_evaluators_list(self):
+        """Bare EvalCheck callables are auto-wrapped in LocalEvaluator."""
+        from agent_framework._evaluation import _run_evaluators
+
+        @evaluator
+        def is_long(response: str) -> bool:
+            return len(response.split()) > 2
+
+        items = [_make_item(response="It is sunny and warm today")]
+        results = await _run_evaluators(is_long, items, eval_name="test")
+        assert len(results) == 1
+        assert results[0].result_counts["passed"] == 1
+
+    @pytest.mark.asyncio
+    async def test_mixed_evaluators_and_checks(self):
+        """Mix of Evaluator instances and bare checks works."""
+        from agent_framework._evaluation import _run_evaluators
+
+        @evaluator
+        def has_words(response: str) -> bool:
+            return len(response.split()) > 0
+
+        local = LocalEvaluator(keyword_check("sunny"))
+
+        items = [_make_item(response="It is sunny")]
+        results = await _run_evaluators([local, has_words], items, eval_name="test")
+        assert len(results) == 2
+        assert all(r.result_counts["passed"] == 1 for r in results)
+
+    @pytest.mark.asyncio
+    async def test_adjacent_checks_grouped(self):
+        """Adjacent bare checks are grouped into a single LocalEvaluator."""
+        from agent_framework._evaluation import _run_evaluators
+
+        @evaluator
+        def check_a(response: str) -> bool:
+            return True
+
+        @evaluator
+        def check_b(response: str) -> bool:
+            return True
+
+        items = [_make_item()]
+        results = await _run_evaluators([check_a, check_b], items, eval_name="test")
+        # Two adjacent checks → one LocalEvaluator → one result
+        assert len(results) == 1
+        assert results[0].result_counts["passed"] == 1
+
+
+# ---------------------------------------------------------------------------
+# Expected Tool Calls
+# ---------------------------------------------------------------------------
+
+
+def _make_tool_call_item(
+    calls: list[tuple[str, dict | None]],
+    expected: list[ExpectedToolCall] | None = None,
+) -> EvalItem:
+    """Build an EvalItem with tool calls in the conversation."""
+    msgs: list[Message] = [Message("user", ["Do something"])]
+    for name, args in calls:
+        msgs.append(Message("assistant", [Content.from_function_call("call_" + name, name, arguments=args)]))
+    msgs.append(Message("assistant", ["Done"]))
+    return EvalItem(conversation=msgs, expected_tool_calls=expected)
+
+
+class TestExpectedToolCallType:
+    def test_name_only(self):
+        tc = ExpectedToolCall("get_weather")
+        assert tc.name == "get_weather"
+        assert tc.arguments is None
+
+    def test_name_and_args(self):
+        tc = ExpectedToolCall("get_weather", {"location": "NYC"})
+        assert tc.name == "get_weather"
+        assert tc.arguments == {"location": "NYC"}
+
+
+class TestToolCallsPresent:
+    def test_all_present(self):
+        item = _make_tool_call_item(
+            calls=[("get_weather", None), ("get_news", None)],
+            expected=[ExpectedToolCall("get_weather"), ExpectedToolCall("get_news")],
+        )
+        result = tool_calls_present(item)
+        assert result.passed is True
+        assert result.check_name == "tool_calls_present"
+
+    def test_missing_tool(self):
+        item = _make_tool_call_item(
+            calls=[("get_weather", None)],
+            expected=[ExpectedToolCall("get_weather"), ExpectedToolCall("get_news")],
+        )
+        result = tool_calls_present(item)
+        assert result.passed is False
+        assert "get_news" in result.reason
+
+    def test_extras_ok(self):
+        item = _make_tool_call_item(
+            calls=[("get_weather", None), ("get_news", None), ("get_stock", None)],
+            expected=[ExpectedToolCall("get_weather")],
+        )
+        result = tool_calls_present(item)
+        assert result.passed is True
+
+    def test_no_expected(self):
+        item = _make_tool_call_item(calls=[("get_weather", None)])
+        result = tool_calls_present(item)
+        assert result.passed is True
+        assert "No expected" in result.reason
+
+
+class TestToolCallArgsMatch:
+    def test_name_only_match(self):
+        item = _make_tool_call_item(
+            calls=[("get_weather", {"location": "NYC"})],
+            expected=[ExpectedToolCall("get_weather")],
+        )
+        result = tool_call_args_match(item)
+        assert result.passed is True
+
+    def test_args_exact_match(self):
+        item = _make_tool_call_item(
+            calls=[("get_weather", {"location": "NYC", "units": "fahrenheit"})],
+            expected=[ExpectedToolCall("get_weather", {"location": "NYC"})],
+        )
+        # Subset match — extra "units" key is OK
+        result = tool_call_args_match(item)
+        assert result.passed is True
+
+    def test_args_mismatch(self):
+        item = _make_tool_call_item(
+            calls=[("get_weather", {"location": "LA"})],
+            expected=[ExpectedToolCall("get_weather", {"location": "NYC"})],
+        )
+        result = tool_call_args_match(item)
+        assert result.passed is False
+        assert "args mismatch" in result.reason
+
+    def test_tool_not_called(self):
+        item = _make_tool_call_item(
+            calls=[("get_news", None)],
+            expected=[ExpectedToolCall("get_weather", {"location": "NYC"})],
+        )
+        result = tool_call_args_match(item)
+        assert result.passed is False
+        assert "not called" in result.reason
+
+    def test_multiple_expected(self):
+        item = _make_tool_call_item(
+            calls=[
+                ("get_weather", {"location": "NYC"}),
+                ("book_flight", {"destination": "LA", "date": "tomorrow"}),
+            ],
+            expected=[
+                ExpectedToolCall("get_weather", {"location": "NYC"}),
+                ExpectedToolCall("book_flight", {"destination": "LA"}),
+            ],
+        )
+        result = tool_call_args_match(item)
+        assert result.passed is True
+
+    def test_no_expected(self):
+        item = _make_tool_call_item(calls=[("get_weather", None)])
+        result = tool_call_args_match(item)
+        assert result.passed is True
+
+
+class TestExpectedToolCallsFieldInjection:
+    """Test that @evaluator can receive expected_tool_calls via parameter injection."""
+
+    @pytest.mark.asyncio
+    async def test_injection(self):
+        @evaluator
+        def check_tools(expected_tool_calls: list) -> bool:
+            return len(expected_tool_calls) == 2
+
+        item = _make_tool_call_item(
+            calls=[],
+            expected=[ExpectedToolCall("a"), ExpectedToolCall("b")],
+        )
+        result = await check_tools(item)
+        assert result.passed is True
+
+    @pytest.mark.asyncio
+    async def test_injection_empty_default(self):
+        @evaluator
+        def check_tools(expected_tool_calls: list) -> bool:
+            return len(expected_tool_calls) == 0
+
+        item = _make_tool_call_item(calls=[])
+        result = await check_tools(item)
+        assert result.passed is True
+
+
+# ---------------------------------------------------------------------------
+# Per-item results (auditing)
+# ---------------------------------------------------------------------------
+
+
+class TestPerItemResults:
+    """LocalEvaluator should produce per-item EvalItemResult with query/response."""
+
+    @pytest.mark.asyncio
+    async def test_items_populated_with_query_and_response(self):
+        @evaluator
+        def is_sunny(response: str) -> bool:
+            return "sunny" in response.lower()
+
+        item = _make_item(query="Weather?", response="It's sunny!")
+        local = LocalEvaluator(is_sunny)
+        results = await local.evaluate([item])
+
+        assert len(results.items) == 1
+        ri = results.items[0]
+        assert ri.item_id == "0"
+        assert ri.status == "pass"
+        assert ri.input_text == "Weather?"
+        assert ri.output_text == "It's sunny!"
+        assert len(ri.scores) == 1
+        assert ri.scores[0].name == "is_sunny"
+        assert ri.scores[0].passed is True
+
+    @pytest.mark.asyncio
+    async def test_items_populated_on_failure(self):
+        @evaluator
+        def always_fail(response: str) -> bool:
+            return False
+
+        item = _make_item(query="Hello", response="World")
+        local = LocalEvaluator(always_fail)
+        results = await local.evaluate([item])
+
+        assert len(results.items) == 1
+        ri = results.items[0]
+        assert ri.status == "fail"
+        assert ri.input_text == "Hello"
+        assert ri.output_text == "World"
+        assert ri.scores[0].passed is False
+        assert ri.scores[0].score == 0.0
+
+    @pytest.mark.asyncio
+    async def test_multiple_items_indexed(self):
+        @evaluator
+        def pass_all(response: str) -> bool:
+            return True
+
+        items = [
+            _make_item(query="Q1", response="R1"),
+            _make_item(query="Q2", response="R2"),
+        ]
+        local = LocalEvaluator(pass_all)
+        results = await local.evaluate(items)
+
+        assert len(results.items) == 2
+        assert results.items[0].item_id == "0"
+        assert results.items[0].input_text == "Q1"
+        assert results.items[0].output_text == "R1"
+        assert results.items[1].item_id == "1"
+        assert results.items[1].input_text == "Q2"
+        assert results.items[1].output_text == "R2"
+
+
+# ---------------------------------------------------------------------------
+# num_repetitions validation
+# ---------------------------------------------------------------------------
+
+
+class TestNumRepetitions:
+    """Tests for the num_repetitions parameter on evaluate_agent."""
+
+    @pytest.mark.asyncio
+    async def test_num_repetitions_validation_rejects_zero(self):
+        from agent_framework._evaluation import evaluate_agent
+
+        with pytest.raises(ValueError, match="num_repetitions must be >= 1"):
+            await evaluate_agent(
+                queries=["Hello"],
+                evaluators=LocalEvaluator(keyword_check("hello")),
+                num_repetitions=0,
+            )
+
+    @pytest.mark.asyncio
+    async def test_num_repetitions_validation_rejects_negative(self):
+        from agent_framework._evaluation import evaluate_agent
+
+        with pytest.raises(ValueError, match="num_repetitions must be >= 1"):
+            await evaluate_agent(
+                queries=["Hello"],
+                evaluators=LocalEvaluator(keyword_check("hello")),
+                num_repetitions=-1,
+            )

python/packages/core/tests/workflow/test_full_conversation.py

@@ -460,10 +460,10 @@ async def test_run_request_with_full_history_clears_service_session_id() -> None
     assert spy_agent._captured_service_session_id is None  # pyright: ignore[reportPrivateUsage]
 
 
-async def test_from_response_preserves_service_session_id() -> None:
-    """from_response hands off a prior agent's full conversation to the next executor.
-    The receiving executor's service_session_id is preserved so the API can continue
-    the conversation using previous_response_id."""
+async def test_from_response_clears_service_session_id_on_new_run() -> None:
+    """service_session_id set before a workflow run is cleared by the executor reset
+    that happens at the start of each run, preventing stale previous_response_id
+    from leaking between runs."""
     tool_agent = _ToolHistoryAgent(id="tool_agent2", name="ToolAgent", summary_text="Done.")
     tool_exec = AgentExecutor(tool_agent, id="tool_agent2")
 
@@ -477,4 +477,6 @@ async def test_from_response_preserves_service_session_id() -> None:
     result = await wf.run("start")
     assert result.get_outputs() is not None
 
-    assert spy_agent._captured_service_session_id == "resp_PREVIOUS_RUN"  # pyright: ignore[reportPrivateUsage]
+    # service_session_id is cleared at the start of run() to prevent stale
+    # previous_response_id from causing "No tool output found" errors on re-runs.
+    assert spy_agent._captured_service_session_id is None  # pyright: ignore[reportPrivateUsage]

python/samples/02-agents/evaluation/evaluate_agent.py

@@ -0,0 +1,68 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Evaluate an agent with local checks — no API keys needed.
+
+Demonstrates the simplest evaluation workflow:
+1. Define checks using the @evaluator decorator
+2. Run evaluate_agent() which calls agent.run() under the covers
+3. Assert results in CI or inspect interactively
+
+Usage:
+    uv run python samples/02-agents/evaluation/evaluate_agent.py
+"""
+
+import asyncio
+
+from agent_framework import (
+    Agent,
+    LocalEvaluator,
+    evaluate_agent,
+    evaluator,
+    keyword_check,
+)
+
+
+# A custom check — parameter names determine what data you receive
+@evaluator
+def is_helpful(response: str) -> bool:
+    """Check the response isn't empty or a refusal."""
+    refusals = ["i can't", "i'm not able", "i don't know"]
+    return len(response) > 10 and not any(r in response.lower() for r in refusals)
+
+
+async def main():
+    agent = Agent(
+        model="gpt-4o-mini",
+        instructions="You are a helpful weather assistant.",
+    )
+
+    # Combine built-in and custom checks
+    local = LocalEvaluator(
+        keyword_check("weather"),  # response must mention "weather"
+        is_helpful,  # custom check
+    )
+
+    # evaluate_agent() calls agent.run() for each query, then evaluates
+    results = await evaluate_agent(
+        agent=agent,
+        queries=[
+            "What's the weather like in Seattle?",
+            "Will it rain in London tomorrow?",
+            "What should I wear for 30°C weather?",
+        ],
+        evaluators=local,
+    )
+
+    for r in results:
+        print(f"{r.provider}: {r.passed}/{r.total} passed")
+        for item in r.items:
+            print(f"  [{item.status}] Q: {item.input_text[:50]}  A: {item.output_text[:50]}...")
+            for score in item.scores:
+                print(f"    {score.name}: {'✓' if score.passed else '✗'}")
+
+    # Use in CI: will raise AssertionError if any check fails
+    # results[0].assert_passed()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/02-agents/evaluation/evaluate_with_expected.py

@@ -0,0 +1,64 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Evaluate an agent with expected outputs and tool call checks.
+
+Demonstrates ground-truth comparison and tool usage evaluation:
+1. Provide expected outputs alongside queries
+2. Use built-in tool_calls_present for tool verification
+3. Combine multiple evaluation criteria
+
+Usage:
+    uv run python samples/02-agents/evaluation/evaluate_with_expected.py
+"""
+
+import asyncio
+
+from agent_framework import (
+    Agent,
+    LocalEvaluator,
+    evaluate_agent,
+    evaluator,
+    tool_calls_present,
+)
+
+
+@evaluator
+def response_matches_expected(response: str, expected_output: str) -> float:
+    """Score based on word overlap with expected output."""
+    if not expected_output:
+        return 1.0
+    response_words = set(response.lower().split())
+    expected_words = set(expected_output.lower().split())
+    return len(response_words & expected_words) / max(len(expected_words), 1)
+
+
+async def main():
+    agent = Agent(
+        model="gpt-4o-mini",
+        instructions="You are a math tutor. Answer concisely.",
+    )
+
+    local = LocalEvaluator(
+        response_matches_expected,
+        tool_calls_present,  # verifies expected tools were called
+    )
+
+    results = await evaluate_agent(
+        agent=agent,
+        queries=["What is 2 + 2?", "What is the square root of 144?"],
+        expected_output=["4", "12"],
+        expected_tool_calls=[
+            [],  # no tools expected for simple math
+            [],
+        ],
+        evaluators=local,
+    )
+
+    for r in results:
+        print(f"{r.provider}: {r.passed}/{r.total} passed")
+        for item in r.items:
+            print(f"  [{item.status}] {item.input_text} → {item.output_text[:80]}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/03-workflows/evaluation/evaluate_workflow.py

@@ -0,0 +1,60 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Evaluate a multi-agent workflow with per-agent breakdown.
+
+Demonstrates workflow evaluation:
+1. Build a simple two-agent workflow
+2. Run evaluate_workflow() which runs the workflow and evaluates each agent
+3. Inspect per-agent results in sub_results
+
+Usage:
+    uv run python samples/03-workflows/evaluation/evaluate_workflow.py
+"""
+
+import asyncio
+
+from agent_framework import (
+    Agent,
+    AgentExecutor,
+    LocalEvaluator,
+    WorkflowBuilder,
+    evaluate_workflow,
+    evaluator,
+    keyword_check,
+)
+
+
+@evaluator
+def is_nonempty(response: str) -> bool:
+    """Check the agent produced a non-trivial response."""
+    return len(response.strip()) > 5
+
+
+async def main():
+    # Build a simple planner → executor workflow
+    planner = Agent(model="gpt-4o-mini", instructions="You plan trips. Output a bullet-point plan.")
+    executor_agent = Agent(model="gpt-4o-mini", instructions="You execute travel plans. Book the items listed.")
+
+    builder = WorkflowBuilder()
+    builder.add_executor(AgentExecutor("planner", planner))
+    builder.add_executor(AgentExecutor("booker", executor_agent))
+    builder.add_edge("planner", "booker")
+    workflow = builder.build()
+
+    # Evaluate with per-agent breakdown
+    local = LocalEvaluator(is_nonempty, keyword_check("plan", "trip"))
+
+    results = await evaluate_workflow(
+        workflow=workflow,
+        queries=["Plan a weekend trip to Paris"],
+        evaluators=local,
+    )
+
+    for r in results:
+        print(f"{r.provider}: {r.passed}/{r.total} passed (overall)")
+        for agent_name, sub in r.sub_results.items():
+            print(f"  {agent_name}: {sub.passed}/{sub.total}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/05-end-to-end/evaluation/foundry_evals/.env.example

@@ -0,0 +1,3 @@
+AZURE_AI_PROJECT_ENDPOINT="<your-project-endpoint>"
+AZURE_AI_MODEL_DEPLOYMENT_NAME="<your-model-deployment>"
+

python/samples/05-end-to-end/evaluation/foundry_evals/README.md

@@ -0,0 +1,46 @@
+# Foundry Evals Integration Samples
+
+These samples demonstrate evaluating agent-framework agents using Azure AI Foundry's built-in evaluators.
+
+## Available Evaluators
+
+| Category | Evaluators |
+|----------|-----------|
+| **Agent behavior** | `intent_resolution`, `task_adherence`, `task_completion`, `task_navigation_efficiency` |
+| **Tool usage** | `tool_call_accuracy`, `tool_selection`, `tool_input_accuracy`, `tool_output_utilization`, `tool_call_success` |
+| **Quality** | `coherence`, `fluency`, `relevance`, `groundedness`, `response_completeness`, `similarity` |
+| **Safety** | `violence`, `sexual`, `self_harm`, `hate_unfairness` |
+
+## Samples
+
+### `evaluate_agent_sample.py` — Dataset Evaluation (Path 3)
+
+The dev inner loop. Two patterns from simplest to most control:
+
+1. **`evaluate_agent()`** — One call: runs agent → converts → evaluates
+2. **`evaluate_dataset()`** — Run agent yourself, convert with `AgentEvalConverter`, inspect/modify, then evaluate
+
+```bash
+uv run samples/05-end-to-end/evaluation/foundry_evals/evaluate_agent_sample.py
+```
+
+### `evaluate_traces_sample.py` — Trace & Response Evaluation (Path 1)
+
+Evaluate what already happened — zero changes to agent code:
+
+1. **`evaluate_responses()`** — Evaluate Responses API responses by ID
+2. **`evaluate_traces()`** — Evaluate from OTel traces in App Insights
+
+```bash
+uv run samples/05-end-to-end/evaluation/foundry_evals/evaluate_traces_sample.py
+```
+
+## Setup
+
+Create a `.env` file with configuration as in the `.env.example` file in this folder.
+
+## Which sample should I start with?
+
+- **"I want to test my agent during development"** → `evaluate_agent_sample.py`, Pattern 1
+- **"I want to evaluate past agent runs"** → `evaluate_traces_sample.py`
+- **"I want to inspect/modify eval data before submitting"** → `evaluate_agent_sample.py`, Pattern 2

python/samples/05-end-to-end/evaluation/foundry_evals/evaluate_agent_sample.py

@@ -0,0 +1,195 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import os
+
+from agent_framework import Agent, AgentEvalConverter, ConversationSplit, evaluate_agent
+from agent_framework.azure import AzureOpenAIResponsesClient
+from agent_framework_azure_ai import FoundryEvals
+from azure.ai.projects.aio import AIProjectClient
+from azure.identity import DefaultAzureCredential
+from dotenv import load_dotenv
+
+load_dotenv()
+
+"""
+This sample demonstrates evaluating an agent using Azure AI Foundry's built-in evaluators.
+
+It shows three patterns:
+1. evaluate_agent(responses=...) — Evaluate a response you already have.
+2. evaluate_agent(queries=...) — Run the agent against test queries and evaluate in one call.
+3. FoundryEvals.evaluate() — Full control with direct evaluator access.
+
+Prerequisites:
+- An Azure AI Foundry project with a deployed model
+- Set AZURE_AI_PROJECT_ENDPOINT and AZURE_AI_MODEL_DEPLOYMENT_NAME in .env
+
+Required components:
+- An Agent with tools (the agent to evaluate)
+- A FoundryEvals instance (the evaluator)
+"""
+
+
+# Define a simple tool for the agent
+def get_weather(location: str) -> str:
+    """Get the current weather for a location."""
+    weather_data = {
+        "seattle": "62°F, cloudy with a chance of rain",
+        "london": "55°F, overcast",
+        "paris": "68°F, partly sunny",
+    }
+    return weather_data.get(location.lower(), f"Weather data not available for {location}")
+
+
+def get_flight_price(origin: str, destination: str) -> str:
+    """Get the price of a flight between two cities."""
+    return f"Flights from {origin} to {destination}: $450 round-trip"
+
+
+async def main():
+    # 1. Set up the Azure AI project client
+    project_client = AIProjectClient(
+        endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
+        credential=DefaultAzureCredential(),
+    )
+
+    deployment = os.environ.get("AZURE_AI_MODEL_DEPLOYMENT_NAME", "gpt-4o")
+
+    # 2. Create an agent with tools
+    agent = Agent(
+        client=AzureOpenAIResponsesClient(
+            project_client=project_client,
+            deployment_name=deployment,
+        ),
+        name="travel-assistant",
+        instructions=(
+            "You are a helpful travel assistant. Use your tools to answer questions about weather and flights."
+        ),
+        tools=[get_weather, get_flight_price],
+    )
+
+    # 3. Create the evaluator — provider config goes here, once
+    evals = FoundryEvals(project_client=project_client, model_deployment=deployment)
+
+    # =========================================================================
+    # Pattern 1: evaluate_agent(responses=...) — evaluate a response you already have
+    # =========================================================================
+    print("=" * 60)
+    print("Pattern 1: evaluate_agent(responses=...) — evaluate existing response")
+    print("=" * 60)
+
+    query = "How much does a flight from Seattle to Paris cost?"
+    response = await agent.run(query)
+    print(f"Agent said: {response.text[:100]}...")
+
+    # Pass agent= so tool definitions are extracted, queries= for the eval item context
+    results = await evaluate_agent(
+        agent=agent,
+        responses=response,
+        queries=[query],
+        evaluators=evals.select(FoundryEvals.RELEVANCE, FoundryEvals.TOOL_CALL_ACCURACY),
+    )
+
+    for r in results:
+        print(f"Status: {r.status}")
+        print(f"Results: {r.passed}/{r.total} passed")
+        print(f"Portal: {r.report_url}")
+        if r.all_passed:
+            print("✓ All passed")
+        else:
+            print(f"✗ {r.failed} failed, {r.errored} errored")
+
+    # =========================================================================
+    # Pattern 2a: evaluate_agent() — batch test queries
+    # =========================================================================
+    print()
+    print("=" * 60)
+    print("Pattern 2a: evaluate_agent()")
+    print("=" * 60)
+
+    # Calls agent.run() under the covers for each query, then evaluates
+    results = await evaluate_agent(
+        agent=agent,
+        queries=[
+            "What's the weather like in Seattle?",
+            "How much does a flight from Seattle to Paris cost?",
+            "What should I pack for London?",
+        ],
+        evaluators=evals,  # uses smart defaults (auto-adds tool_call_accuracy)
+    )
+
+    for r in results:
+        print(f"Status: {r.status}")
+        print(f"Results: {r.passed}/{r.total} passed")
+        print(f"Portal: {r.report_url}")
+        if r.all_passed:
+            print("✓ All passed")
+        else:
+            print(f"✗ {r.failed} failed, {r.errored} errored")
+
+    # =========================================================================
+    # Pattern 2b: evaluate_agent() — with conversation split override
+    # =========================================================================
+    print()
+    print("=" * 60)
+    print("Pattern 2b: evaluate_agent() with conversation_split")
+    print("=" * 60)
+
+    # conversation_split forces all evaluators to use the same split strategy.
+    # FULL evaluates the entire conversation trajectory against the original query.
+    results = await evaluate_agent(
+        agent=agent,
+        queries=[
+            "What's the weather like in Seattle?",
+            "What should I pack for London?",
+        ],
+        evaluators=evals,
+        conversation_split=ConversationSplit.FULL,  # overrides evaluator defaults
+    )
+
+    for r in results:
+        print(f"Status: {r.status}")
+        print(f"Results: {r.passed}/{r.total} passed")
+        print(f"Portal: {r.report_url}")
+        if r.all_passed:
+            print("✓ All passed")
+        else:
+            print(f"✗ {r.failed} failed, {r.errored} errored")
+
+    # =========================================================================
+    # Pattern 3: FoundryEvals.evaluate() — manual control
+    # =========================================================================
+    print()
+    print("=" * 60)
+    print("Pattern 3: FoundryEvals.evaluate() — manual control")
+    print("=" * 60)
+
+    queries = [
+        "What's the weather in Paris?",
+        "Find me a flight from London to Seattle",
+    ]
+
+    items = []
+    for q in queries:
+        response = await agent.run(q)
+        print(f"Query: {q}")
+        print(f"Response: {response.text[:100]}...")
+
+        item = AgentEvalConverter.to_eval_item(query=q, response=response, agent=agent)
+        items.append(item)
+
+        print(f"  Has tools: {item.tools is not None}")
+        if item.tools:
+            print(f"  Tools: {[t.name for t in item.tools]}")
+
+    # Submit directly to the evaluator
+    tool_evals = evals.select(FoundryEvals.RELEVANCE, FoundryEvals.TOOL_CALL_ACCURACY)
+    results = await tool_evals.evaluate(items, eval_name="Travel Assistant Eval")
+
+    print(f"\nStatus: {results.status}")
+    print(f"Results: {results.passed}/{results.total} passed")
+    print(f"Portal: {results.report_url}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/05-end-to-end/evaluation/foundry_evals/evaluate_all_patterns_sample.py

@@ -0,0 +1,544 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""
+Agent Evaluation — Complete Guide
+==================================
+
+This sample shows every way to evaluate agents and workflows in
+Microsoft Agent Framework. Run the sections that match your needs.
+
+                    ┌──────────────────────────────────────┐
+                    │         Evaluation Options            │
+                    ├──────────────────────────────────────┤
+                    │                                      │
+                    │  1. Your own function  (no setup)    │
+                    │  2. Built-in checks   (no setup)     │
+                    │  3. Azure AI Foundry  (cloud)        │
+                    │  4. Mix them all      (recommended)  │
+                    │                                      │
+                    └──────────────────────────────────────┘
+
+Each evaluator plugs into the same two entry points:
+
+    evaluate_agent()     — run agent + evaluate, or evaluate existing responses
+    evaluate_workflow()  — evaluate multi-agent workflows with per-agent breakdown
+"""
+
+import asyncio
+import os
+
+from agent_framework import (
+    Agent,
+    LocalEvaluator,
+    Message,
+    evaluate_agent,
+    evaluate_workflow,
+    evaluator,
+    keyword_check,
+    tool_called_check,
+)
+from agent_framework.azure import AzureOpenAIResponsesClient
+from agent_framework_azure_ai import FoundryEvals
+from agent_framework_orchestrations import GroupChatBuilder, SequentialBuilder
+from azure.ai.projects.aio import AIProjectClient
+from azure.identity import DefaultAzureCredential
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+# ── Tools for our agents ─────────────────────────────────────────────────────
+
+
+def get_weather(location: str) -> str:
+    """Get the current weather for a location."""
+    return {"seattle": "62°F, cloudy", "london": "55°F, overcast", "paris": "68°F, sunny"}.get(
+        location.lower(), f"No data for {location}"
+    )
+
+
+def get_flight_price(origin: str, destination: str) -> str:
+    """Get the price of a flight between two cities."""
+    return f"Flights from {origin} to {destination}: $450 round-trip"
+
+
+# ── Output helpers ────────────────────────────────────────────────────────────
+
+
+def print_workflow_results(results):
+    """Print workflow eval results with clear provider → overall → per-agent hierarchy."""
+    for r in results:
+        status = "✓" if r.all_passed else "✗"
+        print(f"\n  {r.provider}:")
+        print(f"    {status} overall: {r.passed}/{r.total} passed")
+        if r.report_url:
+            print(f"      Portal: {r.report_url}")
+        for agent_name, sub in r.sub_results.items():
+            agent_status = "✓" if sub.all_passed else "✗"
+            print(f"      {agent_status} {agent_name}: {sub.passed}/{sub.total}")
+            if sub.report_url:
+                print(f"        Portal: {sub.report_url}")
+
+
+# ── Agent setup ───────────────────────────────────────────────────────────────
+
+
+def create_agent(project_client, deployment):
+    """Create a travel assistant agent."""
+    return Agent(
+        client=AzureOpenAIResponsesClient(
+            project_client=project_client,
+            deployment_name=deployment,
+        ),
+        name="travel-assistant",
+        instructions="You are a helpful travel assistant. Use your tools to answer questions.",
+        tools=[get_weather, get_flight_price],
+    )
+
+
+def create_workflow(project_client, deployment):
+    """Create a researcher → planner sequential workflow."""
+    client = AzureOpenAIResponsesClient(
+        project_client=project_client,
+        deployment_name=deployment,
+    )
+    researcher = Agent(
+        client=client,
+        name="researcher",
+        instructions="You are a travel researcher. Use tools to gather weather and flight info.",
+        tools=[get_weather, get_flight_price],
+        default_options={"store": False},
+    )
+    planner = Agent(
+        client=client,
+        name="planner",
+        instructions="You are a travel planner. Create a concise recommendation from the research.",
+        default_options={"store": False},
+    )
+    return SequentialBuilder(participants=[researcher, planner]).build()
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# Section 1: Custom Function Evaluators
+# ═════════════════════════════════════════════════════════════════════════════
+#
+# Write a plain Python function. Name your parameters to get the data you need.
+# Return bool, float (≥0.5 = pass), or dict.
+#
+#   Available parameters:
+#     query, response, expected_output, conversation, tool_definitions, context
+#
+
+# ── Simple check: just query + response ──────────────────────────────────────
+
+
+@evaluator
+def is_helpful(response: str) -> bool:
+    """Response should be more than a one-liner."""
+    return len(response.split()) > 10
+
+
+@evaluator
+def no_apologies(query: str, response: str) -> bool:
+    """Agent shouldn't start with 'I'm sorry' or 'I apologize'."""
+    lower = response.lower().strip()
+    return not lower.startswith("i'm sorry") and not lower.startswith("i apologize")
+
+
+# ── Scored check: return a float ─────────────────────────────────────────────
+
+
+@evaluator
+def relevance_keyword_overlap(query: str, response: str) -> float:
+    """Score based on how many query words appear in the response."""
+    query_words = set(query.lower().split()) - {"the", "a", "in", "to", "is", "what", "how"}
+    response_lower = response.lower()
+    if not query_words:
+        return 1.0
+    return sum(1 for w in query_words if w in response_lower) / len(query_words)
+
+
+# ── Ground truth check: compare against expected output ──────────────────────
+
+
+@evaluator
+def mentions_expected_city(response: str, expected_output: str) -> bool:
+    """Response should mention the expected city."""
+    return expected_output.lower() in response.lower()
+
+
+# ── Full context check: inspect conversation and tools ───────────────────────
+
+
+@evaluator
+def used_available_tools(conversation: list, tool_definitions: list) -> dict:
+    """Check that the agent actually called at least one of its tools."""
+    available = {t.get("name", "") for t in (tool_definitions or [])}
+    called = set()
+    for msg in conversation:
+        for tc in msg.get("tool_calls", []):
+            name = tc.get("function", {}).get("name", "")
+            if name:
+                called.add(name)
+        for ci in msg.get("content", []):
+            if isinstance(ci, dict) and ci.get("type") == "tool_call":
+                called.add(ci.get("name", ""))
+    used = called & available
+    return {
+        "passed": len(used) > 0,
+        "reason": f"Used {sorted(used)}" if used else f"No tools called (available: {sorted(available)})",
+    }
+
+
+async def demo_evaluators(project_client, deployment):
+    """Evaluate an agent with custom function evaluators."""
+    print()
+    print("═" * 60)
+    print("  1. Custom Function Evaluators")
+    print("═" * 60)
+
+    agent = create_agent(project_client, deployment)
+
+    local = LocalEvaluator(
+        is_helpful,
+        no_apologies,
+        relevance_keyword_overlap,
+        used_available_tools,
+    )
+
+    results = await evaluate_agent(
+        agent=agent,
+        queries=["What's the weather in Seattle?", "How much is a flight to Paris?"],
+        evaluators=local,
+    )
+
+    for r in results:
+        print(f"\n  {r.provider}: {r.passed}/{r.total} passed")
+        for check, counts in r.per_evaluator.items():
+            status = "✓" if counts["failed"] == 0 else "✗"
+            print(f"    {status} {check}: {counts['passed']}/{counts['passed'] + counts['failed']}")
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# Section 2: Built-in Local Checks
+# ═════════════════════════════════════════════════════════════════════════════
+#
+# Pre-built checks for common patterns — no function needed.
+#
+
+
+async def demo_builtin_checks(project_client, deployment):
+    """Evaluate with built-in keyword and tool checks."""
+    print()
+    print("═" * 60)
+    print("  2. Built-in Local Checks")
+    print("═" * 60)
+
+    agent = create_agent(project_client, deployment)
+
+    local = LocalEvaluator(
+        keyword_check("weather", "seattle"),  # response must contain these words
+        tool_called_check("get_weather"),  # agent must have called this tool
+    )
+
+    results = await evaluate_agent(
+        agent=agent,
+        queries=["What's the weather in Seattle?"],
+        evaluators=local,
+    )
+
+    for r in results:
+        status = "✓" if r.all_passed else "✗"
+        print(f"\n  {status} {r.provider}: {r.passed}/{r.total} passed")
+        for check, counts in r.per_evaluator.items():
+            print(f"    {check}: {counts}")
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# Section 3: Azure AI Foundry Evaluators
+# ═════════════════════════════════════════════════════════════════════════════
+#
+# Cloud-powered AI quality assessment. Evaluates relevance, coherence,
+# task adherence, tool usage, and more.
+#
+
+
+async def demo_foundry_agent(project_client, deployment):
+    """Evaluate a single agent with Foundry."""
+    print()
+    print("═" * 60)
+    print("  3a. Foundry — Single Agent")
+    print("═" * 60)
+
+    agent = create_agent(project_client, deployment)
+    evals = FoundryEvals(project_client=project_client, model_deployment=deployment)
+
+    # evaluate_agent: run + evaluate in one call
+    results = await evaluate_agent(
+        agent=agent,
+        queries=["What's the weather in Seattle?", "Find flights from London to Paris"],
+        evaluators=evals,
+    )
+
+    for r in results:
+        print(f"\n  {r.provider}: {r.passed}/{r.total} passed")
+        print(f"  Portal: {r.report_url}")
+
+
+async def demo_foundry_response(project_client, deployment):
+    """Evaluate a response you already have."""
+    print()
+    print("═" * 60)
+    print("  3b. Foundry — Existing Response")
+    print("═" * 60)
+
+    agent = create_agent(project_client, deployment)
+
+    # Run the agent yourself
+    response = await agent.run([Message("user", ["What's the weather in Seattle?"])])
+    print(f"  Agent said: {response.text[:80]}...")
+
+    # Then evaluate the response (without re-running the agent)
+    quality_evals = FoundryEvals(
+        project_client=project_client,
+        model_deployment=deployment,
+        evaluators=[FoundryEvals.RELEVANCE, FoundryEvals.COHERENCE],
+    )
+    results = await evaluate_agent(
+        agent=agent,
+        responses=response,
+        queries=["What's the weather in Seattle?"],
+        evaluators=quality_evals,
+    )
+
+    for r in results:
+        print(f"\n  {r.provider}: {r.passed}/{r.total} passed")
+
+
+async def demo_foundry_workflow(project_client, deployment):
+    """Evaluate a multi-agent workflow with per-agent breakdown."""
+    print()
+    print("═" * 60)
+    print("  3c. Foundry — Multi-Agent Workflow")
+    print("═" * 60)
+
+    workflow = create_workflow(project_client, deployment)
+    evals = FoundryEvals(project_client=project_client, model_deployment=deployment)
+
+    # Run + evaluate with multiple queries
+    results = await evaluate_workflow(
+        workflow=workflow,
+        queries=["Plan a trip from Seattle to Paris"],
+        evaluators=evals,
+    )
+
+    print_workflow_results(results)
+
+
+async def demo_foundry_select(project_client, deployment):
+    """Choose specific Foundry evaluators."""
+    print()
+    print("═" * 60)
+    print("  3d. Foundry — Selecting Evaluators")
+    print("═" * 60)
+
+    agent = create_agent(project_client, deployment)
+
+    # Pick exactly which evaluators to run
+    evals = FoundryEvals(
+        project_client=project_client,
+        model_deployment=deployment,
+        evaluators=[
+            FoundryEvals.RELEVANCE,
+            FoundryEvals.TASK_ADHERENCE,
+            FoundryEvals.TOOL_CALL_ACCURACY,
+        ],
+    )
+    results = await evaluate_agent(
+        agent=agent,
+        queries=["What's the weather in Seattle?"],
+        evaluators=evals,
+    )
+
+    for r in results:
+        print(f"\n  {r.provider}: {r.passed}/{r.total} passed")
+        for ev_name, counts in r.per_evaluator.items():
+            print(f"    {ev_name}: {counts}")
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# Section 4: Mix Everything Together
+# ═════════════════════════════════════════════════════════════════════════════
+#
+# Pass a list of evaluators — local functions, built-in checks, and Foundry
+# all run together. You get one EvalResults per provider.
+#
+
+
+async def demo_mixed(project_client, deployment):
+    """Combine custom functions, built-in checks, and Foundry in one call."""
+    print()
+    print("═" * 60)
+    print("  4. Mixed Evaluation (recommended)")
+    print("═" * 60)
+
+    agent = create_agent(project_client, deployment)
+
+    # Local: custom functions + built-in checks
+    local = LocalEvaluator(
+        is_helpful,
+        no_apologies,
+        keyword_check("weather"),
+        tool_called_check("get_weather"),
+    )
+
+    # Cloud: Foundry AI quality assessment
+    foundry = FoundryEvals(project_client=project_client, model_deployment=deployment)
+
+    # One call, multiple providers
+    results = await evaluate_agent(
+        agent=agent,
+        queries=[
+            "What's the weather in Seattle?",
+            "How much is a flight from London to Paris?",
+        ],
+        evaluators=[local, foundry],
+    )
+
+    print()
+    for r in results:
+        status = "✓" if r.all_passed else "✗"
+        print(f"  {status} {r.provider}: {r.passed}/{r.total} passed")
+        for ev_name, counts in r.per_evaluator.items():
+            p, f = counts["passed"], counts["failed"]
+            print(f"      {ev_name}: {p}/{p + f}")
+        if r.report_url:
+            print(f"      Portal: {r.report_url}")
+
+    # CI assertion — fails the test if anything didn't pass
+    for r in results:
+        r.assert_passed()
+    print("\n  ✓ All evaluations passed!")
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# Section 5: Workflow + Mixed Evaluation
+# ═════════════════════════════════════════════════════════════════════════════
+
+
+async def demo_workflow_mixed(project_client, deployment):
+    """Evaluate a workflow with both local and Foundry evaluators."""
+    print()
+    print("═" * 60)
+    print("  5. Workflow + Mixed Evaluation")
+    print("═" * 60)
+
+    workflow = create_workflow(project_client, deployment)
+
+    local = LocalEvaluator(is_helpful, no_apologies)
+    foundry = FoundryEvals(project_client=project_client, model_deployment=deployment)
+
+    results = await evaluate_workflow(
+        workflow=workflow,
+        queries=["Plan a trip from Seattle to Paris"],
+        evaluators=[local, foundry],
+    )
+
+    print_workflow_results(results)
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# Section 6: Iterative Workflows (agents run multiple times)
+# ═════════════════════════════════════════════════════════════════════════════
+#
+# When an agent runs multiple times in a single workflow execution (e.g., in
+# a group chat or feedback loop), each invocation becomes a separate eval item.
+# Results are grouped by agent, so you see e.g. "writer: 3/3 passed".
+#
+
+
+def create_iterative_workflow(project_client, deployment):
+    """Create a group chat where a writer and reviewer iterate.
+
+    The writer drafts a response, the reviewer critiques it, and the
+    writer revises — running 2 rounds so each agent is invoked twice.
+    """
+    client = AzureOpenAIResponsesClient(
+        project_client=project_client,
+        deployment_name=deployment,
+    )
+    writer = Agent(
+        client=client,
+        name="writer",
+        instructions=(
+            "You are a travel copywriter. Write or revise a short, "
+            "compelling travel description based on the conversation."
+        ),
+        default_options={"store": False},
+    )
+    reviewer = Agent(
+        client=client,
+        name="reviewer",
+        instructions=("You are an editor. Critique the writer's draft and suggest specific improvements. Be concise."),
+        default_options={"store": False},
+    )
+
+    # Group chat with round-robin selection: writer → reviewer → writer → reviewer
+    # Each agent runs twice per query.
+    def round_robin(state):
+        names = list(state.participants.keys())
+        return names[state.current_round % len(names)]
+
+    return GroupChatBuilder(
+        participants=[writer, reviewer],
+        termination_condition=lambda conversation: len(conversation) >= 5,
+        selection_func=round_robin,
+    ).build()
+
+
+async def demo_iterative_workflow(project_client, deployment):
+    """Evaluate a workflow where agents run multiple times."""
+    print()
+    print("═" * 60)
+    print("  6. Iterative Workflow (multi-run agents)")
+    print("═" * 60)
+
+    workflow = create_iterative_workflow(project_client, deployment)
+
+    local = LocalEvaluator(is_helpful, no_apologies)
+
+    results = await evaluate_workflow(
+        workflow=workflow,
+        queries=["Write a travel description for Kyoto in autumn"],
+        evaluators=local,
+    )
+
+    print_workflow_results(results)
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# Run it
+# ═════════════════════════════════════════════════════════════════════════════
+
+
+async def main():
+    project_client = AIProjectClient(
+        endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
+        credential=DefaultAzureCredential(),
+    )
+    deployment = os.environ.get("AZURE_AI_MODEL_DEPLOYMENT_NAME", "gpt-4o")
+
+    # Run each section — comment out what you don't need
+    # await demo_evaluators(project_client, deployment)
+    # await demo_builtin_checks(project_client, deployment)
+    # await demo_foundry_agent(project_client, deployment)
+    # await demo_foundry_response(project_client, deployment)
+    # await demo_foundry_workflow(project_client, deployment)
+    # await demo_foundry_select(project_client, deployment)
+    # await demo_mixed(project_client, deployment)
+    await demo_workflow_mixed(project_client, deployment)
+    await demo_iterative_workflow(project_client, deployment)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/05-end-to-end/evaluation/foundry_evals/evaluate_mixed_sample.py

@@ -0,0 +1,166 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import os
+
+from agent_framework import (
+    Agent,
+    LocalEvaluator,
+    evaluate_agent,
+    keyword_check,
+    tool_called_check,
+)
+from agent_framework.azure import AzureOpenAIResponsesClient
+from agent_framework_azure_ai import FoundryEvals
+from azure.ai.projects.aio import AIProjectClient
+from azure.identity import DefaultAzureCredential
+from dotenv import load_dotenv
+
+load_dotenv()
+
+"""
+This sample demonstrates mixing local and cloud evaluation providers.
+
+It shows three patterns:
+1. Local-only: Fast, API-free checks for inner-loop development.
+2. Cloud-only: Full Foundry evaluators for comprehensive quality assessment.
+3. Mixed: Local + Foundry evaluators in a single evaluate_agent() call.
+
+Mixing lets you get instant local feedback (keyword presence, tool usage)
+alongside deeper cloud-based quality evaluation (relevance, coherence)
+in one call.
+
+Prerequisites:
+- An Azure AI Foundry project with a deployed model
+- Set AZURE_AI_PROJECT_ENDPOINT and AZURE_AI_MODEL_DEPLOYMENT_NAME in .env
+"""
+
+
+# Define a simple tool for the agent
+def get_weather(location: str) -> str:
+    """Get the current weather for a location."""
+    weather_data = {
+        "seattle": "62°F, cloudy with a chance of rain",
+        "london": "55°F, overcast",
+        "paris": "68°F, partly sunny",
+    }
+    return weather_data.get(location.lower(), f"Weather data not available for {location}")
+
+
+async def main():
+    # 1. Set up the Azure AI project client
+    project_client = AIProjectClient(
+        endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
+        credential=DefaultAzureCredential(),
+    )
+
+    deployment = os.environ.get("AZURE_AI_MODEL_DEPLOYMENT_NAME", "gpt-4o")
+
+    # 2. Create an agent with a tool
+    agent = Agent(
+        client=AzureOpenAIResponsesClient(
+            project_client=project_client,
+            deployment_name=deployment,
+        ),
+        name="weather-assistant",
+        instructions="You are a helpful weather assistant. Use the get_weather tool to answer questions.",
+        tools=[get_weather],
+    )
+
+    # =========================================================================
+    # Pattern 1: Local evaluation only (no API calls, instant results)
+    # =========================================================================
+    print("=" * 60)
+    print("Pattern 1: Local evaluation only")
+    print("=" * 60)
+
+    local = LocalEvaluator(
+        keyword_check("weather", "seattle"),
+        tool_called_check("get_weather"),
+    )
+
+    results = await evaluate_agent(
+        agent=agent,
+        queries=["What's the weather in Seattle?"],
+        evaluators=local,
+    )
+
+    for r in results:
+        print(f"Status: {r.status}")
+        print(f"Results: {r.passed}/{r.total} passed")
+        for check_name, counts in r.per_evaluator.items():
+            print(f"  {check_name}: {counts['passed']} passed, {counts['failed']} failed")
+        if r.all_passed:
+            print("✓ All local checks passed!")
+        else:
+            print(f"✗ Failures: {r.error}")
+
+    # =========================================================================
+    # Pattern 2: Foundry evaluation only (cloud-based quality assessment)
+    # =========================================================================
+    print()
+    print("=" * 60)
+    print("Pattern 2: Foundry evaluation only")
+    print("=" * 60)
+
+    foundry = FoundryEvals(project_client=project_client, model_deployment=deployment)
+
+    results = await evaluate_agent(
+        agent=agent,
+        queries=["What's the weather in Seattle?"],
+        evaluators=foundry,
+    )
+
+    for r in results:
+        print(f"Status: {r.status}")
+        print(f"Results: {r.passed}/{r.total} passed")
+        print(f"Portal: {r.report_url}")
+        if r.all_passed:
+            print("✓ All passed")
+        else:
+            print(f"✗ {r.failed} failed, {r.errored} errored")
+
+    # =========================================================================
+    # Pattern 3: Mixed — local + Foundry in one call
+    # =========================================================================
+    print()
+    print("=" * 60)
+    print("Pattern 3: Mixed local + Foundry evaluation")
+    print("=" * 60)
+
+    # Local checks: fast smoke tests
+    local = LocalEvaluator(
+        keyword_check("weather"),
+        tool_called_check("get_weather"),
+    )
+
+    # Foundry: deep quality assessment
+    foundry = FoundryEvals(project_client=project_client, model_deployment=deployment)
+
+    # Pass both as a list — returns one EvalResults per provider
+    results = await evaluate_agent(
+        agent=agent,
+        queries=[
+            "What's the weather in Seattle?",
+            "Tell me the weather in London",
+        ],
+        evaluators=[local, foundry],
+    )
+
+    for r in results:
+        status = "✓" if r.all_passed else "✗"
+        print(f"  {status} {r.provider}: {r.passed}/{r.total} passed")
+        for check_name, counts in r.per_evaluator.items():
+            print(f"      {check_name}: {counts['passed']}/{counts['passed'] + counts['failed']}")
+        if r.report_url:
+            print(f"      Portal: {r.report_url}")
+
+    if all(r.all_passed for r in results):
+        print("✓ All checks passed (local + Foundry)!")
+    else:
+        failed = [r.provider for r in results if not r.all_passed]
+        print(f"✗ Failed providers: {', '.join(failed)}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/05-end-to-end/evaluation/foundry_evals/evaluate_multiturn_sample.py

@@ -0,0 +1,191 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import os
+
+from agent_framework import ConversationSplit, EvalItem
+from agent_framework_azure_ai import FoundryEvals
+from azure.ai.projects.aio import AIProjectClient
+from azure.identity import DefaultAzureCredential
+from dotenv import load_dotenv
+
+load_dotenv()
+
+"""
+This sample demonstrates how conversation split strategies affect evaluation.
+
+The same multi-turn conversation can be split different ways, each evaluating
+a different aspect of agent behavior:
+
+1. LAST_TURN (default) — "Was the last response good given context?"
+2. FULL — "Did the whole conversation serve the original request?"
+3. per_turn_items — "Was each individual response appropriate?"
+
+Prerequisites:
+- An Azure AI Foundry project with a deployed model
+- Set AZURE_AI_PROJECT_ENDPOINT and AZURE_AI_MODEL_DEPLOYMENT_NAME in .env
+"""
+
+# A multi-turn conversation with tool calls that we'll evaluate three ways.
+CONVERSATION = [
+    # Turn 1: user asks about weather → agent calls tool → responds
+    {"role": "user", "content": "What's the weather in Seattle?"},
+    {
+        "role": "assistant",
+        "content": [
+            {"type": "tool_call", "tool_call_id": "c1", "name": "get_weather", "arguments": {"location": "seattle"}}
+        ],
+    },
+    {
+        "role": "tool",
+        "tool_call_id": "c1",
+        "content": [{"type": "tool_result", "tool_result": "62°F, cloudy with a chance of rain"}],
+    },
+    {"role": "assistant", "content": "Seattle is 62°F, cloudy with a chance of rain."},
+    # Turn 2: user asks about Paris → agent calls tool → responds
+    {"role": "user", "content": "And Paris?"},
+    {
+        "role": "assistant",
+        "content": [
+            {"type": "tool_call", "tool_call_id": "c2", "name": "get_weather", "arguments": {"location": "paris"}}
+        ],
+    },
+    {
+        "role": "tool",
+        "tool_call_id": "c2",
+        "content": [{"type": "tool_result", "tool_result": "68°F, partly sunny"}],
+    },
+    {"role": "assistant", "content": "Paris is 68°F, partly sunny."},
+    # Turn 3: user asks for comparison → agent synthesizes without tool
+    {"role": "user", "content": "Can you compare them?"},
+    {
+        "role": "assistant",
+        "content": "Seattle is cooler at 62°F with rain likely, while Paris is warmer at 68°F and partly sunny. Paris is the better choice for outdoor activities.",
+    },
+]
+
+TOOL_DEFINITIONS = [
+    {
+        "name": "get_weather",
+        "description": "Get the current weather for a location.",
+        "parameters": {"type": "object", "properties": {"location": {"type": "string"}}},
+    },
+]
+
+
+def print_split(item: EvalItem, split: ConversationSplit = ConversationSplit.LAST_TURN):
+    """Print the query/response split for an EvalItem."""
+    d = item.to_eval_data(split=split)
+    print(f"  query_messages ({len(d['query_messages'])}):")
+    for m in d["query_messages"]:
+        content = m.get("content", "")
+        if isinstance(content, list):
+            content = content[0].get("type", str(content[0]))
+        print(f"    {m['role']}: {str(content)[:70]}")
+    print(f"  response_messages ({len(d['response_messages'])}):")
+    for m in d["response_messages"]:
+        content = m.get("content", "")
+        if isinstance(content, list):
+            content = content[0].get("type", str(content[0]))
+        print(f"    {m['role']}: {str(content)[:70]}")
+
+
+async def main():
+    project_client = AIProjectClient(
+        endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
+        credential=DefaultAzureCredential(),
+    )
+    deployment = os.environ.get("AZURE_AI_MODEL_DEPLOYMENT_NAME", "gpt-4o")
+
+    # =========================================================================
+    # Strategy 1: LAST_TURN (default)
+    # "Given all context, was the last response good?"
+    # =========================================================================
+    print("=" * 70)
+    print("Strategy 1: LAST_TURN — evaluate the final response")
+    print("=" * 70)
+
+    item = EvalItem(
+        query="Can you compare them?",
+        response="Seattle is cooler at 62°F with rain likely, while Paris is warmer at 68°F and partly sunny. Paris is the better choice for outdoor activities.",
+        conversation=CONVERSATION,
+        tool_definitions=TOOL_DEFINITIONS,
+    )
+
+    print_split(item, ConversationSplit.LAST_TURN)
+
+    results = await FoundryEvals(
+        project_client=project_client,
+        model_deployment=deployment,
+        evaluators=[FoundryEvals.RELEVANCE, FoundryEvals.COHERENCE],
+        # conversation_split defaults to LAST_TURN
+    ).evaluate([item], eval_name="Split Strategy: LAST_TURN")
+
+    print(f"\n  Result: {results.passed}/{results.total} passed")
+    print(f"  Portal: {results.report_url}")
+    for ir in results.items:
+        for s in ir.scores:
+            print(f"    {'✓' if s.passed else '✗'} {s.name}: {s.score}")
+    print()
+
+    # =========================================================================
+    # Strategy 2: FULL
+    # "Given the original request, did the whole conversation serve the user?"
+    # =========================================================================
+    print("=" * 70)
+    print("Strategy 2: FULL — evaluate the entire conversation trajectory")
+    print("=" * 70)
+
+    print_split(item, ConversationSplit.FULL)
+
+    results = await FoundryEvals(
+        project_client=project_client,
+        model_deployment=deployment,
+        evaluators=[FoundryEvals.RELEVANCE, FoundryEvals.COHERENCE],
+        conversation_split=ConversationSplit.FULL,
+    ).evaluate([item], eval_name="Split Strategy: FULL")
+
+    print(f"\n  Result: {results.passed}/{results.total} passed")
+    print(f"  Portal: {results.report_url}")
+    for ir in results.items:
+        for s in ir.scores:
+            print(f"    {'✓' if s.passed else '✗'} {s.name}: {s.score}")
+    print()
+
+    # =========================================================================
+    # Strategy 3: per_turn_items
+    # "Was each individual response appropriate at that point?"
+    # =========================================================================
+    print("=" * 70)
+    print("Strategy 3: per_turn_items — evaluate each turn independently")
+    print("=" * 70)
+
+    items = EvalItem.per_turn_items(
+        CONVERSATION,
+        tool_definitions=TOOL_DEFINITIONS,
+    )
+    print(f"  Split into {len(items)} items from {len(CONVERSATION)} messages:\n")
+    for i, it in enumerate(items):
+        print(f"  Turn {i + 1}: query={it.query!r}, response={it.response[:60]!r}...")
+    print()
+
+    results = await FoundryEvals(
+        project_client=project_client,
+        model_deployment=deployment,
+        evaluators=[FoundryEvals.RELEVANCE, FoundryEvals.COHERENCE],
+    ).evaluate(items, eval_name="Split Strategy: Per-Turn")
+
+    print(f"\n  Result: {results.passed}/{results.total} passed ({len(items)} items × 2 evaluators)")
+    print(f"  Portal: {results.report_url}")
+    for ir in results.items:
+        for s in ir.scores:
+            print(f"    {'✓' if s.passed else '✗'} {s.name}: {s.score}")
+    print()
+
+    print("=" * 70)
+    print("All strategies complete. Compare results in the Foundry portal.")
+    print("=" * 70)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/05-end-to-end/evaluation/foundry_evals/evaluate_traces_sample.py

@@ -0,0 +1,121 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import os
+
+from agent_framework_azure_ai import FoundryEvals, evaluate_traces
+from azure.ai.projects.aio import AIProjectClient
+from azure.identity import DefaultAzureCredential
+from dotenv import load_dotenv
+
+load_dotenv()
+
+"""
+This sample demonstrates evaluating agent responses that already exist in Foundry.
+
+It shows two patterns:
+1. evaluate_traces(response_ids=...) — Evaluate specific Responses API responses by ID.
+2. evaluate_traces(agent_id=...) — Evaluate agent behavior from OTel traces in App Insights.
+
+These are the "zero-code-change" evaluation paths — the agent has already run,
+and you're evaluating what happened after the fact.
+
+Prerequisites:
+- An Azure AI Foundry project with a deployed model
+- Response IDs from prior agent runs (for Pattern 1)
+- OTel traces exported to App Insights (for Pattern 2)
+- Set AZURE_AI_PROJECT_ENDPOINT and AZURE_AI_MODEL_DEPLOYMENT_NAME in .env
+"""
+
+
+async def main():
+    # 1. Set up the Azure AI project client
+    project_client = AIProjectClient(
+        endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
+        credential=DefaultAzureCredential(),
+    )
+
+    deployment = os.environ.get("AZURE_AI_MODEL_DEPLOYMENT_NAME", "gpt-4o")
+
+    # =========================================================================
+    # Pattern 1: evaluate_traces(response_ids=...) — By response ID
+    # =========================================================================
+    # If your agent uses the Responses API (e.g., AzureOpenAIResponsesClient),
+    # each run produces a response_id. Pass those IDs to evaluate_traces()
+    # and Foundry retrieves the full conversation for evaluation.
+    print("=" * 60)
+    print("Pattern 1: evaluate_traces(response_ids=...)")
+    print("=" * 60)
+
+    # Replace these with actual response IDs from your agent runs
+    response_ids = [
+        "resp_abc123",
+        "resp_def456",
+    ]
+
+    results = await evaluate_traces(
+        response_ids=response_ids,
+        evaluators=[FoundryEvals.RELEVANCE, FoundryEvals.GROUNDEDNESS, FoundryEvals.TOOL_CALL_ACCURACY],
+        project_client=project_client,
+        model_deployment=deployment,
+    )
+
+    print(f"Status: {results.status}")
+    print(f"Results: {results.result_counts}")
+    print(f"Portal: {results.report_url}")
+
+    # =========================================================================
+    # Pattern 2: evaluate_traces(agent_id=...) — From App Insights
+    # =========================================================================
+    # If your agent emits OTel traces to App Insights (via configure_otel_providers),
+    # you can evaluate recent activity without specifying individual response IDs.
+    #
+    # NOTE: Requires OTel traces exported to the App Insights instance connected
+    # to your Foundry project. The exact trace-based data source API is subject
+    # to change as Foundry evolves.
+    print()
+    print("=" * 60)
+    print("Pattern 2: evaluate_traces(agent_id=...)")
+    print("=" * 60)
+
+    # Evaluate by response IDs (uses response-based data source internally)
+    results = await evaluate_traces(
+        response_ids=response_ids,
+        evaluators=[FoundryEvals.RELEVANCE, FoundryEvals.COHERENCE],
+        project_client=project_client,
+        model_deployment=deployment,
+    )
+
+    print(f"Status: {results.status}")
+    print(f"Portal: {results.report_url}")
+
+    # Evaluate by agent ID + time window (when trace-based API is available)
+    # results = await evaluate_traces(
+    #     agent_id="travel-bot",
+    #     evaluators=[FoundryEvals.INTENT_RESOLUTION, FoundryEvals.TASK_ADHERENCE],
+    #     project_client=project_client,
+    #     model_deployment=deployment,
+    #     lookback_hours=24,
+    # )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
+
+"""
+Sample output (with actual Azure AI Foundry project and valid response IDs):
+
+============================================================
+Pattern 1: evaluate_traces(response_ids=...)
+============================================================
+Status: completed
+Results: {'passed': 2, 'failed': 0, 'errored': 0}
+Portal: https://ai.azure.com/...
+
+============================================================
+Pattern 2: evaluate_traces(agent_id=...)
+============================================================
+Status: completed
+Portal: https://ai.azure.com/...
+"""

python/samples/05-end-to-end/evaluation/foundry_evals/evaluate_workflow_sample.py

@@ -0,0 +1,182 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import os
+
+from agent_framework import Agent, evaluate_workflow
+from agent_framework.azure import AzureOpenAIResponsesClient
+from agent_framework_azure_ai import FoundryEvals
+from agent_framework_orchestrations import SequentialBuilder
+from azure.ai.projects.aio import AIProjectClient
+from azure.identity import DefaultAzureCredential
+from dotenv import load_dotenv
+
+load_dotenv()
+
+"""
+This sample demonstrates evaluating a multi-agent workflow using Azure AI Foundry evaluators.
+
+It shows two patterns:
+1. Post-hoc: Run the workflow, then evaluate the result you already have.
+2. Run + evaluate: Pass queries and let evaluate_workflow() run the workflow for you.
+
+Both patterns return a list of results (one per provider), each with a per-agent
+breakdown in sub_results so you can identify which agent is underperforming.
+
+Prerequisites:
+- An Azure AI Foundry project with a deployed model
+- Set AZURE_AI_PROJECT_ENDPOINT and AZURE_AI_MODEL_DEPLOYMENT_NAME in .env
+"""
+
+
+# Simple tools for the agents
+def get_weather(location: str) -> str:
+    """Get the current weather for a location."""
+    weather_data = {
+        "seattle": "62°F, cloudy with a chance of rain",
+        "london": "55°F, overcast",
+        "paris": "68°F, partly sunny",
+    }
+    return weather_data.get(location.lower(), f"Weather data not available for {location}")
+
+
+def get_flight_price(origin: str, destination: str) -> str:
+    """Get the price of a flight between two cities."""
+    return f"Flights from {origin} to {destination}: $450 round-trip"
+
+
+async def main():
+    # 1. Set up the Azure AI project client
+    project_client = AIProjectClient(
+        endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
+        credential=DefaultAzureCredential(),
+    )
+
+    deployment = os.environ.get("AZURE_AI_MODEL_DEPLOYMENT_NAME", "gpt-4o")
+
+    client = AzureOpenAIResponsesClient(
+        project_client=project_client,
+        deployment_name=deployment,
+    )
+
+    # 2. Create agents for a sequential workflow
+    # Use store=False so agents don't chain conversation state via previous_response_id.
+    # This allows the workflow to be run multiple times without stale state issues.
+    researcher = Agent(
+        client=client,
+        name="researcher",
+        instructions=(
+            "You are a travel researcher. Use your tools to gather weather "
+            "and flight information for the destination the user asks about."
+        ),
+        tools=[get_weather, get_flight_price],
+        default_options={"store": False},
+    )
+
+    planner = Agent(
+        client=client,
+        name="planner",
+        instructions=(
+            "You are a travel planner. Based on the research provided, "
+            "create a concise travel recommendation with packing tips."
+        ),
+        default_options={"store": False},
+    )
+
+    # 3. Build a sequential workflow: researcher → planner
+    workflow = SequentialBuilder(participants=[researcher, planner]).build()
+
+    # 4. Create the evaluator — provider config goes here, once
+    evals = FoundryEvals(project_client=project_client, model_deployment=deployment)
+
+    # =========================================================================
+    # Pattern 1: Post-hoc — evaluate a workflow run you already did
+    # =========================================================================
+    print("=" * 60)
+    print("Pattern 1: Post-hoc workflow evaluation")
+    print("=" * 60)
+
+    result = await workflow.run("Plan a trip from Seattle to Paris")
+
+    eval_results = await evaluate_workflow(
+        workflow=workflow,
+        workflow_result=result,
+        evaluators=evals,
+    )
+
+    for r in eval_results:
+        print(f"\nOverall: {r.status}")
+        print(f"  Passed: {r.passed}/{r.total}")
+        print(f"  Portal: {r.report_url}")
+
+        print("\nPer-agent breakdown:")
+        for agent_name, agent_eval in r.sub_results.items():
+            print(f"  {agent_name}: {agent_eval.passed}/{agent_eval.total} passed")
+            if agent_eval.report_url:
+                print(f"    Portal: {agent_eval.report_url}")
+
+    # =========================================================================
+    # Pattern 2: Run + evaluate with multiple queries
+    # =========================================================================
+    # Build a fresh workflow to avoid stale session state from Pattern 1.
+    # The Responses API tracks previous_response_id per session, so reusing
+    # a workflow after a run would reference stale tool calls.
+    workflow2 = SequentialBuilder(participants=[researcher, planner]).build()
+
+    print()
+    print("=" * 60)
+    print("Pattern 2: Run + evaluate with multiple queries")
+    print("=" * 60)
+
+    eval_results = await evaluate_workflow(
+        workflow=workflow2,
+        queries=[
+            "Plan a trip from London to Tokyo",
+            "Plan a trip from New York to Rome",
+        ],
+        evaluators=evals.select(FoundryEvals.RELEVANCE, FoundryEvals.TASK_ADHERENCE),
+    )
+
+    for r in eval_results:
+        print(f"\nOverall: {r.status}")
+        print(f"  Passed: {r.passed}/{r.total}")
+        if r.report_url:
+            print(f"  Portal: {r.report_url}")
+
+        print("\nPer-agent breakdown:")
+        for agent_name, agent_eval in r.sub_results.items():
+            print(f"  {agent_name}: {agent_eval.passed}/{agent_eval.total} passed")
+            if agent_eval.report_url:
+                print(f"    Portal: {agent_eval.report_url}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
+
+"""
+Sample output (with actual Azure AI Foundry project):
+
+============================================================
+Pattern 1: Post-hoc workflow evaluation
+============================================================
+
+Overall: completed
+  Passed: 2/2
+  Portal: https://ai.azure.com/...
+
+Per-agent breakdown:
+  researcher: 1/1 passed
+  planner: 1/1 passed
+
+============================================================
+Pattern 2: Run + evaluate with multiple queries
+============================================================
+
+Overall: completed
+  Passed: 4/4
+
+Per-agent breakdown:
+  researcher: 2/2 passed
+  planner: 2/2 passed
+"""

python/samples/05-end-to-end/hosted_agents/agent_with_local_tools/main.py

@@ -16,7 +16,6 @@ from azure.ai.agentserver.agentframework import from_agent_framework
 from azure.identity.aio import AzureCliCredential, ManagedIdentityCredential
 from dotenv import load_dotenv
 
-
 load_dotenv(override=True)
 
 # Configure these for your Foundry project