feat(adapters/langgraph): compile ManagerWorkers into a hierarchical graph

docs/pyagentspec/source/agentspec/language_spec_nightly.rst

@@ -2270,6 +2270,9 @@ The ManagerWorkers has two main parameters:
   - Workers cannot interact with the end user directly.
   - When invoked, each worker can leverage its equipped tools to complete the assigned task and report the result back to the group manager.
 
+The ``ManagerWorkers`` input and output schemas must match those of its ``group_manager``.
+In particular, the two components must declare the same input property names and the same output property names,
+and each corresponding property must have the same type.
 
 Datastores
 ~~~~~~~~~~

docs/pyagentspec/source/agentspec_config_examples/howto_managerworkers.json

@@ -4,7 +4,16 @@
     "name": "managerworkers",
     "description": null,
     "metadata": {},
-    "inputs": [],
+    "inputs": [
+        {
+            "title": "customer_id",
+            "type": "string"
+        },
+        {
+            "title": "company_policy_info",
+            "type": "string"
+        }
+    ],
     "outputs": [],
     "group_manager": {
         "component_type": "Agent",
@@ -208,5 +217,5 @@
             "model_id": "llama-4-maverick"
         }
     },
-    "agentspec_version": "26.1.0"
+    "agentspec_version": "26.4.0"
 }

docs/pyagentspec/source/agentspec_config_examples/howto_managerworkers.yaml

@@ -9,7 +9,11 @@ id: 248045cb-ca6f-4d6f-9e22-d28b452a25da
 name: managerworkers
 description: null
 metadata: {}
-inputs: []
+inputs:
+- title: company_policy_info
+  type: string
+- title: customer_id
+  type: string
 outputs: []
 group_manager:
   component_type: Agent
@@ -225,4 +229,4 @@ $referenced_components:
     default_generation_parameters: null
     url: http://url.to.my.vllm.server/llama4mav
     model_id: llama-4-maverick
-agentspec_version: 26.1.0
+agentspec_version: 26.4.0

docs/pyagentspec/source/changelog.rst

@@ -71,6 +71,25 @@ New features
   Added ``DbmsVectorChainLlmConfig`` for configuring LLM requests executed
   through Oracle Database ``DBMS_VECTOR_CHAIN``.
 
+* **ManagerWorkers I/O update**
+
+  The ManagerWorkers language specification now requires its input and output
+  schemas to use the same property names and types as its group manager.
+  This allows exposing inputs required by the manager agent (e.g., the placeholders
+  in its system prompt).
+
+  We thank @spichen for the contribution!
+
+* **ManagerWorkers support in the LangGraph adapter**
+
+  The LangGraph adapter now converts ``ManagerWorkers`` into hierarchical graphs:
+  the group manager delegates tasks to workers and receives their results before
+  producing a final response. Nested ``ManagerWorkers`` can be used as workers.
+  The adapter also supports ``ManagerWorkers`` in Flow ``AgentNode`` steps with
+  one string output only.
+
+  We thank @spichen for the contribution!
+
 * **MCP tool retry policies**
 
   Added ``retry_policy`` support to ``MCPTool`` and ``MCPToolBox`` so runtimes can

pyagentspec/src/pyagentspec/adapters/langgraph/__init__.py

@@ -6,10 +6,13 @@
 
 """Agent Spec adapter for the LangGraph agentic framework."""
 
+from ._managerworkers import DELEGATE_TOOL_PREFIX, is_delegation_tool_name
 from .agentspecexporter import AgentSpecExporter
 from .agentspecloader import AgentSpecLoader
 
 __all__ = [
     "AgentSpecLoader",
     "AgentSpecExporter",
+    "DELEGATE_TOOL_PREFIX",
+    "is_delegation_tool_name",
 ]

pyagentspec/src/pyagentspec/adapters/langgraph/_execution_span.py

@@ -0,0 +1,87 @@
+# Copyright © 2025, 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.
+
+"""Wrap a compiled graph's ``stream``/``astream`` in an Agent Spec execution span.
+
+Agent, Flow and ManagerWorkers graphs all need the same wrapper: open a span, emit a
+Start event carrying the invocation inputs, yield the chunks the underlying stream
+produces while remembering the last state chunk, then emit an End event built from
+that final state. Only the span class and the two event payloads differ, so they come
+in as factories. ``invoke``/``ainvoke`` need no patch; they use ``stream``/``astream``
+internally.
+"""
+
+from typing import Any, AsyncGenerator, Callable, Dict, Generator
+
+from pyagentspec.adapters.langgraph._types import CompiledStateGraph
+
+
+def _invocation_inputs(kwargs: Dict[str, Any]) -> Dict[str, Any]:
+    """The ``input=`` argument of the patched call, or ``{}`` when it isn't a dict."""
+    inputs = kwargs.get("input", {})
+    return inputs if isinstance(inputs, dict) else {}
+
+
+def _final_state(chunk: Any, so_far: Any) -> Any:
+    """Fold one streamed chunk into the running "last state seen".
+
+    State arrives as ``(namespace, state)`` tuples; other chunk shapes aren't
+    something to build the End event from, so they leave the fold untouched.
+    """
+    return chunk[1] if isinstance(chunk, tuple) else so_far
+
+
+async def _async_or_sync(
+    async_call: Callable[..., Any], sync_call: Callable[..., Any], *args: Any
+) -> None:
+    """Await ``async_call``, falling back to ``sync_call`` for spans that don't
+    implement the async half of the tracing protocol."""
+    try:
+        await async_call(*args)
+    except NotImplementedError:
+        sync_call(*args)
+
+
+def patch_with_execution_span(
+    compiled_graph: CompiledStateGraph[Any, Any, Any],
+    make_span: Callable[[], Any],
+    make_start_event: Callable[[Dict[str, Any]], Any],
+    make_end_event: Callable[[Dict[str, Any]], Any],
+) -> None:
+    """Monkey-patch ``compiled_graph.stream`` / ``.astream`` to run inside a span.
+
+    ``make_start_event`` receives the invocation inputs; ``make_end_event``
+    receives the final state chunk (``{}`` when the run produced none).
+    """
+    original_stream = compiled_graph.stream
+    original_astream = compiled_graph.astream
+
+    def patched_stream(*args: Any, **kwargs: Any) -> Generator[Any, Any, None]:
+        with make_span() as span:
+            span.add_event(make_start_event(_invocation_inputs(kwargs)))
+            state: Any = {}
+            for chunk in original_stream(*args, **kwargs):
+                yield chunk
+                state = _final_state(chunk, state)
+            span.add_event(make_end_event(state if isinstance(state, dict) else {}))
+
+    async def patched_astream(*args: Any, **kwargs: Any) -> AsyncGenerator[Any, Any]:
+        span = make_span()
+        await _async_or_sync(span.start_async, span.start)
+        try:
+            start_event = make_start_event(_invocation_inputs(kwargs))
+            await _async_or_sync(span.add_event_async, span.add_event, start_event)
+            state: Any = {}
+            async for chunk in original_astream(*args, **kwargs):
+                yield chunk
+                state = _final_state(chunk, state)
+            end_event = make_end_event(state if isinstance(state, dict) else {})
+            await _async_or_sync(span.add_event_async, span.add_event, end_event)
+        finally:
+            await _async_or_sync(span.end_async, span.end)
+
+    compiled_graph.stream = patched_stream  # type: ignore[method-assign]
+    compiled_graph.astream = patched_astream  # type: ignore[method-assign]