agent/agent-framework
1
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: Add explicit input, output, and workflow_output parameters to @handler, @executor and request_info (#3472)

Browse Source 
* Support specifying types via handler and executor decorators

* Add handling for string types

* Fix typing

* Address PR feedback

* All or nothing for handler typing approach

* Fix mypy issues

* type support for request info

* Fix naming issue

* Fix mypy
This commit is contained in:
Evan Mattson
2026-02-04 07:47:40 +09:00
committed by GitHub
Unverified
parent 8d939f8ffa
commit f56218fa1e
17 changed files with 1718 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/samples/getting_started/workflows/_start-here/step1_executors_and_edges.py
+89 -11
View File
@@ -34,6 +34,18 @@ What this example shows
Simple steps can use this form; a terminal step can yield output
using ctx.yield_output() to provide workflow results.
- Explicit type parameters with @handler:
Instead of relying on type introspection from function signatures, you can explicitly
specify `input`, `output`, and/or `workflow_output` on the @handler decorator.
This is "all or nothing": when ANY explicit parameter is provided, ALL types come
from explicit parameters (introspection is disabled). The `input` parameter is
required; `output` and `workflow_output` are optional.
Examples:
@handler(input=str | int) # Accepts str or int, no outputs
@handler(input=str, output=int) # Accepts str, outputs int
@handler(input=str, output=int, workflow_output=bool) # All three specified
- Fluent WorkflowBuilder API:
add_edge(A, B) to connect nodes, set_start_executor(A), then build() -> Workflow.
@@ -46,8 +58,8 @@ Prerequisites
"""
# Example 1: A custom Executor subclass
# ------------------------------------
# Example 1: A custom Executor subclass using introspection (traditional approach)
# ---------------------------------------------------------------------------------
#
# Subclassing Executor lets you define a named node with lifecycle hooks if needed.
# The work itself is implemented in an async method decorated with @handler.
@@ -71,14 +83,15 @@ class UpperCase(Executor):
Note: The WorkflowContext is parameterized with the type this handler will
emit. Here WorkflowContext[str] means downstream nodes should expect str.
"""
result = text.upper()
# Send the result to the next executor in the workflow.
await ctx.send_message(result)
# Example 2: A standalone function-based executor
# -----------------------------------------------
# Example 2: A standalone function-based executor using introspection
# --------------------------------------------------------------------
#
# For simple steps you can skip subclassing and define an async function with the
# same signature pattern (typed input + WorkflowContext[T_Out, T_W_Out]) and decorate it with
@@ -102,30 +115,95 @@ async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
await ctx.yield_output(result)
async def main():
"""Build and run a simple 2-step workflow using the fluent builder API."""
# Example 3: Using explicit type parameters on @handler
# -----------------------------------------------------
#
# Instead of relying on type introspection, you can explicitly specify input,
# output, and/or workflow_output on the @handler decorator. This is "all or nothing":
# when ANY explicit parameter is provided, ALL types come from explicit parameters
# (introspection is completely disabled). The input parameter is required.
#
# This is useful when:
# - You want to accept multiple types (union types) without complex type annotations
# - The function signature uses Any or a base type for flexibility
# - You want to decouple the runtime type routing from the static type annotations
class ExclamationAdder(Executor):
"""An executor that adds exclamation marks, demonstrating explicit @handler types.
This example shows how to use explicit input and output parameters
on the @handler decorator instead of relying on introspection from the function
signature. This approach is especially useful for union types.
"""
def __init__(self, id: str):
super().__init__(id=id)
@handler(input=str, output=str)
async def add_exclamation(self, message: str, ctx: WorkflowContext) -> None:
"""Add exclamation marks to the input.
Note: The input=str and output=str are explicitly specified on @handler,
so the framework uses those instead of introspecting the function signature.
The WorkflowContext here has no type parameters because the explicit types
on @handler take precedence.
"""
result = f"{message}!!!"
await ctx.send_message(result)
async def main():
"""Build and run workflows using the fluent builder API."""
# Workflow 1: Using introspection-based type detection
# -----------------------------------------------------
upper_case = UpperCase(id="upper_case_executor")
# Build the workflow using a fluent pattern:
# 1) add_edge(from_node, to_node) defines a directed edge upper_case -> reverse_text
# 2) set_start_executor(node) declares the entry point
# 3) build() finalizes and returns an immutable Workflow object
workflow = WorkflowBuilder().add_edge(upper_case, reverse_text).set_start_executor(upper_case).build()
workflow1 = WorkflowBuilder().add_edge(upper_case, reverse_text).set_start_executor(upper_case).build()
# Run the workflow by sending the initial message to the start node.
# The run(...) call returns an event collection; its get_outputs() method
# retrieves the outputs yielded by any terminal nodes.
events = await workflow.run("hello world")
print(events.get_outputs())
# Summarize the final run state (e.g., IDLE)
print("Final state:", events.get_final_state())
print("Workflow 1 (introspection-based types):")
events1 = await workflow1.run("hello world")
print(events1.get_outputs())
print("Final state:", events1.get_final_state())
# Workflow 2: Using explicit type parameters on @handler
# -------------------------------------------------------
exclamation_adder = ExclamationAdder(id="exclamation_adder")
# This workflow demonstrates the explicit input/output feature:
# exclamation_adder uses @handler(input=str, output=str) to
# explicitly declare types instead of relying on introspection.
workflow2 = (
WorkflowBuilder()
.add_edge(upper_case, exclamation_adder)
.add_edge(exclamation_adder, reverse_text)
.set_start_executor(upper_case)
.build()
)
print("\nWorkflow 2 (explicit @handler types):")
events2 = await workflow2.run("hello world")
print(events2.get_outputs())
print("Final state:", events2.get_final_state())
"""
Sample Output:
Workflow 1 (introspection-based types):
['DLROW OLLEH']
Final state: WorkflowRunState.IDLE
Workflow 2 (explicit @handler types):
['!!!DLROW OLLEH']
Final state: WorkflowRunState.IDLE
"""
python/samples/getting_started/workflows/checkpoint/handoff_with_tool_approval_checkpoint_resume.py
+27 -23
View File
@@ -11,8 +11,8 @@ from agent_framework import (
ChatMessage,
Content,
FileCheckpointStorage,
HandoffAgentUserRequest,
HandoffBuilder,
HandoffUserInputRequest,
RequestInfoEvent,
Workflow,
WorkflowOutputEvent,
@@ -102,7 +102,7 @@ def create_workflow(checkpoint_storage: FileCheckpointStorage) -> tuple[Workflow
name="checkpoint_handoff_demo",
participants=[triage, refund, order],
)
.set_coordinator("triage_agent")
.with_start_agent(triage)
.with_checkpointing(checkpoint_storage)
.with_termination_condition(
# Terminate after 5 user messages for this demo
@@ -114,25 +114,27 @@ def create_workflow(checkpoint_storage: FileCheckpointStorage) -> tuple[Workflow
return workflow, triage, refund, order
def _print_handoff_request(request: HandoffUserInputRequest, request_id: str) -> None:
def _print_handoff_agent_user_request(response: AgentResponse) -> None:
"""Display the agent's response messages when requesting user input."""
if not response.messages:
print("(No agent messages)")
return
print("\n[Agent is requesting your input...]")
for message in response.messages:
if not message.text:
continue
speaker = message.author_name or message.role.value
print(f" {speaker}: {message.text}")
def _print_handoff_request(request: HandoffAgentUserRequest, request_id: str) -> None:
"""Log pending handoff request details for debugging."""
print(f"\n{'=' * 60}")
print("WORKFLOW PAUSED - User input needed")
print(f"Request ID: {request_id}")
print(f"Awaiting agent: {request.awaiting_agent_id}")
print(f"Prompt: {request.prompt}")
# Note: After checkpoint restore, conversation may be empty because it's not serialized
# to prevent duplication (the conversation is preserved in the coordinator's state).
# See issue #2667.
if request.conversation:
print("\nConversation so far:")
for msg in request.conversation[-3:]:
author = msg.author_name or msg.role.value
snippet = msg.text[:120] + "..." if len(msg.text) > 120 else msg.text
print(f" {author}: {snippet}")
else:
print("\n(Conversation restored from checkpoint - context preserved in workflow state)")
_print_handoff_agent_user_request(request.agent_response)
print(f"{'=' * 60}\n")
@@ -157,7 +159,7 @@ def _build_responses_for_requests(
"""Create response payloads for each pending request."""
responses: dict[str, object] = {}
for request in pending_requests:
if isinstance(request.data, HandoffUserInputRequest):
if isinstance(request.data, HandoffAgentUserRequest):
if user_response is None:
raise ValueError("User response is required for HandoffUserInputRequest")
responses[request.request_id] = user_response
@@ -199,7 +201,7 @@ async def run_until_user_input_needed(
elif isinstance(event, RequestInfoEvent):
pending_requests.append(event)
if isinstance(event.data, HandoffUserInputRequest):
if isinstance(event.data, HandoffAgentUserRequest):
_print_handoff_request(event.data, event.request_id)
elif isinstance(event.data, Content) and event.data.type == "function_approval_request":
_print_function_approval_request(event.data, event.request_id)
@@ -256,7 +258,7 @@ async def resume_with_responses(
async for event in workflow.run_stream(checkpoint_id=latest_checkpoint.checkpoint_id): # type: ignore[attr-defined]
if isinstance(event, RequestInfoEvent):
restored_requests.append(event)
if isinstance(event.data, HandoffUserInputRequest):
if isinstance(event.data, HandoffAgentUserRequest):
_print_handoff_request(event.data, event.request_id)
elif isinstance(event.data, Content) and event.data.type == "function_approval_request":
_print_function_approval_request(event.data, event.request_id)
@@ -289,7 +291,7 @@ async def resume_with_responses(
elif isinstance(event, RequestInfoEvent):
new_pending_requests.append(event)
if isinstance(event.data, HandoffUserInputRequest):
if isinstance(event.data, HandoffAgentUserRequest):
_print_handoff_request(event.data, event.request_id)
elif isinstance(event.data, Content) and event.data.type == "function_approval_request":
_print_function_approval_request(event.data, event.request_id)
@@ -302,7 +304,7 @@ async def main() -> None:
Demonstrate the checkpoint-based pause/resume pattern for handoff workflows.
This sample shows:
1. Starting a workflow and getting a HandoffUserInputRequest
1. Starting a workflow and getting a HandoffAgentUserRequest
2. Pausing (checkpoint is saved automatically)
3. Resuming from checkpoint with a user response or tool approval (two-step pattern)
4. Continuing the conversation until completion
@@ -361,8 +363,10 @@ async def main() -> None:
print("\n>>> Simulating process restart...\n")
workflow_step, _, _, _ = create_workflow(checkpoint_storage=storage)
needs_user_input = any(isinstance(req.data, HandoffUserInputRequest) for req in pending_requests)
needs_tool_approval = any(isinstance(req.data, Content) and req.data.type == "function_approval_request" for req in pending_requests)
needs_user_input = any(isinstance(req.data, HandoffAgentUserRequest) for req in pending_requests)
needs_tool_approval = any(
isinstance(req.data, Content) and req.data.type == "function_approval_request" for req in pending_requests
)
user_response = None
if needs_user_input: