feat(beta): add on_human_input middleware hook (#2477)

* feat(beta): add on_human_input middleware hook

Add a new `on_human_input` lifecycle hook to `BaseMiddleware`, allowing
middleware to intercept, modify, or short-circuit human-in-the-loop
requests and responses. Wire the hook through `wrap_hitl` and update
docs and exports accordingly.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* docs(beta): update roadmap details and fix tools.mdx formatting

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(beta): replace _CapturingStream stub with MemoryStream in client tool tests

Use MemoryStream and MagicMock instead of a custom _CapturingStream helper,
making tests exercise the real stream infrastructure.

* refactor(beta): extract shared ClientTool fixture in client tool tests

Deduplicate schema/ClientTool construction into a pytest fixture
and unify tool names across all three tests.

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Mark Sze <66362098+marklysze@users.noreply.github.com>

Author: 3 people

autogen/beta/agent.py

@@ -328,7 +328,7 @@ async def _call_client(context: Context) -> None:
             )
 
             stack.enter_context(
-                context.stream.where(HumanInputRequest).sub_scope(self.__hitl_hook),
+                context.stream.where(HumanInputRequest).sub_scope(self.__hitl_hook(middleware_instances)),
             )
 
             self.__tool_executor.register(

autogen/beta/hitl.py

@@ -2,22 +2,28 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Iterable
 from contextlib import AsyncExitStack
+from functools import partial
 from typing import TypeAlias
 
 from .annotations import Context
 from .events import HumanInputRequest, HumanMessage
 from .exceptions import HumanInputNotProvidedError
+from .middleware.base import BaseMiddleware, HumanInputHook
 from .utils import CONTEXT_OPTION_NAME, build_model
 
 HumanHook: TypeAlias = Callable[..., HumanMessage] | Callable[..., Awaitable[HumanMessage]]
 
+HitlExecution: TypeAlias = Callable[[HumanInputRequest, Context], Awaitable[None]]
 
-def wrap_hitl(func: HumanHook) -> None:
+
+def wrap_hitl(
+    func: HumanHook,
+) -> Callable[[Iterable["BaseMiddleware"]], HitlExecution]:
     call_model = build_model(func)
 
-    async def wrapper(event: HumanInputRequest, context: Context) -> None:
+    async def _call_model(event: HumanInputRequest, context: Context) -> HumanMessage:
         async with AsyncExitStack() as stack:
             event = await call_model.asolve(
                 event,
@@ -26,10 +32,24 @@ async def wrapper(event: HumanInputRequest, context: Context) -> None:
                 dependency_provider=context.dependency_provider,
                 **{CONTEXT_OPTION_NAME: context},
             )
-        await context.send(event)
+        return event
+
+    def make_hook(middlewares: Iterable["BaseMiddleware"]) -> HitlExecution:
+        ask_user: HumanInputHook = _call_model
+        for middleware in middlewares:
+            ask_user = partial(middleware.on_human_input, ask_user)
+
+        async def wrapper(event: HumanInputRequest, context: Context) -> None:
+            event = await ask_user(event, context)
+            await context.send(event)
+
+        return wrapper
+
+    return make_hook
 
-    return wrapper
 
+def default_hitl_hook(middlewares: Iterable["BaseMiddleware"]) -> HitlExecution:
+    async def _call_model(event: HumanInputRequest, context: Context) -> None:
+        raise HumanInputNotProvidedError
 
-def default_hitl_hook() -> HumanMessage:
-    raise HumanInputNotProvidedError
+    return _call_model

autogen/beta/middleware/__init__.py

@@ -2,13 +2,27 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from .base import AgentTurn, BaseMiddleware, LLMCall, Middleware, ToolExecution, ToolResultType
-from .builtin import HistoryLimiter, LoggingMiddleware, RetryMiddleware, TokenLimiter
+from .base import (
+    AgentTurn,
+    BaseMiddleware,
+    HumanInputHook,
+    LLMCall,
+    Middleware,
+    ToolExecution,
+    ToolResultType,
+)
+from .builtin import (
+    HistoryLimiter,
+    LoggingMiddleware,
+    RetryMiddleware,
+    TokenLimiter,
+)
 
 __all__ = (
     "AgentTurn",
     "BaseMiddleware",
     "HistoryLimiter",
+    "HumanInputHook",
     "LLMCall",
     "LoggingMiddleware",
     "Middleware",

autogen/beta/middleware/base.py

@@ -6,7 +6,16 @@
 from typing import Any, Protocol, TypeAlias
 
 from autogen.beta.annotations import Context
-from autogen.beta.events import BaseEvent, ClientToolCall, ModelResponse, ToolCall, ToolError, ToolResult
+from autogen.beta.events import (
+    BaseEvent,
+    ClientToolCall,
+    HumanInputRequest,
+    HumanMessage,
+    ModelResponse,
+    ToolCall,
+    ToolError,
+    ToolResult,
+)
 
 
 class MiddlewareFactory(Protocol):
@@ -36,6 +45,7 @@ def __call__(
 AgentTurn: TypeAlias = Callable[["BaseEvent", "Context"], Awaitable["ModelResponse"]]
 ToolExecution: TypeAlias = Callable[["ToolCall", "Context"], Awaitable[ToolResultType]]
 LLMCall: TypeAlias = Callable[["Sequence[BaseEvent]", "Context"], Awaitable["ModelResponse"]]
+HumanInputHook: TypeAlias = Callable[["HumanInputRequest", "Context"], Awaitable["HumanMessage"]]
 
 
 class BaseMiddleware:
@@ -70,3 +80,11 @@ async def on_llm_call(
         context: "Context",
     ) -> "ModelResponse":
         return await call_next(events, context)
+
+    async def on_human_input(
+        self,
+        call_next: HumanInputHook,
+        event: "HumanInputRequest",
+        context: "Context",
+    ) -> "HumanMessage":
+        return await call_next(event, context)

test/beta/middleware/test_human_input.py

@@ -0,0 +1,213 @@
+# Copyright (c) 2023 - 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from autogen.beta import Agent, Context
+from autogen.beta.events import BaseEvent, HumanInputRequest, HumanMessage, ToolCall
+from autogen.beta.middleware import BaseMiddleware, Middleware
+from autogen.beta.middleware.base import HumanInputHook
+from autogen.beta.testing import TestConfig
+
+
+@pytest.fixture()
+def test_config() -> TestConfig:
+    return TestConfig(
+        ToolCall(name="my_tool"),
+        "result",
+    )
+
+
+class MockHumanInputMiddleware(BaseMiddleware):
+    def __init__(
+        self,
+        event: BaseEvent,
+        ctx: Context,
+        mock: MagicMock,
+    ) -> None:
+        super().__init__(event, ctx)
+        self.mock = mock
+
+    async def on_human_input(
+        self,
+        call_next: HumanInputHook,
+        event: HumanInputRequest,
+        ctx: Context,
+    ) -> HumanMessage:
+        self.mock.enter(event.content)
+        result = await call_next(event, ctx)
+        self.mock.exit(result.content)
+        return result
+
+
+@pytest.mark.asyncio()
+async def test_human_input_middleware(mock: MagicMock, test_config: TestConfig) -> None:
+    async def my_tool(ctx: Context) -> str:
+        await ctx.input("Say smth", timeout=1.0)
+        return ""
+
+    def hitl_hook(event: HumanInputRequest) -> HumanMessage:
+        return HumanMessage(content="answer")
+
+    agent = Agent(
+        "",
+        config=test_config,
+        tools=[my_tool],
+        hitl_hook=hitl_hook,
+        middleware=[Middleware(MockHumanInputMiddleware, mock=mock)],
+    )
+
+    await agent.ask("Hi!")
+
+    mock.enter.assert_called_once_with("Say smth")
+    mock.exit.assert_called_once_with("answer")
+
+
+class OrderingHumanInputMiddleware(BaseMiddleware):
+    def __init__(
+        self,
+        event: BaseEvent,
+        ctx: Context,
+        mock: MagicMock,
+        position: int,
+    ) -> None:
+        super().__init__(event, ctx)
+        self.mock = mock
+        self.position = position
+
+    async def on_human_input(
+        self,
+        call_next: HumanInputHook,
+        event: HumanInputRequest,
+        ctx: Context,
+    ) -> HumanMessage:
+        self.mock.enter(self.position)
+        result = await call_next(event, ctx)
+        self.mock.exit(self.position)
+        return result
+
+
+@pytest.mark.asyncio()
+async def test_human_input_middleware_call_sequence(mock: MagicMock, test_config: TestConfig) -> None:
+    async def my_tool(ctx: Context) -> str:
+        await ctx.input("Say smth", timeout=1.0)
+        return ""
+
+    def hitl_hook(event: HumanInputRequest) -> HumanMessage:
+        return HumanMessage(content="answer")
+
+    agent = Agent(
+        "",
+        config=test_config,
+        tools=[my_tool],
+        hitl_hook=hitl_hook,
+        middleware=[Middleware(OrderingHumanInputMiddleware, mock=mock, position=i) for i in range(1, 4)],
+    )
+
+    await agent.ask("Hi!")
+
+    assert [c.args[0] for c in mock.enter.call_args_list] == [1, 2, 3]
+    assert [c.args[0] for c in mock.exit.call_args_list] == [3, 2, 1]
+
+
+@pytest.mark.asyncio()
+async def test_human_input_middleware_mutates_request(mock: MagicMock, test_config: TestConfig) -> None:
+    class MutatingMiddleware(BaseMiddleware):
+        async def on_human_input(
+            self,
+            call_next: HumanInputHook,
+            event: HumanInputRequest,
+            ctx: Context,
+        ) -> HumanMessage:
+            event = HumanInputRequest(content=event.content + "!")
+            return await call_next(event, ctx)
+
+    async def my_tool(ctx: Context) -> str:
+        await ctx.input("Say smth", timeout=1.0)
+        return ""
+
+    def hitl_hook(event: HumanInputRequest) -> HumanMessage:
+        mock.hitl(event.content)
+        return HumanMessage(content="answer")
+
+    agent = Agent(
+        "",
+        config=test_config,
+        tools=[my_tool],
+        hitl_hook=hitl_hook,
+        middleware=[MutatingMiddleware, MutatingMiddleware, MutatingMiddleware],
+    )
+
+    await agent.ask("Hi!")
+
+    mock.hitl.assert_called_once_with("Say smth!!!")
+
+
+@pytest.mark.asyncio()
+async def test_human_input_middleware_mutates_response(mock: MagicMock, test_config: TestConfig) -> None:
+    class MutatingMiddleware(BaseMiddleware):
+        async def on_human_input(
+            self,
+            call_next: HumanInputHook,
+            event: HumanInputRequest,
+            ctx: Context,
+        ) -> HumanMessage:
+            result = await call_next(event, ctx)
+            return HumanMessage(content=result.content + "!")
+
+    async def my_tool(ctx: Context) -> str:
+        mock(await ctx.input("Say smth", timeout=1.0))
+        return ""
+
+    def hitl_hook(event: HumanInputRequest) -> HumanMessage:
+        return HumanMessage(content="answer")
+
+    agent = Agent(
+        "",
+        config=test_config,
+        tools=[my_tool],
+        hitl_hook=hitl_hook,
+        middleware=[MutatingMiddleware, MutatingMiddleware, MutatingMiddleware],
+    )
+
+    await agent.ask("Hi!")
+
+    mock.assert_called_once_with("answer!!!")
+
+
+@pytest.mark.asyncio()
+async def test_human_input_middleware_short_circuits(mock: MagicMock, test_config: TestConfig) -> None:
+    class ShortCircuitMiddleware(BaseMiddleware):
+        async def on_human_input(
+            self,
+            call_next: HumanInputHook,
+            event: HumanInputRequest,
+            ctx: Context,
+        ) -> HumanMessage:
+            mock.intercepted(event.content)
+            return HumanMessage(content="intercepted")
+
+    async def my_tool(ctx: Context) -> str:
+        mock(await ctx.input("Say smth", timeout=1.0))
+        return ""
+
+    def hitl_hook(event: HumanInputRequest) -> HumanMessage:
+        mock.hitl()
+        return HumanMessage(content="answer")
+
+    agent = Agent(
+        "",
+        config=test_config,
+        tools=[my_tool],
+        hitl_hook=hitl_hook,
+        middleware=[Middleware(ShortCircuitMiddleware)],
+    )
+
+    await agent.ask("Hi!")
+
+    mock.intercepted.assert_called_once_with("Say smth")
+    mock.hitl.assert_not_called()
+    mock.assert_called_once_with("intercepted")