Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Microsoft Agent framework Response Handler Decorator

From Leeroopedia
Revision as of 11:31, 16 February 2026 by Admin (talk | contribs) (Auto-imported from implementations/Microsoft_Agent_framework_Response_Handler_Decorator.md)
(diff) ← Older revision | Latest revision (diff) | Newer revision → (diff)


Template:Microsoft Agent framework Sidebar

Overview

The Response Handler Decorator (@response_handler) marks an Executor method as the handler for responses to previously emitted request_info events in a human-in-the-loop workflow. When workflow.run(responses={request_id: data}) is called, the framework routes each response to the matching decorated method, passing the original request payload, the human-provided response, and the current workflow context.

Property Value
Source python/packages/core/agent_framework/_workflows/_request_info_mixin.py (lines L114-133)
Import from agent_framework import response_handler
Inputs A method on an Executor subclass with signature (self, original_request: RequestType, response: ResponseType, ctx: WorkflowContext) -> None
Outputs The same method, annotated with internal metadata for response routing
Related Principle Human-in-the-Loop Response

Code Reference

Full Signature

def response_handler(
    func: Callable[[Any, RequestType, ResponseType, WorkflowContext], Awaitable[None]],
) -> Callable[[Any, RequestType, ResponseType, WorkflowContext], Awaitable[None]]:

Source Location

python/packages/core/agent_framework/_workflows/_request_info_mixin.py, lines L114-133. The decorator is defined as a top-level function that annotates the target method with internal routing metadata used by the workflow engine to dispatch responses.

I/O Contract

Inputs

Parameter Type Description
self Executor The executor instance that owns this handler method.
original_request RequestType The original request payload that was passed to ctx.request_info() when the request was emitted. Allows the handler to recall the context of what was asked.
response ResponseType The response data provided by the human operator, matching the type specified in the request_info() call.
ctx WorkflowContext The workflow context providing access to yield_output(), request_info(), state management, and other workflow operations.

Outputs

The decorated method returns None. Side effects are performed through the WorkflowContext:

Side Effect Method Description
Emit output ctx.yield_output(value) Produces a workflow output based on the processed response.
Request more input ctx.request_info(prompt, type, request_id=...) Emits a follow-up human-in-the-loop request, creating an iterative collection loop.
Update state ctx.state Modifies shared workflow state for downstream executors.

Description

The @response_handler decorator works through the following steps:

  1. Method annotation: When applied to an executor method, the decorator attaches internal metadata that identifies the method as a response handler. This metadata is inspected by the workflow engine during response dispatch.
  2. Type extraction: The decorator extracts RequestType and ResponseType from the method's type annotations. These types are used for runtime validation of incoming response data.
  3. Registration: When the executor is instantiated within a workflow, the framework scans for methods decorated with @response_handler and registers them in the response routing table, keyed by the executor and handler identity.
  4. Dispatch: When workflow.run(responses={request_id: data}) is called, the framework resolves the pending request associated with each request_id, identifies the corresponding executor and handler, and invokes the decorated method with the original request, the response, and the current WorkflowContext.

The decorator supports the standard async method signature. The handler must be defined as an async def method returning None.

Examples

Basic Usage: Collecting User Input

from agent_framework import Executor, handler, response_handler, WorkflowContext

class InputCollector(Executor):
    @handler
    async def start(self, msg: str, ctx: WorkflowContext) -> None:
        await ctx.request_info("What is your name?", str, request_id="name")

    @response_handler
    async def handle_name(
        self, original_request: str, response: str, ctx: WorkflowContext
    ) -> None:
        await ctx.yield_output(f"Hello, {response}!")

# Resume workflow with response
result = await workflow.run(responses={"name": "Alice"})

Multi-Step Collection

from agent_framework import Executor, handler, response_handler, WorkflowContext

class FormCollector(Executor):
    @handler
    async def start(self, msg: str, ctx: WorkflowContext) -> None:
        await ctx.request_info("Enter your name:", str, request_id="name")

    @response_handler
    async def handle_name(
        self, original_request: str, response: str, ctx: WorkflowContext
    ) -> None:
        ctx.state["name"] = response
        await ctx.request_info("Enter your email:", str, request_id="email")

    @response_handler
    async def handle_email(
        self, original_request: str, response: str, ctx: WorkflowContext
    ) -> None:
        name = ctx.state["name"]
        await ctx.yield_output(f"Registered {name} with email {response}")

# First resume: provide name
result = await workflow.run(responses={"name": "Alice"})
# Second resume: provide email
result = await workflow.run(responses={"email": "alice@example.com"})

Wiring Into a Workflow

from agent_framework import Workflow, Executor, handler, response_handler, WorkflowContext

class ApprovalGate(Executor):
    @handler
    async def check(self, msg: str, ctx: WorkflowContext) -> None:
        await ctx.request_info(
            f"Approve processing of: {msg}?", bool, request_id="approve"
        )

    @response_handler
    async def handle_approval(
        self, original_request: str, response: bool, ctx: WorkflowContext
    ) -> None:
        if response:
            await ctx.yield_output(f"Approved: {original_request}")
        else:
            await ctx.yield_output("Processing rejected by user.")

workflow = Workflow(executors=[ApprovalGate()])

# Initial run triggers the request_info event
result = await workflow.run("Sensitive operation XYZ")

# Resume with human decision
result = await workflow.run(responses={"approve": True})
print(result.output)  # "Approved: Approve processing of: Sensitive operation XYZ?"

Related Pages

Template:Sources

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment