Part 11: DIY runner for the FAQ agent

The cancellation demo used mock functions. This step packages the pattern into a reusable runner and connects it back to the real FAQ agent and topic guardrail.

Build the generic runner

The runner accepts the agent coroutine and a list of guardrail coroutines:

async def run_with_guardrails(agent_coro, guardrails):
    """
    Run an agent with guardrails.

    Args:
        agent_coro: The agent coroutine to run
        guardrails: List of async guardrail functions

    Returns:
        The agent's result if all guardrails pass

    Raises:
        GuardrailException: If any guardrail trips
    """
    agent_task = asyncio.create_task(agent_coro)
    guardrail_tasks = [asyncio.create_task(g) for g in guardrails]

Then it waits for all tasks. If nothing raises, the agent result is returned:

    try:
        await asyncio.gather(agent_task, *guardrail_tasks)
        return agent_task.result()

If a guardrail raises, the runner cancels the agent and the remaining guardrails:

    except GuardrailException as e:
        print(f"[Guardrail tripped] {e.result.reasoning}")

        agent_task.cancel()
        try:
            await agent_task
        except asyncio.CancelledError:
            print("[Agent cancelled - saved tokens]")

        for t in guardrail_tasks:
            t.cancel()
        await asyncio.gather(*guardrail_tasks, return_exceptions=True)

        raise

Keep the final raise in reusable code because the caller can decide how to display the blocked result.

If one guardrail fails while other guardrails are still running, cancel the remaining guardrail tasks too. There is no reason to keep checking policies after one policy has already stopped the run.

Execution flow

The runner does four things:

1. It creates one task for the agent and one task per guardrail.
2. It waits with asyncio.gather.
3. It returns the agent result when every task finishes without an exception.
4. It cancels unfinished work and re-raises when a guardrail trips.

This works across frameworks because the runner only needs coroutines and a shared exception type. The agent can come from the OpenAI Agents SDK, PydanticAI, LangChain, or custom code.

Convert the topic guardrail

The SDK input guardrail returns GuardrailFunctionOutput. The DIY runner expects a coroutine that raises GuardrailException, so define a separate function:

async def topic_guardrail_check(input):
    """Check if the user's question is about the course."""
    result = await Runner.run(topic_guardrail_agent, input)
    output = result.final_output

    if output.fail:
        raise GuardrailException(GuardrailResult(
            reasoning=output.reasoning,
            triggered=True
        ))

I use topic_guardrail_check instead of topic_guardrail so we do not overwrite the SDK-decorated guardrail from the earlier step.

Run a course question

Run the unguarded FAQ agent and the topic check in parallel:

prompt = "I just discovered the course, can I still join?"
result = await run_with_guardrails(
    Runner.run(faq_agent, prompt),
    [
        topic_guardrail_check(prompt),
    ]
)

The result is the normal RunResult from the FAQ agent:

print(result.final_output)

A typical RunResult includes a search_faq tool call for "join course", followed by the tool output and final answer.

Run an off-topic question

Now use the same runner with an unrelated prompt:

prompt = "How can I cook pizza?"

try:
    await run_with_guardrails(
        Runner.run(faq_agent, prompt),
        [
            topic_guardrail_check(prompt),
        ]
    )
except GuardrailException as e:
    print(f"[BLOCKED] {e.result.reasoning}")

The topic guardrail raises, run_with_guardrails cancels the FAQ agent, and the caller prints the reason. That gives you the same user-facing shape as the SDK input guardrail, but it runs as plain async Python.

Continue with Summary and next steps.