Implementation:Microsoft Agent framework AgentExecutor Init

Overview

This page documents the implementation of the AgentExecutor class, which wraps an agent conforming to the SupportsAgentRun protocol as a workflow Executor node. The implementation spans lines L64--410 of the source file and provides multiple handler methods to accept a variety of input types, normalizing them into a uniform AgentExecutorResponse output.

Source Location

Property	Value
File	python/packages/core/agent_framework/_workflows/_agent_executor.py
Lines	L64--410
Import	from agent_framework import AgentExecutor

Class Signature

class AgentExecutor(Executor):
    def __init__(
        self,
        agent: SupportsAgentRun,
        *,
        agent_thread: AgentThread | None = None,
        id: str | None = None,
    ):

Constructor Parameters

__init__ Parameters

Parameter	Type	Default	Description
agent	SupportsAgentRun	(required)	The agent instance to wrap. Must implement the SupportsAgentRun protocol.
agent_thread	AgentThread | None	None	Optional conversational thread. When provided, the executor reuses this thread across invocations to maintain conversation continuity.
id	str | None	None	Optional identifier for this executor node within the workflow graph. Surfaces as AgentExecutorResponse.executor_id.

Handler Methods

AgentExecutor exposes five @handler-decorated methods, one for each supported input type. The workflow engine dispatches to the appropriate handler based on the runtime type of the incoming payload.

Handler Dispatch Table

Handler Method	Input Type	Description
run	AgentExecutorRequest	Primary entry point. Accepts a structured request carrying a message payload and optional thread override.
from_response	AgentExecutorResponse	Enables chaining: consumes the output of an upstream AgentExecutor and feeds it into this agent.
from_str	str	Convenience handler that wraps a plain string into a message before invoking the agent.
from_message	Message	Accepts a single Message object and passes it directly to the agent.
from_messages	list	Accepts a list of Message objects representing a conversation fragment.

Input / Output Contract

I/O Contract

Direction	Type	Description
Input	str	Plain text; handled by from_str.
Input	Message	Single message; handled by from_message.
Input	list[Message]	Message sequence; handled by from_messages.
Input	AgentExecutorRequest	Structured request; handled by run.
Input	AgentExecutorResponse	Upstream executor output; handled by from_response.
Output	AgentExecutorResponse	Contains .executor_id, .agent_response, and .full_conversation.

Output Fields

Field	Type	Description
.executor_id	str | None	The identifier of the AgentExecutor node that produced this response.
.agent_response	AgentResponse	The raw response returned by the wrapped agent after execution.
.full_conversation	list[Message]	The complete message history for the thread after the agent has responded.

Usage Example

from agent_framework import AgentExecutor, WorkflowBuilder

# Wrap an agent as an executor node with an explicit ID
agent_node = AgentExecutor(my_agent, id="assistant")

# Build a workflow that starts with the agent node
workflow = WorkflowBuilder(start_executor=agent_node).build()

# Run the workflow with a plain string input
result = await workflow.run("Hello")

# Access the output
print(result.executor_id)        # "assistant"
print(result.agent_response)     # The agent's response object
print(result.full_conversation)  # Full thread history

Chaining Multiple Agent Executors

from agent_framework import AgentExecutor, WorkflowBuilder

# Create two executor nodes wrapping different agents
researcher = AgentExecutor(research_agent, id="researcher")
writer = AgentExecutor(writing_agent, id="writer")

# Chain them: researcher output feeds into writer via from_response handler
workflow = (
    WorkflowBuilder(start_executor=researcher)
    .add_edge(researcher, writer)
    .build()
)

result = await workflow.run("Summarize recent AI developments")
# result is produced by the writer node
print(result.executor_id)  # "writer"

Internal Flow

The following pseudocode outlines the internal dispatch and execution logic:

1. Workflow engine sends payload to AgentExecutor
2. @handler dispatch selects method based on payload type:
     str              -> from_str()
     Message          -> from_message()
     list[Message]    -> from_messages()
     AgentExecutorRequest   -> run()
     AgentExecutorResponse  -> from_response()
3. Handler normalizes input into message(s)
4. Agent is invoked via agent.run(messages, thread=agent_thread)
5. Response is wrapped in AgentExecutorResponse:
     - executor_id = self.id
     - agent_response = agent result
     - full_conversation = thread messages
6. AgentExecutorResponse is returned to the workflow engine

Metadata

Property	Value
Domains	Workflow_Engine, Agent_Architecture
Implementation ID	Microsoft_Agent_framework_AgentExecutor_Init
Source File	python/packages/core/agent_framework/_workflows/_agent_executor.py
Line Range	L64--410
Status	Active

Related

• Principle:Microsoft_Agent_framework_Agent_Executor_Wrapping -- The design principle that this implementation realizes, describing why agents are wrapped as executor nodes.