Connect agents through typed executor graphs

Learning objective

Implement agent orchestration with Microsoft Agent Framework workflows, agents, executors, and edges.

Core idea

For an architect implementing multi-agent orchestration, Microsoft Agent Framework’s graph API makes the execution contract explicit. A workflow coordinates processing units; an executor processes typed messages; an agent performs model-driven work; an edge determines where a message goes. These are related abstractions, but they have different responsibilities.

Put uncertain reasoning inside an agent and deterministic processing inside an ordinary executor. Connect them with edges whose conditions examine defined results. This lets you inspect the allowed paths without reading a prompt and hoping that the model follows the intended sequence.

This lesson uses the Python graph API. The workflow concepts page also documents .NET and Go graph APIs and labels Python’s functional workflow API experimental. Do not transfer identifiers between languages or assume that the functional API’s native control flow is the same interface as graph edges.

How it works

Define the executable graph

Use WorkflowBuilder(start_executor=...) to identify the entry executor, .add_edge(source, target) to connect processing units, and .build() to construct the workflow. The workflow owns execution and routing; the builder describes those rules.

An executor can be a class derived from Executor with methods decorated by @handler, or a function decorated by @executor. In the Python graph API, handler type annotations identify accepted message types. Alternatively, explicit decorator parameters declare the types: input is required, while output and workflow_output are optional. Do not mix explicit parameters with type annotations. A class can define handlers for multiple input types. Treat those types as contracts between components rather than passing unrelated dictionaries through every stage.

The following complete deterministic example shows the two output channels:

import asyncio
from typing_extensions import Never
from agent_framework import WorkflowBuilder, WorkflowContext, executor

@executor(id="normalize")
async def normalize(text: str, ctx: WorkflowContext[str]) -> None:
    await ctx.send_message(text.strip())

@executor(id="finish")
async def finish(text: str, ctx: WorkflowContext[Never, str]) -> None:
    await ctx.yield_output(text.upper())

async def main() -> None:
    workflow = (
        WorkflowBuilder(start_executor=normalize, output_from=[finish])
        .add_edge(normalize, finish)
        .build()
    )
    result = await workflow.run("  approved  ")
    print(result.get_outputs())

asyncio.run(main())

The output is ["APPROVED"]. Normalization sends a message to another executor; finishing yields a result to the caller. No model call is needed for either operation.

Adapt agents to message processing

An agent does not have to implement a custom executor class. Passing an agent directly to WorkflowBuilder or .add_edge() automatically wraps it in an AgentExecutor in Python. Create an explicit AgentExecutor(agent, id="reviewer") when you need to configure the adapter separately.

The adapter normalizes supported inputs into messages, invokes agent.run(), and dispatches an AgentExecutorResponse downstream. Its agent_response carries the agent’s response; its full_conversation preserves the conversation used for chaining. Ordinary executors can inspect or transform that response before routing further work.

Use AgentExecutorRequest(messages=[...], should_respond=...) when you need explicit control over invocation. With should_respond=False, the executor caches the messages without running the agent. With True, it runs the agent. Other supported input types, including a string and an AgentExecutorResponse, trigger execution directly. A message arriving at an executor and a model producing a response are therefore separate events.

The Python adapter’s context_mode controls what it consumes from an upstream response. The default "full" passes the upstream conversation; "last_agent" uses the last agent’s contribution, and "custom" uses the required context_filter callable for custom selection. Choose deliberately: a reviewer might need the request and evidence, while a formatter might need only a validated draft. See 2.4.1 for sequential context-transfer policy.

Route messages and select outputs

A direct edge routes a message from its source to its target. Add condition=... to .add_edge() for conditional routing. The predicate examines the source message, so write it against the actual output type, not the original request.

When several conditional paths represent alternatives, specify what happens to every supported outcome. An unrecognized classification should reach a defined review or failure path, rather than disappearing because every predicate returns false. See 2.4.3 for supervisor routing and 2.4.2 for fan-out and fan-in policies.

Use ctx.send_message(...) for downstream processing and ctx.yield_output(...) for caller-visible workflow results. On the Python builder, output_from designates terminal-output executors and intermediate_output_from designates progress-output executors. Their yielded events have type == "output" and type == "intermediate", respectively. Retrieve them with get_outputs() and get_intermediate_outputs().

If either designation list is supplied, yields from executors in neither list are hidden, although those executors can still send messages. If both lists are omitted, yields emit output events for compatibility. Explicit output selection prevents a specialist’s draft from appearing as the workflow’s accepted answer. Sending a message advances internal processing, while a yielded value reaches the caller only as output selection allows.

Execute and observe the contract

For non-streaming execution, await workflow.run(input_message) and inspect the returned outputs. For streaming execution, iterate workflow.run(input_message, stream=True) and inspect event types and data. Agent adapters aggregate streaming updates into a complete response for downstream dispatch.

A progress event establishes that processing occurred; it does not establish that the application accepted the result. Keep the terminal executor responsible for producing the accepted response. Checkpoint-based approval and resumption have their own lifecycle; see 2.4.6.

Key distinctions and trade-offs

Agent versus executor: An agent reasons with a model. An executor integrates that work, ordinary code, or an external operation into message processing. Making every step an agent adds uncertainty where a deterministic transformation would suffice.

Routing versus reasoning: An edge expresses permitted movement through the graph. A model’s suggestion to consult a specialist does not create an edge or authorize its execution.

Messages versus outputs: Sending a message advances internal processing. Yielding output exposes a value to the caller, subject to builder output selection. The same string can have different architectural meanings depending on the channel.

Python versus .NET: Python uses AgentExecutor, @handler, and ctx.send_message(). .NET automatically adapts AIAgent through its agent binding and uses AddEdge and WithOutputFrom. Use one language’s documented protocol consistently.

Architecture / decision scenario

A claims workflow receives a description, asks a specialist to draft an assessment, validates the assessment, and returns only the approved summary. Build the graph with a deterministic intake executor, an agent wrapped by AgentExecutor, a validation executor, and a final response executor.

Intake emits the defined agent input. Validation receives AgentExecutorResponse, extracts the assessment, and emits an application-defined result containing a validity decision and normalized evidence. Conditional edges route valid results to final formatting and invalid results to review. Neither predicate examines free-form prose as if it were a trusted routing instruction.

Configure output_from to include the accepted response and explicit review outcome. Configure intermediate_output_from for diagnostic progress if needed. A streamed specialist draft remains progress until validation succeeds. The design uses all four abstractions while keeping reasoning, routing, and acceptance independently inspectable.

Common exam traps

• Treating an agent, an executor, and a workflow as interchangeable objects.
• Calling send_message() and expecting a terminal caller-visible result.
• Forgetting that an agent can be wrapped automatically.
• Assuming should_respond=False invokes the model.
• Evaluating an edge condition against the wrong message type.
• Presenting Python’s experimental functional API as an identical graph interface.

What to remember

Build the execution contract with WorkflowBuilder, typed executors, and explicit edges. Use AgentExecutor to adapt reasoning to that contract. Select terminal and intermediate output sources deliberately, and validate a specialist’s result before routing or exposing it.

Knowledge check

References

• Microsoft, Workflow concepts
• Microsoft, Executors
• Microsoft, Edges
• Microsoft, Agent Executor