Python: Default DevUI workflows to string input when start node is auto-wrapped agent (#1143)

* DevUI workflows default to string input when start node is AgentExecutor

* Remove unnecessary sample

* Fix lint issue
This commit is contained in:
Evan Mattson
2025-10-03 17:11:38 +09:00
committed by GitHub
Unverified
parent 5a208ab1e1
commit c543dbaa92
5 changed files with 230 additions and 20 deletions
@@ -15,6 +15,26 @@ from agent_framework_devui._mapper import MessageMapper
from agent_framework_devui.models._openai_custom import AgentFrameworkExtraBody, AgentFrameworkRequest
class _DummyStartExecutor:
"""Minimal executor stub exposing handler metadata for tests."""
def __init__(self, *, input_types=None, handlers=None):
if input_types is not None:
self.input_types = list(input_types)
if handlers is not None:
self._handlers = dict(handlers)
class _DummyWorkflow:
"""Simple workflow stub returning configured start executor."""
def __init__(self, start_executor):
self._start_executor = start_executor
def get_start_executor(self):
return self._start_executor
@pytest.fixture
def test_entities_dir():
"""Use the samples directory which has proper entity structure."""
@@ -137,6 +157,50 @@ async def test_executor_missing_entity_id(executor):
assert entity_id is None
def test_executor_get_start_executor_message_types_uses_handlers():
"""Ensure handler metadata is surfaced when input_types missing."""
executor = AgentFrameworkExecutor(EntityDiscovery(None), MessageMapper())
start_executor = _DummyStartExecutor(handlers={str: lambda *_: None})
workflow = _DummyWorkflow(start_executor)
start, message_types = executor._get_start_executor_message_types(workflow)
assert start is start_executor
assert str in message_types
def test_executor_select_primary_input_prefers_string():
"""Select string input even when discovered after other handlers."""
executor = AgentFrameworkExecutor(EntityDiscovery(None), MessageMapper())
placeholder_type = type("Placeholder", (), {})
chosen = executor._select_primary_input_type([placeholder_type, str])
assert chosen is str
def test_executor_parse_structured_prefers_input_field():
"""Structured payloads map to string when agent start requires text."""
executor = AgentFrameworkExecutor(EntityDiscovery(None), MessageMapper())
start_executor = _DummyStartExecutor(handlers={type("Req", (), {}): None, str: lambda *_: None})
workflow = _DummyWorkflow(start_executor)
parsed = executor._parse_structured_workflow_input(workflow, {"input": "hello"})
assert parsed == "hello"
def test_executor_parse_raw_falls_back_to_string():
"""Raw inputs remain untouched when start executor expects text."""
executor = AgentFrameworkExecutor(EntityDiscovery(None), MessageMapper())
start_executor = _DummyStartExecutor(handlers={str: lambda *_: None})
workflow = _DummyWorkflow(start_executor)
parsed = executor._parse_raw_workflow_input(workflow, "hi there")
assert parsed == "hi there"
if __name__ == "__main__":
# Simple test runner
async def run_tests():
@@ -9,9 +9,20 @@ from pathlib import Path
import pytest
from agent_framework_devui import DevServer
from agent_framework_devui._server import _extract_executor_message_types, _select_primary_input_type
from agent_framework_devui.models._openai_custom import AgentFrameworkExtraBody, AgentFrameworkRequest
class _StubExecutor:
"""Simple executor stub exposing handler metadata."""
def __init__(self, *, input_types=None, handlers=None):
if input_types is not None:
self.input_types = list(input_types)
if handlers is not None:
self._handlers = dict(handlers)
@pytest.fixture
def test_entities_dir():
"""Use the samples directory which has proper entity structure."""
@@ -97,6 +108,36 @@ def test_configuration():
assert server.ui_enabled
def test_extract_executor_message_types_prefers_input_types():
"""Input types property is used when available."""
stub = _StubExecutor(input_types=[str, dict])
types = _extract_executor_message_types(stub)
assert types == [str, dict]
def test_extract_executor_message_types_falls_back_to_handlers():
"""Handlers provide message metadata when input_types missing."""
stub = _StubExecutor(handlers={str: object(), int: object()})
types = _extract_executor_message_types(stub)
assert str in types
assert int in types
def test_select_primary_input_type_prefers_string_and_dict():
"""Primary type selection prefers user-friendly primitives."""
string_first = _select_primary_input_type([dict[str, str], str])
dict_first = _select_primary_input_type([dict[str, str]])
fallback = _select_primary_input_type([int, float])
assert string_first is str
assert dict_first is dict
assert fallback is int
if __name__ == "__main__":
# Simple test runner
async def run_tests():