Implementation:Microsoft Agent framework Workflow As Agent

Property	Value
Implementation Name	Workflow As Agent
SDK	Microsoft Agent Framework
Repository	Microsoft Agent Framework
Source File	python/packages/core/agent_framework/_workflows/_workflow.py
Line Range	L841-866
Import	Access via Workflow instance (e.g. workflow.as_agent())
Type	Instance method returning WorkflowAgent
Domains	Agent_Architecture, Multi_Agent_Systems

Overview

The Workflow.as_agent() method converts a Workflow instance into a WorkflowAgent -- a BaseAgent subclass that satisfies the SupportsAgentRun protocol. This enables orchestrated workflows (sequential, concurrent, or custom graphs) to be used in any context that expects a single agent interface, including as participants in higher-level orchestrations.

Code Reference

Source Location

Property	Value
File	python/packages/core/agent_framework/_workflows/_workflow.py
Class	Workflow
Method	as_agent
Lines	841-866
Delegates To	WorkflowAgent in agent_framework._workflows._agent

Signature

def as_agent(self, name: str | None = None) -> WorkflowAgent:

Import Statement

# No direct import needed -- access via a Workflow instance:
from agent_framework.orchestrations import SequentialBuilder

workflow = SequentialBuilder(participants=[agent_a, agent_b]).build()
workflow_agent = workflow.as_agent(name="my_team")

I/O Contract

Inputs

Parameter	Type	Default	Description
name	None	None	Optional human-readable name for the returned WorkflowAgent. When None, the agent is assigned a generated identifier of the form WorkflowAgent_<hex8>.

Output

Type	Description
WorkflowAgent	A BaseAgent subclass wrapping the workflow. Exposes the standard run() method with support for streaming, threads, and checkpoints.

Raised Exceptions

Exception	Condition
ValueError	The workflow's start executor does not accept list[Message] as an input type.
ValueError	The workflow has no configured start executor (entry point is undefined).

WorkflowAgent.run() I/O Contract

Once wrapped, the WorkflowAgent exposes the following run() interface:

Inputs

Parameter	Type	Default	Description
messages	Message | list[str] | list[Message] | None	None	User input to the workflow. Strings are converted to Message(role="user", text=...). Required for new runs; None when resuming from a checkpoint.
stream	bool	False	When True, returns an AsyncIterable[AgentResponseUpdate] for real-time streaming. When False, returns an awaitable AgentResponse.
thread	None	None	Optional conversation thread for multi-turn context. History from the thread is prepended to the input messages.
checkpoint_id	None	None	When provided, resumes the workflow from the specified checkpoint instead of starting fresh.
checkpoint_storage	None	None	Runtime storage backend for checkpointing. Required when using checkpoint_id or when enabling checkpoint capture for the run.

Output

Condition	Type	Description
stream=False (default)	AgentResponse	Complete response containing: .text -- Aggregated text from all output events. .messages -- List of Message objects produced by workflow executors. .usage_details -- Merged token usage across all LLM calls.
stream=True	AsyncIterable[AgentResponseUpdate]	Yields incremental updates as workflow executors produce output events. Each update contains .text, .contents, and .author_name identifying the originating executor.

Execution Flow

The as_agent() method and the resulting WorkflowAgent implement the following sequence:

1. Wrapping: Workflow.as_agent(name) instantiates a WorkflowAgent, passing self as the workflow reference.
2. Validation: The WorkflowAgent.__init__ retrieves the workflow's start executor and verifies that at least one of its input_types is compatible with list[Message]. If not, a ValueError is raised.
3. Input Normalization: When run() is called, normalize_messages_input() converts the caller's input (str, Message, or list) into a list[Message].
4. Conversation Assembly: If a thread is provided, historical messages from the thread's message store are prepended to the normalized input.
5. Workflow Execution: The assembled list[Message] is passed to self._workflow.run(message=...). The workflow executes its graph of executors.
6. Event Translation: Workflow events are filtered and converted:
   • type='output' events are converted to AgentResponse messages or AgentResponseUpdate objects.
   • type='request_info' events are converted to FunctionCallContent and FunctionApprovalRequestContent pairs.
   • All other event types are silently discarded.
7. Thread Notification: After execution completes, both the input messages and the response messages are written to the thread's message store for future turns.

Usage Examples

Basic Workflow Wrapping

from agent_framework.orchestrations import SequentialBuilder

# Build a two-agent sequential workflow
workflow = SequentialBuilder(participants=[writer, reviewer]).build()

# Wrap the workflow as a single agent
workflow_agent = workflow.as_agent(name="writing_team")

# Use the standard agent interface
response = await workflow_agent.run("Write a blog post")
print(response.text)

Streaming Execution

workflow_agent = workflow.as_agent(name="writing_team")

async for update in workflow_agent.run("Draft a product description", stream=True):
    print(update.text, end="")

Multi-Turn Conversation with Thread

from agent_framework import AgentThread

workflow_agent = workflow.as_agent(name="writing_team")
thread = AgentThread()

r1 = await workflow_agent.run("Write a tagline for an eBike", thread=thread)
r2 = await workflow_agent.run("Make it shorter and punchier", thread=thread)

Nested Composition

from agent_framework.orchestrations import SequentialBuilder

# Inner workflow: writer -> reviewer
inner_workflow = SequentialBuilder(participants=[writer, reviewer]).build()
inner_agent = inner_workflow.as_agent(name="content_team")

# Outer workflow: content_team -> publisher
outer_workflow = SequentialBuilder(participants=[inner_agent, publisher]).build()
response = await outer_workflow.run("Create and publish a press release")

Internal Components

Component	Location	Role
Workflow.as_agent()	_workflow.py:L841-866	Factory method; instantiates WorkflowAgent.
WorkflowAgent	_agent.py	BaseAgent subclass; implements run(), event translation, and pending-request management.
normalize_messages_input()	_message_utils.py	Message | Sequence[str | Message] | None into list[Message].
WorkflowAgent.merge_updates()	_agent.py	Aggregates streaming AgentResponseUpdate objects into a final AgentResponse.

Related Pages

• Principle:Microsoft_Agent_framework_Workflow_Agent_Wrapping

Sources

Type	Name	URL
Repo	Microsoft Agent Framework	https://github.com/microsoft/agent-framework