Evaluate Agents SDK agents in a workflow

If the system you evaluate is an agent built with the Agents SDK, pass it as an AgentSystem, like in the Evaluation SDK guide. The task runs the agent as ctx.agent, and the run records which agent was evaluated, so Studio can show it.

The one difference with the SDK: an Agent is a Python object, and it can't cross Temporal. The workflow only sends its name and overrides, and the task's activity rebuilds ctx.agent from your code. So you declare an agent from code on the task that runs it.

You need the Workflow evaluation plugin and the Agents SDK, mistralai-agents 1.1.0 or later, in the worker's environment:

pip install -U mistralai-agents

pip install -U mistralai-agents

A worker without an API key (a managed deployment) reaches the Agents API with its service-account token. This also needs mistralai-agents 1.1.0 or later.

Declare the agent on the task with @evaluation.task(agents=[...]), then pass it in the AgentSystem of evaluation.run():

from mistralai import agents
from mistralai.workflows import workflow
from mistralai.workflows.plugins.evaluations import AgentSystem, evaluation
from mistralai.workflows.plugins.evaluations.types import Evaluation, Project, TaskContext


@agents.tool
async def lookup_order(order_id: str) -> dict[str, str]:
    """Get the status of an order."""
    return {"order_id": order_id, "status": "shipped on Monday"}  # read from your database


support_bot = agents.Agent(
    name="support-bot",
    model="mistral-small-latest",
    instructions="You are the support agent of Shopi.",
    tools=[lookup_order],
)


@evaluation.task(agents=[support_bot])
async def answer(ctx: TaskContext) -> str:
    result = await ctx.agent.run(ctx.input_record["question"])
    return result.text or ""


@workflow.define(name="support-bot-eval")
class SupportBotEval:
    @workflow.entrypoint
    async def run(self) -> None:
        await evaluation.run(
            system=AgentSystem(agent=support_bot, overrides={"model": "mistral-medium-latest"}),
            task=answer,
            dataset=dataset,
            evaluators=evaluators,
            project=Project(name="Support"),
            evaluation=Evaluation(name="Support bot"),
        )

from mistralai import agents
from mistralai.workflows import workflow
from mistralai.workflows.plugins.evaluations import AgentSystem, evaluation
from mistralai.workflows.plugins.evaluations.types import Evaluation, Project, TaskContext


@agents.tool
async def lookup_order(order_id: str) -> dict[str, str]:
    """Get the status of an order."""
    return {"order_id": order_id, "status": "shipped on Monday"}  # read from your database


support_bot = agents.Agent(
    name="support-bot",
    model="mistral-small-latest",
    instructions="You are the support agent of Shopi.",
    tools=[lookup_order],
)


@evaluation.task(agents=[support_bot])
async def answer(ctx: TaskContext) -> str:
    result = await ctx.agent.run(ctx.input_record["question"])
    return result.text or ""


@workflow.define(name="support-bot-eval")
class SupportBotEval:
    @workflow.entrypoint
    async def run(self) -> None:
        await evaluation.run(
            system=AgentSystem(agent=support_bot, overrides={"model": "mistral-medium-latest"}),
            task=answer,
            dataset=dataset,
            evaluators=evaluators,
            project=Project(name="Support"),
            evaluation=Evaluation(name="Support bot"),
        )

dataset and evaluators are the ones you build for any run: see the plugin quickstart.

ctx.agent is the agent with the run's overrides. It runs in the task's activity, on the worker, with its real tools and hooks. overrides, params, and what can be overridden work as in the SDK guide.

ctx.agent.run() takes what Agent.run() takes and returns the Completed turn. A turn that does not complete raises AgentTurnError, so its record fails and is not scored. See Turns that don't complete.

In a workflow defined with on_behalf_of=True, only a deployed agent is supported: it is looked up and run as the executing user. evaluation.run() and evaluation.optimize() refuse an agent from code there, because its model calls, connectors, and MCP servers would use the worker's credentials, not the user's. See On-behalf-of workflows.

A task can run several agents. Declare them all, and the AgentSystem of each run picks one. Their names must differ, because the worker finds the agent by name.

@evaluation.task(agents=[support_bot, support_bot_v2])
async def answer(ctx: TaskContext) -> str: ...


for agent in (support_bot, support_bot_v2):
    await evaluation.run(system=AgentSystem(agent=agent), task=answer, ...)

@evaluation.task(agents=[support_bot, support_bot_v2])
async def answer(ctx: TaskContext) -> str: ...


for agent in (support_bot, support_bot_v2):
    await evaluation.run(system=AgentSystem(agent=agent), task=answer, ...)

Variants of one agent (another model, other instructions) need no new declaration: they are overrides.

Pass the name or id of a deployed agent. It needs no declaration: the task's activity loads the deployment from the Agents API of the server the workflow evaluates on, with the evaluation's credentials. These are the worker's API key or service-account token, or the executing user's identity in an on_behalf_of workflow.

@evaluation.task
async def answer(ctx: TaskContext) -> str:
    result = await ctx.agent.run(ctx.input_record["question"])
    return result.text or ""


await evaluation.run(
    system=AgentSystem(agent="support-bot", overrides={"additional_instructions": "Answer in French."}),
    task=answer,
    ...
)

@evaluation.task
async def answer(ctx: TaskContext) -> str:
    result = await ctx.agent.run(ctx.input_record["question"])
    return result.text or ""


await evaluation.run(
    system=AgentSystem(agent="support-bot", overrides={"additional_instructions": "Answer in French."}),
    task=answer,
    ...
)

The worker needs the Agents SDK installed, and the run must be saved in Studio (not local=True). For what a deployed agent lets you override, see the SDK guide.

evaluation.optimize() takes an AgentSystem too. Wrap the overrides to search in Tunable. Each candidate is a run of the agent with the candidate's values. See Optimization.

from mistralai.workflows.plugins.evaluations import GEPA
from mistralai.workflows.plugins.evaluations.types import Tunable

result = await evaluation.optimize(
    system=AgentSystem(agent=support_bot, overrides={"instructions": Tunable(support_bot.instructions)}),
    task=answer,
    dataset=dataset,
    evaluators=evaluators,
    algo=GEPA(),
)

from mistralai.workflows.plugins.evaluations import GEPA
from mistralai.workflows.plugins.evaluations.types import Tunable

result = await evaluation.optimize(
    system=AgentSystem(agent=support_bot, overrides={"instructions": Tunable(support_bot.instructions)}),
    task=answer,
    dataset=dataset,
    evaluators=evaluators,
    algo=GEPA(),
)

The run is refused before it is created when its task can't run the agent. The task must be:

• an @evaluation.task, not a workflow, because ctx.agent.run() does I/O, which a workflow can't do;
• async, with a single ctx: TaskContext argument (annotated);
• for an agent from code, declared with @evaluation.task(agents=[...]).

• Evaluate Agents SDK agents: the SDK guide, with overrides, params, and what a run records.
• Task modes: the kinds of task that evaluation.run() accepts.
• Optimization: run the optimizer as a workflow.