Implementation:Microsoft Agent framework Edge Condition

Overview

The Edge Condition implementation defines the EdgeCondition type alias and the Edge dataclass that together provide conditional routing capabilities within the Microsoft Agent Framework's workflow engine. An EdgeCondition is a callable that receives the source executor's output data and returns a boolean (synchronous or asynchronous) to determine whether a given edge should be traversed at runtime. The Edge dataclass wraps this condition alongside source and target executor identifiers, providing serialisation, deserialisation, and runtime evaluation through its should_route method.

Code Reference

Source Location

Property	Value
File	`python/packages/core/agent_framework/_workflows/_edge.py`
Line Range	L20-240
Primary Type	`EdgeCondition` (type alias, line 20)
Primary Class	`Edge` (dataclass, lines 75-240)

Import Statement

from agent_framework import WorkflowBuilder

Edge conditions are passed to the builder's add_edge method rather than being imported directly. The EdgeCondition type alias is available for annotation purposes from the internal module:

from agent_framework._workflows._edge import EdgeCondition

Type Definition

EdgeCondition: TypeAlias = Callable[[Any], bool | Awaitable[bool]]

The type alias captures the full contract: a callable that accepts the source executor's output data (typed as Any) and returns either a plain bool or an Awaitable[bool]. The runtime transparently handles both synchronous and asynchronous conditions.

I/O Contract

EdgeCondition Callable

Parameter	Type	Description
`data`	`Any`	The output payload produced by the source executor. This is the message data that would be forwarded along the edge if the condition passes.

Return	Type	Description
result	Awaitable[bool]	`True` if the edge should be traversed (message forwarded to the target executor); `False` if the edge should be skipped. May be returned synchronously or as an awaitable.

Edge Constructor

Parameter	Type	Default	Description
`source_id`	`str`	(required)	Canonical identifier of the upstream executor instance.
`target_id`	`str`	(required)	Canonical identifier of the downstream executor instance.
`condition`	None	`None`	Optional predicate that receives the message data and returns `True` when the edge should be traversed. When omitted, the edge is unconditionally active.
`condition_name`	None	`None`	Optional human-friendly name for the condition. Auto-extracted from the callable's `__name__` when not provided.

Edge.should_route Method

Parameter	Type	Description
`data`	`Any`	The message payload to evaluate the condition against.

Return	Type	Description
result	`bool`	`True` if the edge should be traversed. Returns `True` unconditionally when no condition was set.

Builder Integration

Edge conditions are attached to edges through the WorkflowBuilder.add_edge method:

class WorkflowBuilder:
    def add_edge(
        self,
        source: Executor | SupportsAgentRun,
        target: Executor | SupportsAgentRun,
        condition: EdgeCondition | None = None,
    ) -> Self:
        ...

When condition is provided, the builder internally creates a SingleEdgeGroup containing an Edge with that condition. The runtime later calls edge.should_route(data) to decide whether to forward the message.

Usage Examples

Simple Boolean Routing

from agent_framework import WorkflowBuilder

def is_positive(data) -> bool:
    return data.agent_response.value.sentiment == "positive"

def is_negative(data) -> bool:
    return data.agent_response.value.sentiment == "negative"

builder = WorkflowBuilder(start_executor=classifier)
builder.add_edge(classifier, positive_handler, condition=is_positive)
builder.add_edge(classifier, negative_handler, condition=is_negative)

workflow = builder.build()
result = await workflow.run("Analyze this review")

Async Condition

from agent_framework import WorkflowBuilder

async def is_premium_customer(data) -> bool:
    customer_id = data.agent_response.value.customer_id
    return await customer_service.check_premium(customer_id)

builder = WorkflowBuilder(start_executor=intake)
builder.add_edge(intake, premium_handler, condition=is_premium_customer)
builder.add_edge(intake, standard_handler)  # unconditional fallback

workflow = builder.build()

Threshold-Based Guard

from agent_framework import WorkflowBuilder

def score_above_threshold(data) -> bool:
    return data.agent_response.value.confidence > 0.8

builder = WorkflowBuilder(start_executor=scorer)
builder.add_edge(scorer, auto_approve, condition=score_above_threshold)
builder.add_edge(scorer, manual_review, condition=lambda data: data.agent_response.value.confidence <= 0.8)

workflow = builder.build()

Switch-Case Pattern via SwitchCaseEdgeGroup

For first-match switch-case semantics (where exactly one branch fires), use the dedicated SwitchCaseEdgeGroup:

from agent_framework._workflows._edge import (
    SwitchCaseEdgeGroup,
    SwitchCaseEdgeGroupCase,
    SwitchCaseEdgeGroupDefault,
)

cases = [
    SwitchCaseEdgeGroupCase(lambda data: data["kind"] == "csv", target_id="csv_handler"),
    SwitchCaseEdgeGroupCase(lambda data: data["kind"] == "json", target_id="json_handler"),
    SwitchCaseEdgeGroupDefault(target_id="fallback_handler"),
]
group = SwitchCaseEdgeGroup("router", cases)

Serialisation

Edges support round-trip serialisation for workflow persistence:

Method	Behaviour
`Edge.to_dict()`	Serialises `source_id`, `target_id`, and `condition_name` into a JSON-friendly dictionary. The live callable is not persisted.
`Edge.from_dict(data)`	Reconstructs an `Edge` from a dictionary payload. The deserialised edge lacks the executable predicate; the stored `condition_name` is preserved for diagnostics and re-registration.

When a deserialised edge is invoked without re-registering its condition callable, the runtime raises a RuntimeError identifying the missing callable by name.

Edge Group Hierarchy

The Edge class participates in a hierarchy of edge groups that implement higher-order routing patterns:

Class	Pattern	Condition Support
SingleEdgeGroup	One-to-one routing	Optional `EdgeCondition` on the single edge
FanOutEdgeGroup	One-to-many broadcast	Selection function narrows targets at runtime
FanInEdgeGroup	Many-to-one convergence	No conditions (all sources feed the same target)
SwitchCaseEdgeGroup	First-match dispatch	Ordered case predicates with a mandatory default branch
InternalEdgeGroup	System-internal routing	No user-facing conditions

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment