Python: [Feature]: Support async conditions in switch-case edge groups #9642

Problem

A normal edge condition can be async. A switch-case Case condition can't. If you pass an async function to Case, it is called without await. It returns a coroutine object, which is always truthy, so the first case always matches. The only sign is a RuntimeWarning: coroutine '...' was never awaited.

What happens today (agent-framework-core 1.23.0, Python):

• _workflows/_edge.py:21: EdgeCondition: TypeAlias = Callable[[Any], bool | Awaitable[bool]]. Normal edges accept async.
• _workflows/_edge.py:195-198: Edge.should_route awaits the result when it is awaitable.
• _workflows/_edge.py:267 and :663: Case.condition and SwitchCaseEdgeGroupCase are typed Callable[[Any], bool].
• _workflows/_edge.py:862-871: the switch-case selection_func is sync and calls if case.condition(message): with no await.
• _workflows/_edge_runner.py:282: FanOutEdgeRunner calls self._selection_func(...) with no await.

So the same function works on add_edge(condition=...) but routes to the wrong branch when used in a `Case

Use Case and Evidence

Routing often needs an async call: an LLM classifier, a database lookup, or an API check. Switch-case is the natural fit for "first match wins, else default", but today it can only use sync conditions.

Minimal repro. Input 1 should go to small (the default), but goes to big:

import asyncio
from typing import Never
from agent_framework import Case, Default, Executor, WorkflowBuilder, WorkflowContext, handler


class Start(Executor):
    @handler
    async def run(self, n: int, ctx: WorkflowContext[int]) -> None:
        await ctx.send_message(n)


class Leaf(Executor):
    @handler
    async def run(self, n: int, ctx: WorkflowContext[Never, str]) -> None:
        await ctx.yield_output(f"{self.id} got {n}")


async def is_big(n: int) -> bool:
    return n > 10


async def main() -> None:
    start, big, small = Start(id="start"), Leaf(id="big"), Leaf(id="small")
    workflow = (
        WorkflowBuilder(start_executor=start)
        .add_switch_case_edge_group(start, [Case(condition=is_big, target=big), Default(target=small)])
        .build()
    )
    print((await workflow.run(1)).get_outputs())


asyncio.run(main())

Output:

_edge.py:870: RuntimeWarning: coroutine 'is_big' was never awaited
  if case.condition(message):
['big got 1']

Desired Outcome

• Case(condition=...) accepts a function that returns bool or Awaitable[bool], the same as add_edge(condition=...).
• Cases are still checked in order, and the first match wins.
• Existing sync conditions keep working.

Alternatives Considered

• Do the async work in the executor before the switch, put the result on the message, and use a sync condition that reads it. This works, but it changes the message type just for routing.
• Use several add_edge(condition=async_fn) edges instead of switch-case. This loses "first match wins" and the default branch, so the conditions must be written to never overlap.
• Call ctx.send_message(msg, target_id=...) from the executor. This moves routing out of the graph.

Proposed Direction

Make the switch-case selection_func async and await each condition when its result is awaitable, as Edge.should_route already does:

async def selection_func(message: Any, targets: list[str]) -> list[str]:
    for case in cases:
        if isinstance(case, SwitchCaseEdgeGroupDefault):
            return [case.target_id]
        result = case.condition(message)
        if inspect.isawaitable(result):
            result = await result
        if result:
            return [case.target_id]
    raise RuntimeError("No matching case found in SwitchCaseEdgeGroup")

FanOutEdgeRunner.send_message would then await self._selection_func(...) when its result is awaitable. Update the type of Case.condition and SwitchCaseEdgeGroupCase to EdgeCondition.

Scope and Non-Goals

• In scope: Python, switch-case Case conditions.
• Not in scope: other changes to switch-case behaviour.

Language

Python

Contribution Intent

I am willing to implement this

Acknowledgements

• I searched existing issues and did not find a duplicate.
• I will wait for explicit maintainer agreement before starting implementation of a non-trivial change.