feat(adapters/langgraph): allow ManagerWorkers members and flow steps in a Swarm

pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py

@@ -49,8 +49,10 @@
 from pyagentspec.flows.nodes import OutputMessageNode as AgentSpecOutputMessageNode
 from pyagentspec.flows.nodes import StartNode as AgentSpecStartNode
 from pyagentspec.flows.nodes import ToolNode as AgentSpecToolNode
+from pyagentspec.managerworkers import ManagerWorkers as AgentSpecManagerWorkers
 from pyagentspec.property import Property as AgentSpecProperty
 from pyagentspec.property import _empty_default as pyagentspec_empty_default
+from pyagentspec.swarm import Swarm as AgentSpecSwarm
 from pyagentspec.tracing.events import NodeExecutionEnd as AgentSpecNodeExecutionEnd
 from pyagentspec.tracing.events import NodeExecutionStart as AgentSpecNodeExecutionStart
 from pyagentspec.tracing.events.exception import ExceptionRaised
@@ -529,16 +531,95 @@ def _create_react_agent_with_given_input_values(
             )
         return self._agents_cache[system_prompt]
 
+    def _create_composite_graph_with_given_input_values(
+        self, inputs: Dict[str, Any]
+    ) -> CompiledStateGraph[Any, Any]:
+        """Compile the node's ``ManagerWorkers`` / ``Swarm`` into a runnable graph for
+        these inputs, cached by the rendered entry-agent prompt.
+
+        Such a graph runs over ``MessagesState`` and can't carry structured inputs to its
+        inner agents, so the node inputs are baked into the entry agent's ``system_prompt``
+        (the ``group_manager`` for a ManagerWorkers, the ``first_agent`` for a Swarm) and
+        the now-satisfied input ports dropped, so declared == inferred for the downstream
+        span re-validation. A non-Agent entry is passed through unchanged so the converter
+        raises its own clear error.
+        """
+        from pyagentspec.adapters.langgraph._langgraphconverter import (
+            AgentSpecToLangGraphConverter,
+        )
+
+        converter = AgentSpecToLangGraphConverter()
+        component = self.node.agent
+        if isinstance(component, AgentSpecManagerWorkers):
+            entry_agent = component.group_manager
+            convert = converter._manager_workers_convert_to_langgraph
+
+            def rebuild(rendered_entry: Any) -> Any:
+                return component.model_copy(update={"group_manager": rendered_entry, "inputs": []})
+
+        elif isinstance(component, AgentSpecSwarm):
+            entry_agent = component.first_agent
+            convert = converter._swarm_convert_to_langgraph
+
+            def rebuild(rendered_entry: Any) -> Any:
+                # The entry agent appears as first_agent and inside the relationship
+                # tuples, so swap it (matched by id) in both.
+                def _swap(agent: Any) -> Any:
+                    return rendered_entry if agent.id == entry_agent.id else agent
+
+                return component.model_copy(
+                    update={
+                        "first_agent": rendered_entry,
+                        "relationships": [
+                            (_swap(caller), _swap(recipient))
+                            for caller, recipient in component.relationships
+                        ],
+                        "inputs": [],
+                    }
+                )
+
+        else:
+            raise TypeError(
+                "_create_composite_graph_with_given_input_values requires a "
+                "ManagerWorkers or Swarm"
+            )
+
+        is_agent_entry = isinstance(entry_agent, AgentSpecAgent)
+        cache_key = (
+            render_template(entry_agent.system_prompt, inputs) if is_agent_entry else component.id
+        )
+        if cache_key not in self._agents_cache:
+            rendered = (
+                rebuild(entry_agent.model_copy(update={"system_prompt": cache_key, "inputs": []}))
+                if is_agent_entry
+                else component
+            )
+            self._agents_cache[cache_key] = convert(
+                rendered,
+                tool_registry=self.tool_registry,
+                converted_components=self.converted_components,
+                checkpointer=self.checkpointer,
+                config=self.config,
+                middleware=self._middleware,
+            )
+        return self._agents_cache[cache_key]
+
     def _prepare_agent_and_inputs(
         self, inputs: Dict[str, Any], messages: Messages
     ) -> Tuple[CompiledStateGraph[Any, Any], Dict[str, Any]]:
-        agent = self._create_react_agent_with_given_input_values(inputs)
         # LangGraph's agent expects at least one user message to drive execution.
         # When an AgentNode is used with a templated system prompt and no messages are provided
         # by the flow, the agent can crash. To avoid this, we artificially insert an empty
         # user message when the message list is empty.
         if not messages:
             messages = cast(Messages, [{"role": "user", "content": ""}])
+        if isinstance(self.node.agent, (AgentSpecManagerWorkers, AgentSpecSwarm)):
+            # A ManagerWorkers / Swarm flow step runs as a multi-agent graph over
+            # MessagesState: node inputs were baked into the entry agent's prompt, so the
+            # graph is driven by messages alone (not the agent's remaining_steps state).
+            graph = self._create_composite_graph_with_given_input_values(inputs)
+            return graph, {"messages": messages}
+        agent = self._create_react_agent_with_given_input_values(inputs)
         inputs |= {
             "remaining_steps": 20,  # Get the right number of steps left
             "messages": messages,

pyagentspec/src/pyagentspec/managerworkers.py

@@ -13,6 +13,7 @@
 from typing_extensions import Self
 
 from pyagentspec.agenticcomponent import AgenticComponent
+from pyagentspec.property import Property
 from pyagentspec.validation_helpers import model_validator_with_error_accumulation
 from pyagentspec.versioning import AgentSpecVersionEnum
 
@@ -65,6 +66,24 @@ class ManagerWorkers(AgenticComponent):
         default=AgentSpecVersionEnum.v25_4_2, init=False, exclude=True
     )
 
+    def _get_inferred_inputs(self) -> List[Property]:
+        """A ``ManagerWorkers`` exposes the inputs of its group manager.
+
+        The group manager is the component that drives the conversation and whose prompt
+        the run-time renders, so the manager-workers component accepts exactly the inputs
+        the group manager accepts (e.g. the ``{{placeholder}}`` inputs of an ``Agent``
+        group manager). Without this, the base default infers no inputs, so a
+        ``ManagerWorkers`` used as a flow ``AgentNode`` would expose no input ports and a
+        data-flow edge into it could not resolve.
+        """
+        group_manager = getattr(self, "group_manager", None)
+        return list(getattr(group_manager, "inputs", None) or [])
+
+    def _get_inferred_outputs(self) -> List[Property]:
+        """Outputs of the group manager; see :meth:`_get_inferred_inputs`."""
+        group_manager = getattr(self, "group_manager", None)
+        return list(getattr(group_manager, "outputs", None) or [])
+
     @model_validator_with_error_accumulation
     def _validate_one_or_more_workers(self) -> Self:
         if len(self.workers) == 0:

pyagentspec/src/pyagentspec/swarm.py

@@ -15,6 +15,7 @@
 
 from pyagentspec.agenticcomponent import AgenticComponent
 from pyagentspec.component import SerializeAsEnum
+from pyagentspec.property import Property
 from pyagentspec.validation_helpers import model_validator_with_error_accumulation
 from pyagentspec.versioning import AgentSpecVersionEnum
 
@@ -127,6 +128,24 @@ class Swarm(AgenticComponent):
         default=AgentSpecVersionEnum.v25_4_2, init=False, exclude=True
     )
 
+    def _get_inferred_inputs(self) -> List[Property]:
+        """A ``Swarm`` exposes the inputs of its entry agent (``first_agent``).
+
+        Symmetric with :meth:`ManagerWorkers._get_inferred_inputs`. The ``first_agent``
+        is the swarm's entry point — it interacts with the user before any handoff — so
+        the swarm component accepts exactly the inputs that agent accepts (e.g. the
+        ``{{placeholder}}`` inputs of an ``Agent`` entry's prompt). Without this, the base
+        default infers no inputs, so a flow ``AgentNode`` wrapping a swarm declares no
+        input ports and a ``DataFlowEdge`` into it fails to resolve at load.
+        """
+        first_agent = getattr(self, "first_agent", None)
+        return list(getattr(first_agent, "inputs", None) or [])
+
+    def _get_inferred_outputs(self) -> List[Property]:
+        """Outputs of the entry agent (``first_agent``); see :meth:`_get_inferred_inputs`."""
+        first_agent = getattr(self, "first_agent", None)
+        return list(getattr(first_agent, "outputs", None) or [])
+
     @model_validator(mode="before")
     def _raise_warning_if_handoff_is_bool(cls: Self, values: Any) -> Any:
         import warnings

pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py

@@ -0,0 +1,157 @@
+# Copyright © 2026 Oracle and/or its affiliates.
+#
+# This software is under the Apache License 2.0
+# (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
+# (UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
+
+"""A ManagerWorkers used as a flow step (AgentNode).
+
+Regression coverage for two coupled behaviours:
+  * ``ManagerWorkers._get_inferred_inputs`` exposes the group manager's inputs, so a
+    flow ``AgentNode`` wrapping a manager declares input ports and a ``DataFlowEdge``
+    into it resolves at load (previously: "node does not have any input property...").
+  * ``AgentNodeExecutor`` runs a ManagerWorkers node (previously: TypeError "can only
+    be used with AgentSpecAgent agents"), rendering the node inputs into the group
+    manager's prompt and returning its result.
+"""
+
+from pyagentspec.agent import Agent
+from pyagentspec.managerworkers import ManagerWorkers
+from pyagentspec.property import StringProperty
+
+
+def test_managerworkers_infers_inputs_from_group_manager_prompt() -> None:
+    """A ManagerWorkers exposes the group manager's prompt placeholders as inputs."""
+    llm = {"name": "m", "model_id": "fake", "url": "null"}
+    from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig
+
+    cfg = OpenAiCompatibleConfig(**llm)
+    manager = Agent(
+        name="manager",
+        llm_config=cfg,
+        system_prompt="Translate the following to Arabic:\n\n{{joke}}\n\nMake {{count}} variants.",
+    )
+    worker = Agent(name="worker", llm_config=cfg, system_prompt="You translate.")
+    mw = ManagerWorkers(name="mw", group_manager=manager, workers=[worker])
+
+    assert sorted(p.title for p in (mw.inputs or [])) == ["count", "joke"]
+
+
+def test_managerworkers_infers_outputs_from_group_manager() -> None:
+    """Symmetric with inputs: a ManagerWorkers exposes the group manager's outputs,
+    so a flow AgentNode wrapping it can wire its result downstream (or surface it as a
+    leaf)."""
+    from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig
+
+    cfg = OpenAiCompatibleConfig(name="m", model_id="fake", url="null")
+    answer = StringProperty(title="answer")
+    manager = Agent(
+        name="manager",
+        llm_config=cfg,
+        system_prompt="Answer the question.",
+        outputs=[answer],
+    )
+    worker = Agent(name="worker", llm_config=cfg, system_prompt="You help.")
+    mw = ManagerWorkers(name="mw", group_manager=manager, workers=[worker])
+
+    assert [p.title for p in (mw.outputs or [])] == ["answer"]
+
+
+def test_managerworkers_runs_as_a_flow_step_with_data_edge_inputs() -> None:
+    """A ManagerWorkers flow step loads (data edge resolves) and executes offline.
+
+    The model is stubbed (no real LLM, no delegation), so the manager produces a final
+    message and the manager graph routes straight to END. Asserts the flow both loads —
+    proving the manager node exposes the ``joke`` input the data edge targets — and runs,
+    surfacing the manager's answer as the node's single string output.
+    """
+    from unittest.mock import patch
+
+    from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+    from langchain_core.messages import AIMessage
+    from langchain_openai import ChatOpenAI
+    from langgraph.checkpoint.memory import MemorySaver
+
+    from pyagentspec.adapters.langgraph import AgentSpecLoader
+    from pyagentspec.adapters.langgraph._langgraphconverter import (
+        AgentSpecToLangGraphConverter,
+    )
+    from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge
+    from pyagentspec.flows.flow import Flow
+    from pyagentspec.flows.nodes import AgentNode, EndNode, StartNode
+    from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig
+
+    class _FakeModel(FakeMessagesListChatModel, ChatOpenAI):
+        pass
+
+    # Final message has no tool_calls → the manager routes to END without delegating.
+    fake_llm = _FakeModel(responses=[AIMessage(content="لماذا...")])
+
+    cfg = OpenAiCompatibleConfig(name="agent_llm", model_id="fake", url="null")
+    joke = StringProperty(title="joke")
+    translated = StringProperty(title="translated")
+
+    manager = Agent(
+        name="manager",
+        llm_config=cfg,
+        system_prompt="Translate the following to Arabic:\n\n{{joke}}",
+        outputs=[translated],
+    )
+    worker = Agent(name="worker", llm_config=cfg, system_prompt="You translate.")
+    mw = ManagerWorkers(name="translator", group_manager=manager, workers=[worker])
+    # The manager node exposes the group manager's `joke` input, and the single
+    # `translated` output (inherited from the group manager) for the leaf edge.
+    assert [p.title for p in (mw.inputs or [])] == ["joke"]
+
+    manager_node = AgentNode(name="manager_node", agent=mw)
+    start_node = StartNode(name="start", inputs=[joke])
+    end_node = EndNode(name="end", outputs=[translated])
+    flow = Flow(
+        name="flow",
+        start_node=start_node,
+        nodes=[start_node, manager_node, end_node],
+        control_flow_connections=[
+            ControlFlowEdge(name="start_to_node", from_node=start_node, to_node=manager_node),
+            ControlFlowEdge(name="node_to_end", from_node=manager_node, to_node=end_node),
+        ],
+        data_flow_connections=[
+            DataFlowEdge(
+                name="joke_edge",
+                source_node=start_node,
+                source_output=joke.title,
+                destination_node=manager_node,
+                destination_input=joke.title,
+            ),
+            DataFlowEdge(
+                name="translated_edge",
+                source_node=manager_node,
+                source_output=translated.title,
+                destination_node=end_node,
+                destination_input=translated.title,
+            ),
+        ],
+        outputs=[translated],
+    )
+
+    loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver())
+    with patch.object(
+        AgentSpecToLangGraphConverter,
+        "_llm_convert_to_langgraph",
+        autospec=True,
+        side_effect=lambda self_obj, llm_config, *a, **k: fake_llm,
+    ), patch.object(
+        FakeMessagesListChatModel,
+        "bind_tools",
+        new=lambda self_obj, *a, **k: self_obj,
+    ):
+        compiled = loader.load_component(flow)
+        result = compiled.invoke(
+            {
+                "inputs": {"joke": "Why did the car..."},
+                "messages": [{"role": "user", "content": ""}],
+            },
+            {"configurable": {"thread_id": "managerworkers-node"}},
+        )
+
+    assert "outputs" in result
+    assert result["outputs"]["translated"] == "لماذا..."