diff --git a/libs/langchain_v1/langchain/agents/middleware/_utils.py b/libs/langchain_v1/langchain/agents/middleware/_utils.py
new file mode 100644
index 0000000000..8063f39794
--- /dev/null
+++ b/libs/langchain_v1/langchain/agents/middleware/_utils.py
@@ -0,0 +1,23 @@
+"""Utility functions for middleware."""
+
+from langchain_core.messages import ContentBlock, SystemMessage
+
+
+def append_to_system_message(
+ system_message: SystemMessage | None,
+ text: str,
+) -> SystemMessage:
+ """Append text to a system message.
+
+ Args:
+ system_message: Existing system message or None.
+ text: Text to add to the system message.
+
+ Returns:
+ New SystemMessage with the text appended.
+ """
+ new_content: list[ContentBlock] = list(system_message.content_blocks) if system_message else []
+ if new_content:
+ text = f"\n\n{text}"
+ new_content.append({"type": "text", "text": text})
+ return SystemMessage(content_blocks=new_content)
diff --git a/libs/langchain_v1/langchain/agents/middleware/subagents.py b/libs/langchain_v1/langchain/agents/middleware/subagents.py
new file mode 100644
index 0000000000..1a6bace0e5
--- /dev/null
+++ b/libs/langchain_v1/langchain/agents/middleware/subagents.py
@@ -0,0 +1,817 @@
+"""Middleware for providing subagents to an agent via a `task` tool."""
+
+import contextlib
+import dataclasses
+import json
+from collections.abc import Awaitable, Callable, Generator, Sequence
+from typing import Annotated, Any, TypedDict, cast, get_args, get_origin, get_type_hints
+
+from typing_extensions import NotRequired
+
+from langchain.agents import create_agent
+from langchain.agents.middleware._utils import append_to_system_message
+from langchain.agents.middleware.shell_tool import DEFAULT_TOOL_DESCRIPTION
+from langchain.agents.middleware.types import (
+ AgentMiddleware,
+ ContextT,
+ ModelRequest,
+ ModelResponse,
+ PrivateStateAttr,
+ ResponseT,
+ StateT,
+)
+from langchain.agents.structured_output import ResponseFormat
+from langchain.tools import BaseTool
+from langchain_core.language_models import BaseChatModel
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from langchain_core.runnables import Runnable, RunnableConfig
+from langchain_core.tools import StructuredTool
+from langgraph.types import Command
+from langsmith.run_helpers import get_tracing_context, tracing_context
+from pydantic import BaseModel, Field
+
+from langchain.tools.tool_node import ToolRuntime
+
+class SubAgent(TypedDict):
+ """Specification for a raw subagent created by the `task` tool.
+
+ Required fields:
+ name: Unique identifier used as the task tool's `subagent_type`.
+ model: Model used to create the subagent. This may be a model instance or a
+ provider-qualified model name such as `'openai:gpt-5.5'`.
+
+ Optional fields:
+ description: Concise, action-oriented explanation of the subagent's role.
+ It is shown to the main agent when it decides whether to delegate. If
+ omitted, the subagent name is shown without a description.
+ system_prompt: Instructions for the subagent, including tool-use guidance
+ and expected response format. If omitted, no system prompt is provided.
+ tools: Tools available to the subagent. If omitted, the subagent is created
+ without tools.
+ middleware: Additional middleware used when creating the subagent.
+ response_format: Schema or structured-output strategy for the subagent.
+ Structured responses are serialized as JSON and returned to the main
+ agent instead of the subagent's final AI-message text.
+ """
+
+ name: str
+ """Unique identifier for the subagent."""
+
+ description: NotRequired[str]
+ """What this subagent does, shown to the main agent when delegating."""
+
+ system_prompt: NotRequired[str]
+ """Instructions for the subagent."""
+
+ tools: NotRequired[Sequence[BaseTool | Callable | dict[str, Any]]]
+ """Tools the subagent can use. Defaults to an empty sequence."""
+
+ model: str | BaseChatModel
+ """Override the main agent's model.
+
+ Use `'provider:model-name'` format.
+ """
+
+ middleware: NotRequired[list[AgentMiddleware]]
+ """Additional middleware for custom behavior."""
+
+ response_format: NotRequired[ResponseFormat[Any] | type | dict[str, Any]]
+ """Structured output response format for the subagent.
+
+ When specified, the subagent will produce a `structured_response` conforming
+ to the given schema. The structured response is JSON-serialized and returned
+ as the `ToolMessage` content to the parent agent, replacing the default
+ last-message extraction.
+
+ Accepted formats (from `langchain.agents.structured_output`):
+
+ - `ToolStrategy(schema)`: Use tool calling to extract structured output from the model.
+ - `ProviderStrategy(schema)`: Use the model provider's native structured output mode.
+ - `AutoStrategy(schema)`: Automatically select the best strategy.
+ - A bare Python `type`: A Pydantic `BaseModel` subclass, `dataclass`,
+ or `TypedDict` class.
+
+ Equivalent to `AutoStrategy(schema)`.
+ - `dict[str, Any]`: A JSON schema dictionary
+ (e.g., `{"type": "object", "properties": {...}, "required": [...]}`).
+
+ Example:
+ ```python
+ from pydantic import BaseModel
+
+ class Findings(BaseModel):
+ findings: str
+ confidence: float
+
+ analyzer: SubAgent = {
+ "name": "analyzer",
+ "description": "Analyzes data and returns structured findings",
+ "system_prompt": "Analyze the data and return your findings.",
+ "model": "openai:gpt-5.5",
+ "tools": [],
+ "response_format": Findings,
+ }
+ ```
+ """
+
+
+class CompiledSubAgent(TypedDict):
+ """A pre-compiled agent spec.
+
+ !!! note
+
+ The `runnable`'s state schema must include a 'messages' key.
+
+ This is required for the subagent to communicate results back to
+ the main agent.
+
+ !!! note
+
+ `CompiledSubAgent` runnables are used as provided. They do not
+ inherit `create_deep_agent(state_schema=...)`; if the runnable
+ needs custom state fields, compile it with a compatible state
+ schema yourself.
+
+ When the subagent completes, the parent reads the returned state:
+ if `structured_response` is non-`None`, it is JSON-serialized and used as
+ the `ToolMessage` content; otherwise, the last non-empty `AIMessage`
+ text is used.
+
+ Examples:
+ Using `create_agent` with `response_format`:
+
+ ```python
+ from pydantic import BaseModel
+ from langchain.agents import create_agent
+
+
+ class Findings(BaseModel):
+ summary: str
+ confidence: float
+
+
+ researcher: CompiledSubAgent = {
+ "name": "researcher",
+ "description": "Researches a topic and returns findings.",
+ "runnable": create_agent(
+ "openai:gpt-5.5",
+ tools=[], # your tools here
+ response_format=Findings,
+ ),
+ }
+ ```
+
+ Custom `langgraph` graph (write `structured_response` directly):
+
+ ```python
+ def node(state):
+ return {
+ "messages": [...],
+ "structured_response": Findings(summary="...", confidence=0.9),
+ }
+ ```
+ """
+
+ name: str
+ """Unique identifier for the subagent."""
+
+ description: str
+ """What this subagent does.
+
+ The main agent uses this to decide when to delegate.
+ """
+
+ runnable: Runnable
+ """A custom agent implementation.
+
+ Create a custom agent using either:
+
+ 1. LangChain's [`create_agent()`](https://docs.langchain.com/oss/python/langchain/quickstart)
+ 2. A custom graph using [`langgraph`](https://docs.langchain.com/oss/python/langgraph/quickstart)
+
+ If you're creating a custom graph, make sure the state schema includes
+ a 'messages' key. This is required for the subagent to communicate
+ results back to the main agent.
+ """
+
+
+_EXCLUDED_STATE_KEYS = {
+ "messages",
+ "todos",
+ "structured_response",
+}
+"""State keys that are excluded when passing state to subagents and when
+returning updates from subagents.
+
+When returning updates:
+
+1. The messages key is handled explicitly to ensure only the final message
+ is included
+2. The todos and `structured_response` keys are excluded as they do not have
+ a defined reducer and no clear meaning for returning them from a subagent
+ to the main agent.
+3. Agent-private fields on middleware state schemas are excluded from both
+ subagent output and subagent inputs.
+"""
+
+
+class TaskToolSchema(BaseModel):
+ """Input schema for the `task` tool."""
+
+ description: str = Field(
+ description=(
+ "A detailed description of the task for the subagent to perform autonomously. "
+ "Include all necessary context and specify the expected output format."
+ )
+ )
+
+ subagent_type: str = Field(description=("The type of subagent to use. Must be one of the available agent types listed in the tool description."))
+
+
+TASK_TOOL_DESCRIPTION = """Launch an ephemeral subagent to handle complex, multi-step independent tasks with isolated context windows.
+
+Available agent types and the tools they have access to:
+{available_agents}
+
+When using the Task tool, you must specify a subagent_type parameter to select which agent type to use.
+
+## Usage notes:
+1. Launch multiple agents concurrently whenever possible, to maximize performance; to do that, use a single message with multiple tool uses
+2. When the agent is done, it will return a single message back to you. The result returned by the agent is not visible to the user. To show the user the result, you should send a text message back to the user with a concise summary of the result.
+3. Each agent invocation is stateless. You will not be able to send additional messages to the agent, nor will the agent be able to communicate with you outside of its final report. Therefore, your prompt should contain a highly detailed task description for the agent to perform autonomously and you should specify exactly what information the agent should return back to you in its final and only message to you.
+4. The agent's outputs should generally be trusted
+5. Clearly tell the agent whether you expect it to create content, perform analysis, or just do research (search, file reads, web fetches, etc.), since it is not aware of the user's intent
+6. If the agent description mentions that it should be used proactively, then you should try your best to use it without the user having to ask for it first. Use your judgement.
+7. When only the general-purpose agent is provided, you should use it for all tasks. It is great for isolating context and token usage, and completing specific, complex tasks, as it has all the same capabilities as the main agent.
+
+### Example usage of the general-purpose agent:
+
+
+"general-purpose": use this agent for general purpose tasks, it has access to all tools as the main agent.
+
+
+
+User: "I want to conduct research on the accomplishments of Lebron James, Michael Jordan, and Kobe Bryant, and then compare them."
+Assistant: *Uses the task tool in parallel to conduct isolated research on each of the three players*
+Assistant: *Synthesizes the results of the three isolated research tasks and responds to the User*
+
+Research is a complex, multi-step task in it of itself.
+The research of each individual player is not dependent on the research of the other players.
+The assistant uses the task tool to break down the complex objective into three isolated tasks.
+Each research task only needs to worry about context and tokens about one player, then returns synthesized information about each player as the Tool Result.
+This means each research task can dive deep and spend tokens and context deeply researching each player, but the final result is synthesized information, and saves us tokens in the long run when comparing the players to each other.
+
+
+
+
+User: "Analyze a single large code repository for security vulnerabilities and generate a report."
+Assistant: *Launches a single `task` subagent for the repository analysis*
+Assistant: *Receives report and integrates results into final summary*
+
+Subagent is used to isolate a large, context-heavy task, even though there is only one. This prevents the main thread from being overloaded with details.
+If the user then asks followup questions, we have a concise report to reference instead of the entire history of analysis and tool calls, which is good and saves us time and money.
+
+
+
+
+User: "Schedule two meetings for me and prepare agendas for each."
+Assistant: *Calls the task tool in parallel to launch two `task` subagents (one per meeting) to prepare agendas*
+Assistant: *Returns final schedules and agendas*
+
+Tasks are simple individually, but subagents help silo agenda preparation.
+Each subagent only needs to worry about the agenda for one meeting.
+
+
+
+
+User: "I want to order a pizza from Dominos, order a burger from McDonald's, and order a salad from Subway."
+Assistant: *Calls tools directly in parallel to order a pizza from Dominos, a burger from McDonald's, and a salad from Subway*
+
+The assistant did not use the task tool because the objective is super simple and clear and only requires a few trivial tool calls.
+It is better to just complete the task directly and NOT use the `task` tool.
+
+
+
+### Example usage with custom agents:
+
+
+"content-reviewer": use this agent after you are done creating significant content or documents
+"greeting-responder": use this agent when to respond to user greetings with a friendly joke
+"research-analyst": use this agent to conduct thorough research on complex topics
+
+
+
+user: "Please write a function that checks if a number is prime"
+assistant: Sure let me write a function that checks if a number is prime
+assistant: First let me use the Write tool to write a function that checks if a number is prime
+assistant: I'm going to use the Write tool to write the following code:
+
+function isPrime(n) {{
+ if (n <= 1) return false
+ for (let i = 2; i * i <= n; i++) {{
+ if (n % i === 0) return false
+ }}
+ return true
+}}
+
+
+Since significant content was created and the task was completed, now use the content-reviewer agent to review the work
+
+assistant: Now let me use the content-reviewer agent to review the code
+assistant: Uses the Task tool to launch with the content-reviewer agent
+
+
+
+user: "Can you help me research the environmental impact of different renewable energy sources and create a comprehensive report?"
+
+This is a complex research task that would benefit from using the research-analyst agent to conduct thorough analysis
+
+assistant: I'll help you research the environmental impact of renewable energy sources. Let me use the research-analyst agent to conduct comprehensive research on this topic.
+assistant: Uses the Task tool to launch with the research-analyst agent, providing detailed instructions about what research to conduct and what format the report should take
+
+
+
+user: "Hello"
+
+Since the user is greeting, use the greeting-responder agent to respond with a friendly joke
+
+assistant: "I'm going to use the Task tool to launch with the greeting-responder agent"
+""" # noqa: E501
+
+TASK_SYSTEM_PROMPT = """## `task` (subagent spawner)
+
+You have access to a `task` tool to launch short-lived subagents that handle isolated tasks. These agents are ephemeral — they live only for the duration of the task and return a single result.
+
+When to use the task tool:
+
+- When a task is complex and multi-step, and can be fully delegated in isolation
+- When a task is independent of other tasks and can run in parallel
+- When a task requires focused reasoning or heavy token/context usage that would bloat the orchestrator thread
+- When sandboxing improves reliability (e.g. code execution, structured searches, data formatting)
+- When you only care about the output of the subagent, and not the intermediate steps (ex. performing a lot of research and then returned a synthesized report, performing a series of computations or lookups to achieve a concise, relevant answer.)
+
+Subagent lifecycle:
+
+1. **Spawn** → Provide clear role, instructions, and expected output
+2. **Run** → The subagent completes the task autonomously
+3. **Return** → The subagent provides a single structured result
+4. **Reconcile** → Incorporate or synthesize the result into the main thread
+
+When NOT to use the task tool:
+
+- If you need to see the intermediate reasoning or steps after the subagent has completed (the task tool hides them)
+- If the task is trivial (a few tool calls or simple lookup)
+- If delegating does not reduce token usage, complexity, or context switching
+- If splitting would add latency without benefit
+
+## Important Task Tool Usage Notes to Remember
+
+- Whenever possible, parallelize the work that you do. This is true for both tool_calls, and for tasks. Whenever you have independent steps to complete - make tool_calls, or kick off tasks (subagents) in parallel to accomplish them faster. This saves time for the user, which is incredibly important.
+- Remember to use the `task` tool to silo independent tasks within a multi-part objective.
+- You should use the `task` tool whenever you have a complex task that will take multiple steps, and is independent from other tasks that the agent needs to complete. These agents are highly competent and efficient.
+
+Available subagent types:
+
+{available_agents}""" # noqa: E501
+
+SUBAGENT_RESPONSE_FORMAT_CONFIG_KEY = "__deepagents_subagent_response_format"
+"""Configurable key used by task-tool callers to request dynamic response format."""
+
+def _get_subagent_response_format_config(
+ runtime: ToolRuntime,
+) -> ResponseFormat[Any] | type | dict[str, Any] | None:
+ """Return the response format carried in this task tool call's config."""
+ config = runtime.config
+ configurable = config.get("configurable") if isinstance(config, dict) else None
+ if not isinstance(configurable, dict):
+ return None
+ value = configurable.get(SUBAGENT_RESPONSE_FORMAT_CONFIG_KEY)
+ if value is None:
+ return None
+ return value
+
+
+
+@contextlib.contextmanager
+def _subagent_tracing_context() -> Generator[None, None, None]:
+ """Context manager that tags subagent runs with `ls_agent_type="subagent"`.
+
+ Sets `ls_agent_type` on the langsmith tracing context `metadata`, which is
+ propagated to LangSmith runs. This mirrors
+ langchain's `ls_agent_type="root"` tagging behavior.
+
+ Forwards all other current tracing-context fields (parent, client, tags,
+ etc.) unchanged so this wrapper does not clobber the enclosing context.
+ """
+ current = get_tracing_context()
+
+ merged_metadata = {**(current.get("metadata") or {}), "ls_agent_type": "subagent"}
+ # Pass every field from the current tracing context through to
+ # `tracing_context` so we don't accidentally clobber fields that may be
+ # added to langsmith in the future. The only change is `metadata`.
+
+ kwargs: dict[str, Any] = {**current, "metadata": merged_metadata}
+
+ with tracing_context(**kwargs):
+ yield
+
+
+def _has_marker(annotation: object, marker: object) -> bool:
+ origin = get_origin(annotation)
+ if origin is Annotated:
+ args = get_args(annotation)
+ return any(meta is marker for meta in args[1:])
+ if origin is not None:
+ return any(_has_marker(arg, marker) for arg in get_args(annotation))
+ return False
+
+def _private_state_field_names(state_schema: type[object]) -> frozenset[str]:
+ """Return fields annotated with `PrivateStateAttr` across state schemas."""
+ names: set[str] = set()
+ with contextlib.suppress(Exception):
+ hints = get_type_hints(state_schema, include_extras=True)
+ for name, annotation in hints.items():
+ if _has_marker(annotation, PrivateStateAttr):
+ names.add(name)
+ return frozenset(names)
+
+def _compile_sub_agent_spec(
+ spec: SubAgent,
+ *,
+ state_schema: type | None = None,
+) -> Runnable:
+ """Create a runnable agent from a raw `SubAgent` spec.
+
+ This is the shared entrypoint for the `create_agent` path used by
+ raw subagent specs. Pre-compiled `CompiledSubAgent` runnables are already
+ created by the caller and are handled separately by `SubAgentMiddleware`.
+
+ Args:
+ spec: Subagent spec to compile. Must specify `name` and `model`.
+ state_schema: Optional state schema for the raw subagent agent.
+
+ Returns:
+ Runnable agent ready for task-tool invocation.
+
+ Raises:
+ ValueError: If `spec` is missing `name` or `model`.
+ """
+ if "name" not in spec:
+ msg = "SubAgent must specify 'name'"
+ raise ValueError(msg)
+ if "model" not in spec:
+ msg = f"SubAgent '{spec['name']}' must specify 'model'"
+ raise ValueError(msg)
+
+ create_agent_kwargs: dict[str, Any] = {
+ "model": spec["model"],
+ "system_prompt": spec.get("system_prompt"),
+ "tools": list(spec.get("tools", [])),
+ "middleware": list(spec.get("middleware", [])),
+ "name": spec["name"],
+ "response_format": spec.get("response_format", None),
+ }
+ if state_schema is not None:
+ create_agent_kwargs["state_schema"] = state_schema
+ return create_agent(**create_agent_kwargs).with_config(
+ {
+ "metadata": {"lc_agent_name": spec["name"]},
+ "run_name": spec["name"],
+ }
+ )
+
+def _build_task_tool(
+ subagents: Sequence[SubAgent | CompiledSubAgent],
+ *,
+ description: str | None = TASK_TOOL_DESCRIPTION,
+ excluded_state_keys: frozenset[str] | None = None,
+ state_schema: type | None = None,
+) -> BaseTool:
+ """Create a task tool from subagent specs.
+
+ Args:
+ subagents: List of raw or compiled subagent specs.
+ description: Custom description for the task tool. If `None`,
+ uses default template. Supports `{available_agents}` placeholder.
+ excluded_state_keys: State keys marked with `PrivateStateAttr` that
+ should be stripped from input state before invoking subagents.
+ state_schema: State schema passed when compiling raw subagent specs.
+
+ Returns:
+ A StructuredTool that can invoke subagents by type.
+ """
+ if not subagents:
+ msg = "At least one subagent must be specified"
+ raise ValueError(msg)
+
+ subagents_by_name: dict[str, SubAgent | CompiledSubAgent] = {}
+ for spec in subagents:
+ name = spec.get("name")
+ if not isinstance(name, str) or not name:
+ msg = "Every subagent must specify a non-empty string 'name'"
+ raise ValueError(msg)
+ if name in subagents_by_name:
+ msg = f"Duplicate subagent name: {name!r}"
+ raise ValueError(msg)
+ if "runnable" in spec:
+ if spec["runnable"] is None:
+ msg = f"CompiledSubAgent {name!r} must specify 'runnable'"
+ raise ValueError(msg)
+ elif "model" not in spec:
+ msg = f"SubAgent {name!r} must specify 'model'"
+ raise ValueError(msg)
+ subagents_by_name[name] = spec
+
+ subagent_runnables: dict[str, Runnable] = {
+ name: (
+ cast("CompiledSubAgent", spec)["runnable"]
+ if "runnable" in spec
+ else _compile_sub_agent_spec(
+ cast("SubAgent", spec),
+ state_schema=state_schema,
+ )
+ )
+ for name, spec in subagents_by_name.items()
+ }
+
+ def _get_subagent_spec(subagent_type: str) -> SubAgent | CompiledSubAgent:
+ """Validates and returns a subagent spec from the provided input"""
+ try:
+ return subagents_by_name[subagent_type]
+ except KeyError:
+ allowed_types = ", ".join(f"`{name}`" for name in subagents_by_name)
+ msg = (
+ f"Cannot use subagent type `{subagent_type}` because it does "
+ f"not exist. The only allowed types are {allowed_types}"
+ )
+ raise ValueError(msg) from None
+
+ def _get_subagent_runnable(
+ spec: SubAgent | CompiledSubAgent,
+ runtime: ToolRuntime,
+ ) -> Runnable:
+ """Return the baseline runnable or compile a dynamic-format raw spec."""
+ response_format = _get_subagent_response_format_config(runtime)
+
+ if "runnable" in spec:
+ if response_format is not None:
+ msg = (
+ f'response_format cannot be used with compiled subagent "{spec["name"]}"; '
+ "dynamic schemas require a raw SubAgent spec."
+ )
+ raise ValueError(msg)
+ return subagent_runnables[spec["name"]].with_config(
+ {
+ "metadata": {"lc_agent_name": spec["name"]},
+ "run_name": spec["name"],
+ }
+ )
+
+ if response_format is None:
+ return subagent_runnables[spec["name"]]
+
+ # When a custom response format is provided,
+ dynamic_spec = cast(
+ "SubAgent",
+ {**spec, "response_format": response_format},
+ )
+ return _compile_sub_agent_spec(dynamic_spec, state_schema=state_schema)
+
+ def _filter_subagent_state(result: dict) -> dict:
+ """Filter out excluded keys from input given to the subagent"""
+ return {k: v for k, v in result.items() if k not in _EXCLUDED_STATE_KEYS and k not in excluded_state_keys}
+
+ def _extract_subagent_response(result: dict) -> str:
+ """Extract the response returned by a completed subagent run.
+
+ Structured responses take precedence and are serialized as JSON. Otherwise,
+ the text of the last non-empty `AIMessage` is returned.
+
+ Args:
+ result: Final state returned by the subagent runnable. The state must
+ contain a `messages` sequence and may contain a
+ `structured_response`.
+
+ Returns:
+ The serialized structured response or the final AI message text.
+
+ Raises:
+ ValueError: If `result` has no `messages` key or contains neither a
+ structured response nor an AI message with text.
+ """
+ if "messages" not in result:
+ error_msg = (
+ "CompiledSubAgent must return a dict containing a 'messages' key. "
+ "Custom StateGraphs used with CompiledSubAgent should include 'messages' "
+ "in their state schema to communicate results back to the main agent."
+ )
+ raise ValueError(error_msg)
+
+ # If the subagent has a structured response output, serialize it and return it
+ structured = result.get("structured_response")
+ if structured is not None:
+ if hasattr(structured, "model_dump_json"):
+ return structured.model_dump_json()
+ elif dataclasses.is_dataclass(structured) and not isinstance(structured, type):
+ return json.dumps(dataclasses.asdict(structured))
+ else:
+ return json.dumps(structured)
+
+ # Walk back to the last AIMessage with non-empty text.
+ for msg in reversed(result["messages"]):
+ if isinstance(msg, AIMessage):
+ text = msg.text.rstrip() if msg.text else ""
+ if text:
+ return text
+
+ error_msg = (
+ "SubAgent didn't return a usable response. Subagent runs must return either "
+ "contain a message list with vali text content in the last AIMessage, or a "
+ "structured response dict"
+ )
+ raise ValueError(error_msg)
+
+ def _format_subagent_update(result: Any, tool_call_id: str) -> Command:
+ return Command(
+ update={
+ **_filter_subagent_state(result),
+ "messages": [
+ ToolMessage(
+ content=_extract_subagent_response(result),
+ tool_call_id=tool_call_id
+ )
+ ]
+ }
+ )
+
+ def task(
+ description: str,
+ subagent_type: str,
+ runtime: ToolRuntime,
+ ) -> str | Command:
+ spec = _get_subagent_spec(subagent_type)
+ subagent = _get_subagent_runnable(spec, runtime)
+ subagent_input = _filter_subagent_state(runtime.state)
+ subagent_input["messages"] = [HumanMessage(content=description)]
+ # The parent's callbacks, tags and configurable reach the subagent
+ # automatically: langgraph's `ensure_config` seeds each run from the
+ # ambient parent config and (as of langgraph#7926) merges it per-key, so
+ # the subagent's bound config still wins collisions (e.g. `lc_agent_name`,
+ # `recursion_limit`) and parent metadata propagates (deepagents#3634).
+ # Forwarding those keys explicitly would double-count under the merge
+ # (e.g. duplicate `tags`), so we only stamp the subagent tracing tag.
+ subagent_config: RunnableConfig = {"configurable": {"ls_agent_type": "subagent"}}
+ with _subagent_tracing_context():
+ result = subagent.invoke(subagent_input, subagent_config)
+ if not runtime.tool_call_id:
+ return _extract_subagent_response(result)
+ return _format_subagent_update(result, runtime.tool_call_id)
+
+ async def atask(
+ description: str,
+ subagent_type: str,
+ runtime: ToolRuntime
+ ) -> str | Command:
+ spec = _get_subagent_spec(subagent_type)
+ subagent = _get_subagent_runnable(spec, runtime)
+ subagent_input = _filter_subagent_state(runtime.state)
+ subagent_input["messages"] = [HumanMessage(content=description)]
+ # The parent's callbacks, tags and configurable reach the subagent
+ # automatically: langgraph's `ensure_config` seeds each run from the
+ # ambient parent config and (as of langgraph#7926) merges it per-key, so
+ # the subagent's bound config still wins collisions (e.g. `lc_agent_name`,
+ # `recursion_limit`) and parent metadata propagates (deepagents#3634).
+ # Forwarding those keys explicitly would double-count under the merge
+ # (e.g. duplicate `tags`), so we only stamp the subagent tracing tag.
+ subagent_config: RunnableConfig = {"configurable": {"ls_agent_type": "subagent"}}
+ with _subagent_tracing_context():
+ result = await subagent.ainvoke(subagent_input, subagent_config)
+ if not runtime.tool_call_id:
+ return _extract_subagent_response(result)
+ return _format_subagent_update(result, runtime.tool_call_id)
+
+ if description is not None:
+ subagent_description_str = "\n".join(
+ f"- {s['name']}: {s.get('description', '')}" for s in subagents
+ )
+ description = description.format(available_agents=subagent_description_str)
+
+ return StructuredTool.from_function(
+ name="task",
+ func=task,
+ coroutine=atask,
+ description=description,
+ infer_schema=False,
+ args_schema=TaskToolSchema
+ )
+
+class SubAgentMiddleware(AgentMiddleware[StateT, ContextT, ResponseT]):
+ """Middleware for providing subagents to an agent via a `task` tool.
+
+ This middleware adds a `task` tool to the agent that can be used
+ to invoke subagents.
+
+ Subagents are useful for handling complex tasks that require multiple steps,
+ or tasks that require a lot of context to resolve.
+
+ A chief benefit of subagents is that they can handle multi-step tasks,
+ and then return a clean, concise response to the main agent.
+
+ Subagents are also great for different domains of expertise that require
+ a narrower subset of tools and focus.
+
+ Args:
+ subagents: Subagents available to the `task` tool. Raw `SubAgent` specs must
+ define `name` and `model`; `description`, `system_prompt`, and `tools`
+ are optional. Precompiled specs must define `name`, `description`, and
+ `runnable`.
+ Names must be unique because the task's `subagent_type` selects a spec by
+ name.
+ system_prompt: Instructions appended to the main agent's system prompt.
+ The `{available_agents}` placeholder is replaced with the configured
+ subagent names and descriptions.
+ tool_description: Description for the `task` tool. The
+ `{available_agents}` placeholder is replaced with the configured
+ subagent names and descriptions.
+ state_schema: Main-agent state schema passed to raw subagent compilation.
+ Fields annotated with `PrivateStateAttr` are excluded from state passed to
+ subagents and from state updates returned to the main agent. Compiled
+ subagents retain the schema of their caller-provided runnables.
+
+ Example:
+ ```python
+ from langchain.agents import create_agent
+ from langchain.agents.middleware.subagents import SubAgentMiddleware
+
+ agent = create_agent(
+ "openai:gpt-5.5",
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "researcher",
+ "description": "Research a topic and summarize findings.",
+ "system_prompt": "You are a research specialist.",
+ "model": "openai:gpt-5.5",
+ "tools": [],
+ }
+ ],
+ )
+ ],
+ )
+ ```
+ """
+ def __init__(
+ self,
+ *,
+ subagents: Sequence[SubAgent | CompiledSubAgent],
+ system_prompt: str | None = TASK_SYSTEM_PROMPT,
+ tool_description: str | None = DEFAULT_TOOL_DESCRIPTION,
+ state_schema: type | None = None,
+ ) -> None:
+ super().__init__()
+
+ subagent_description_str = "\n".join(
+ f"- {s.get('name', '')}: {s.get('description', '')}" for s in subagents
+ )
+ self._system_prompt = (
+ system_prompt.format(available_agents=subagent_description_str)
+ if system_prompt is not None
+ else None
+ )
+
+ self.tools = [
+ _build_task_tool(
+ subagents,
+ description=tool_description,
+ excluded_state_keys=(
+ _private_state_field_names(state_schema)
+ if state_schema is not None
+ else frozenset()
+ ),
+ state_schema=state_schema,
+ )
+ ]
+
+ def wrap_model_call(
+ self,
+ request: ModelRequest[ContextT],
+ handler: Callable[[ModelRequest[ContextT]], ModelResponse[ResponseT]],
+ ) -> ModelResponse[ResponseT]:
+ """Update the system message to include instructions on using subagents."""
+ if self._system_prompt is not None:
+ new_system_message = append_to_system_message(request.system_message, self._system_prompt)
+ return handler(request.override(system_message=new_system_message))
+ return handler(request)
+
+
+ async def awrap_model_call(
+ self,
+ request: ModelRequest[ContextT],
+ handler: Callable[[ModelRequest[ContextT]], Awaitable[ModelResponse[ResponseT]]],
+ ) -> ModelResponse[ResponseT]:
+ """Update the system message to include instructions on using subagents."""
+ if self._system_prompt is not None:
+ new_system_message = append_to_system_message(request.system_message, self._system_prompt)
+ return await handler(request.override(system_message=new_system_message))
+ return await handler(request)
diff --git a/libs/langchain_v1/tests/integration_tests/agents/middleware/test_subagent_middleware.py b/libs/langchain_v1/tests/integration_tests/agents/middleware/test_subagent_middleware.py
new file mode 100644
index 0000000000..caba454f23
--- /dev/null
+++ b/libs/langchain_v1/tests/integration_tests/agents/middleware/test_subagent_middleware.py
@@ -0,0 +1,262 @@
+import json
+from typing import ClassVar
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.tools import tool
+from pydantic import BaseModel, Field
+
+from langchain.agents import create_agent
+from langchain.agents.middleware import AgentMiddleware
+from langchain.agents.middleware.subagents import SubAgent, SubAgentMiddleware
+from langchain.agents.structured_output import ToolStrategy
+
+
+@tool
+def get_weather(city: str) -> str:
+ """Get the weather in a city."""
+ return f"The weather in {city} is sunny."
+
+
+class WeatherMiddleware(AgentMiddleware):
+ tools: ClassVar = [get_weather]
+
+
+GENERAL_PURPOSE_SUBAGENT: SubAgent = {
+ "name": "general-purpose",
+ "description": "General-purpose agent for answering weather questions.",
+ "system_prompt": "Use the available tools to answer the user's weather question.",
+}
+
+
+def assert_expected_subgraph_actions(expected_tool_calls, agent, inputs):
+ current_idx = 0
+ for update in agent.stream(
+ inputs,
+ subgraphs=True,
+ stream_mode="updates",
+ ):
+ if "model" in update[1]:
+ ai_message = update[1]["model"]["messages"][-1]
+ tool_calls = ai_message.tool_calls
+ for tool_call in tool_calls:
+ if tool_call["name"] == expected_tool_calls[current_idx]["name"]:
+ if "model" in expected_tool_calls[current_idx]:
+ # Providers may return date-suffixed names
+ expected_model = expected_tool_calls[current_idx]["model"]
+ actual_model = ai_message.response_metadata["model_name"]
+ assert actual_model == expected_model or actual_model.startswith(expected_model + "-"), (
+ f"Expected model {expected_model!r}, got {actual_model!r}"
+ )
+ for arg in expected_tool_calls[current_idx]["args"]:
+ assert arg in tool_call["args"]
+ assert tool_call["args"][arg] == expected_tool_calls[current_idx]["args"][arg]
+ current_idx += 1
+ assert current_idx == len(expected_tool_calls)
+
+
+@pytest.mark.requires("langchain_anthropic", "langchain_openai")
+class TestSubagentMiddleware:
+ """Integration tests for the SubagentMiddleware class."""
+
+ def test_general_purpose_subagent(self):
+ agent = create_agent(
+ model="claude-sonnet-4-6",
+ system_prompt="Use the general-purpose subagent to get the weather in a city.",
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ {
+ **GENERAL_PURPOSE_SUBAGENT,
+ "model": "claude-sonnet-4-6",
+ "tools": [get_weather],
+ }
+ ],
+ )
+ ],
+ )
+ assert "task" in agent.nodes["tools"].bound._tools_by_name
+ response = agent.invoke({"messages": [HumanMessage(content="What is the weather in Tokyo?")]})
+ assert response["messages"][1].tool_calls[0]["name"] == "task"
+ assert response["messages"][1].tool_calls[0]["args"]["subagent_type"] == "general-purpose"
+
+ def test_defined_subagent_tool_calls(self):
+ agent = create_agent(
+ model="claude-sonnet-4-6",
+ system_prompt="Use the task tool to call a subagent.",
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "weather",
+ "description": "This subagent can get weather in cities.",
+ "system_prompt": "Use the get_weather tool to get the weather in a city.",
+ "model": "claude-sonnet-4-6",
+ "tools": [get_weather],
+ }
+ ],
+ )
+ ],
+ )
+ expected_tool_calls = [
+ {"name": "task", "args": {"subagent_type": "weather"}},
+ {"name": "get_weather", "args": {}},
+ ]
+ assert_expected_subgraph_actions(
+ expected_tool_calls,
+ agent,
+ {"messages": [HumanMessage(content="What is the weather in Tokyo?")]},
+ )
+
+ def test_defined_subagent_custom_model(self):
+ agent = create_agent(
+ model="claude-sonnet-4-6",
+ system_prompt="Use the task tool to call a subagent.",
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "weather",
+ "description": "This subagent can get weather in cities.",
+ "system_prompt": "Use the get_weather tool to get the weather in a city.",
+ "tools": [get_weather],
+ "model": "gpt-5.4",
+ }
+ ],
+ )
+ ],
+ )
+ expected_tool_calls = [
+ {
+ "name": "task",
+ "args": {"subagent_type": "weather"},
+ "model": "claude-sonnet-4-6",
+ },
+ {"name": "get_weather", "args": {}, "model": "gpt-5.4"},
+ ]
+ assert_expected_subgraph_actions(
+ expected_tool_calls,
+ agent,
+ {"messages": [HumanMessage(content="What is the weather in Tokyo?")]},
+ )
+
+ def test_defined_subagent_custom_middleware(self):
+ agent = create_agent(
+ model="claude-sonnet-4-6",
+ system_prompt="Use the task tool to call a subagent.",
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "weather",
+ "description": "This subagent can get weather in cities.",
+ "system_prompt": "Use the get_weather tool to get the weather in a city.",
+ "tools": [], # No tools, only in middleware
+ "model": "gpt-5.4",
+ "middleware": [WeatherMiddleware()],
+ }
+ ],
+ )
+ ],
+ )
+ expected_tool_calls = [
+ {
+ "name": "task",
+ "args": {"subagent_type": "weather"},
+ "model": "claude-sonnet-4-6",
+ },
+ {"name": "get_weather", "args": {}, "model": "gpt-5.4"},
+ ]
+ assert_expected_subgraph_actions(
+ expected_tool_calls,
+ agent,
+ {"messages": [HumanMessage(content="What is the weather in Tokyo?")]},
+ )
+
+ def test_defined_subagent_custom_runnable(self):
+ custom_subagent = create_agent(
+ model="gpt-5.4",
+ system_prompt="Use the get_weather tool to get the weather in a city.",
+ tools=[get_weather],
+ )
+ agent = create_agent(
+ model="claude-sonnet-4-6",
+ system_prompt="Use the task tool to call a subagent.",
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "weather",
+ "description": "This subagent can get weather in cities.",
+ "runnable": custom_subagent,
+ }
+ ],
+ )
+ ],
+ )
+ expected_tool_calls = [
+ {
+ "name": "task",
+ "args": {"subagent_type": "weather"},
+ "model": "claude-sonnet-4-6",
+ },
+ {"name": "get_weather", "args": {}, "model": "gpt-5.4"},
+ ]
+ assert_expected_subgraph_actions(
+ expected_tool_calls,
+ agent,
+ {"messages": [HumanMessage(content="What is the weather in Tokyo?")]},
+ )
+
+ def test_subagent_response_format_serialized_as_json(self):
+ """Test that subagent responseFormat produces JSON-serialized ToolMessage content.
+
+ Verifies the end-to-end flow when `response_format` is set directly on a
+ `SubAgent` spec: the subagent's `structured_response` is JSON-serialized
+ into the ToolMessage content returned to the parent agent.
+ """
+
+ class SubagentFindings(BaseModel):
+ findings: str = Field(description="The findings")
+ confidence: float = Field(description="Confidence score")
+ summary: str = Field(description="Brief summary")
+
+ agent = create_agent(
+ model="claude-sonnet-4-20250514",
+ system_prompt="You are an orchestrator. Always delegate tasks to the appropriate subagent via the task tool.",
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "foo",
+ "description": "Call this when the user says 'foo'",
+ "system_prompt": "You are a foo agent",
+ "model": "claude-haiku-4-5",
+ "tools": [],
+ "response_format": ToolStrategy(schema=SubagentFindings),
+ },
+ ],
+ )
+ ],
+ )
+
+ result = agent.invoke(
+ {"messages": [HumanMessage(content="foo - tell me how confident you are that pineapple belongs on pizza")]},
+ {"recursion_limit": 100},
+ )
+
+ agent_messages = [msg for msg in result["messages"] if isinstance(msg, AIMessage)]
+ tool_calls = [tc for msg in agent_messages for tc in (msg.tool_calls or [])]
+ assert any(tc["name"] == "task" and tc["args"].get("subagent_type") == "foo" for tc in tool_calls)
+
+ task_tool_messages = [msg for msg in result["messages"] if msg.type == "tool" and msg.name == "task"]
+ assert len(task_tool_messages) > 0
+
+ task_tool_message = task_tool_messages[0]
+ parsed = json.loads(task_tool_message.content)
+ assert "findings" in parsed
+ assert "confidence" in parsed
+ assert "summary" in parsed
+ assert isinstance(parsed["findings"], str)
+ assert isinstance(parsed["confidence"], (int, float))
+ assert isinstance(parsed["summary"], str)
diff --git a/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_subagent_init.py b/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_subagent_init.py
new file mode 100644
index 0000000000..0c5beae55c
--- /dev/null
+++ b/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_subagent_init.py
@@ -0,0 +1,455 @@
+"""Unit tests for SubAgentMiddleware initialization and configuration."""
+
+import json
+from typing import Any, get_type_hints
+
+from langchain.agents.middleware.subagents import SUBAGENT_RESPONSE_FORMAT_CONFIG_KEY, SubAgent, SubAgentMiddleware
+import pytest
+from langchain.agents import create_agent
+from langchain.agents.structured_output import AutoStrategy
+from langchain.tools import ToolRuntime
+from langchain_core.callbacks import CallbackManagerForLLMRun
+from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
+from langchain_core.outputs import ChatResult
+from langchain_core.runnables import RunnableLambda
+from langchain_core.tools import tool
+from langgraph.graph import START, MessagesState, StateGraph
+from tests.unit_tests.chat_model import GenericFakeChatModel
+
+
+@tool
+def get_weather(city: str) -> str:
+ """Get the weather in a city."""
+ return f"The weather in {city} is sunny."
+
+GENERAL_PURPOSE_SUBAGENT: SubAgent = {
+ "name": "general-purpose",
+ "description": "General-purpose agent for answering weather questions.",
+ "system_prompt": "Use the available tools to answer the user's weather question.",
+}
+
+
+class _DynamicStructuredOutputModel(GenericFakeChatModel):
+ """Fake model that calls whichever structured-output tool is bound."""
+
+ def _generate(
+ self,
+ messages: list[BaseMessage],
+ stop: list[str] | None = None,
+ run_manager: CallbackManagerForLLMRun | None = None,
+ **kwargs: Any,
+ ) -> ChatResult:
+ tool_name = getattr(self.tools[-1], "name", None)
+ if not isinstance(tool_name, str):
+ msg = "Expected a structured-output tool to be bound"
+ raise TypeError(msg)
+ self.messages = iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": tool_name,
+ "args": {
+ "name": "Maya Thornton",
+ "age": 29,
+ "city": "Portland",
+ },
+ "id": "call_payload",
+ "type": "tool_call",
+ }
+ ],
+ )
+ ]
+ )
+ return super()._generate(
+ messages,
+ stop=stop,
+ run_manager=run_manager,
+ **kwargs,
+ )
+
+
+class TestSubagentMiddlewareInit:
+ """Tests for SubAgentMiddleware initialization that don't require LLM invocation."""
+
+ @pytest.fixture(autouse=True)
+ def set_env_vars(self, monkeypatch: pytest.MonkeyPatch) -> None:
+ """Set dummy API key for model initialization."""
+ monkeypatch.setenv("OPENAI_API_KEY", "test-key")
+
+ def test_subagent_middleware_init(self) -> None:
+ """Test basic SubAgentMiddleware initialization with general-purpose subagent."""
+ middleware = SubAgentMiddleware(
+ subagents=[
+ {
+ **GENERAL_PURPOSE_SUBAGENT,
+ "model": "gpt-5.4-mini",
+ "tools": [],
+ }
+ ],
+ )
+ assert middleware is not None
+ assert "Available subagent types:" in middleware.system_prompt
+ assert len(middleware.tools) == 1
+ assert middleware.tools[0].name == "task"
+
+ def test_task_tool_compiles_dynamic_response_format_for_declarative_subagent(self) -> None:
+ """Dynamic schemas are present when declarative subagent variants compile."""
+ schema = {
+ "type": "object",
+ "properties": {
+ "name": {"type": "string"},
+ "age": {"type": "integer"},
+ "city": {"type": "string"},
+ },
+ "required": ["name", "age", "city"],
+ }
+ response_format = AutoStrategy(schema)
+ middleware = SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "worker",
+ "description": "Does work.",
+ "system_prompt": "Return structured data.",
+ "model": _DynamicStructuredOutputModel(messages=iter(())),
+ "tools": [],
+ }
+ ],
+ system_prompt=None,
+ )
+ task_tool = middleware.tools[0]
+ runtime = ToolRuntime(
+ state={},
+ context={},
+ config={
+ "configurable": {
+ SUBAGENT_RESPONSE_FORMAT_CONFIG_KEY: response_format,
+ }
+ },
+ stream_writer=lambda _chunk: None,
+ tools=[task_tool],
+ tool_call_id="call_worker",
+ store=None,
+ )
+
+ result = task_tool.func(
+ description="Make a person.",
+ subagent_type="worker",
+ runtime=runtime,
+ )
+
+ assert json.loads(result.update["messages"][0].content) == {
+ "name": "Maya Thornton",
+ "age": 29,
+ "city": "Portland",
+ }
+
+ def test_task_tool_rejects_response_format_for_compiled_subagent(self) -> None:
+ """Dynamic schemas require a declarative spec to compile a variant."""
+ schema = {
+ "type": "object",
+ "properties": {"ok": {"type": "boolean"}},
+ "required": ["ok"],
+ }
+ response_format = AutoStrategy(schema)
+
+ runnable = RunnableLambda(lambda _state, _config: (_ for _ in ()).throw(AssertionError("compiled runnable should not be invoked")))
+
+ task_tool = _build_task_tool(
+ [
+ {
+ "name": "worker",
+ "description": "Does work.",
+ "runnable": runnable,
+ }
+ ]
+ )
+ runtime = ToolRuntime(
+ state={},
+ context={},
+ config={
+ "configurable": {
+ SUBAGENT_RESPONSE_FORMAT_CONFIG_KEY: response_format,
+ }
+ },
+ stream_writer=lambda _chunk: None,
+ tools=[task_tool],
+ tool_call_id="call_worker",
+ store=None,
+ )
+
+ with pytest.raises(
+ ValueError,
+ match='response_schema cannot be used with compiled subagent "worker"',
+ ):
+ task_tool.func(
+ description="Do work.",
+ subagent_type="worker",
+ runtime=runtime,
+ )
+
+ def test_subagent_middleware_with_custom_subagent(self) -> None:
+ """Test SubAgentMiddleware initialization with a custom subagent."""
+ middleware = SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "weather",
+ "description": "Weather subagent",
+ "system_prompt": "Get weather.",
+ "model": "gpt-5.4-mini",
+ "tools": [get_weather],
+ }
+ ],
+ )
+ assert middleware is not None
+ # System prompt includes TASK_SYSTEM_PROMPT plus available subagent types
+ assert middleware.system_prompt.startswith(TASK_SYSTEM_PROMPT)
+ assert "weather" in middleware.system_prompt
+
+ def test_subagent_middleware_custom_system_prompt(self) -> None:
+ """Test SubAgentMiddleware with a custom system prompt."""
+ middleware = SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "weather",
+ "description": "Weather subagent",
+ "system_prompt": "Get weather.",
+ "model": "gpt-5.4-mini",
+ "tools": [],
+ }
+ ],
+ system_prompt="Use the task tool to call a subagent.",
+ )
+ assert middleware is not None
+ # Custom system prompt plus available subagent types
+ assert middleware.system_prompt.startswith("Use the task tool to call a subagent.")
+
+ def test_requires_subagents(self) -> None:
+ """Test that at least one subagent is required."""
+ with pytest.raises(ValueError, match="At least one subagent"):
+ SubAgentMiddleware(
+ subagents=[],
+ )
+
+ def test_subagent_requires_model(self) -> None:
+ """Test that subagents must specify model."""
+ with pytest.raises(ValueError, match="must specify 'model'"):
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "test",
+ "description": "Test",
+ "system_prompt": "Test.",
+ "tools": [],
+ # Missing "model"
+ }
+ ],
+ )
+
+ def test_subagent_requires_tools(self) -> None:
+ """Test that subagents must specify tools."""
+ with pytest.raises(ValueError, match="must specify 'tools'"):
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "test",
+ "description": "Test",
+ "system_prompt": "Test.",
+ "model": "gpt-5.4-mini",
+ # Missing "tools"
+ }
+ ],
+ )
+
+ def _make_echo_graph(self) -> object:
+ """Build a minimal MessagesState graph for use in CompiledSubAgent tests."""
+
+ def echo_node(_state: MessagesState) -> dict:
+ return {"messages": [AIMessage(content="hello")]}
+
+ builder = StateGraph(MessagesState)
+ builder.add_node("echo", echo_node)
+ builder.add_edge(START, "echo")
+ return builder.compile()
+
+ def _task_runtime(self, task_tool: object, tool_call_id: str) -> ToolRuntime:
+ return ToolRuntime(
+ state={},
+ context={},
+ config={"configurable": {}},
+ stream_writer=lambda _chunk: None,
+ tools=[task_tool],
+ tool_call_id=tool_call_id,
+ store=None,
+ )
+
+ def test_compiled_subagent_name_propagated_via_config(self) -> None:
+ """CompiledSubAgent.name is forwarded into metadata.lc_agent_name and run_name."""
+ configs: list[dict[str, object]] = []
+
+ class _Runnable:
+ def with_config(self, config: dict[str, object]) -> "_Runnable":
+ configs.append(config)
+ return self
+
+ def invoke(
+ self,
+ state: dict[str, object],
+ config: object = None,
+ ) -> dict[str, object]:
+ del state, config
+ return {"messages": [AIMessage(content="hello")]}
+
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "my-subagent",
+ "description": "A custom subagent",
+ "runnable": _Runnable(),
+ }
+ ],
+ )
+
+ assert configs == [
+ {
+ "metadata": {"lc_agent_name": "my-subagent"},
+ "run_name": "my-subagent",
+ }
+ ]
+
+ def test_compiled_subagent_does_not_mutate_original_runnable(self) -> None:
+ """Task-tool setup must not mutate the original runnable."""
+ graph = self._make_echo_graph()
+ original_config = getattr(graph, "config", None)
+
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "my-subagent",
+ "description": "A custom subagent",
+ "runnable": graph,
+ }
+ ],
+ )
+
+ assert graph.config == original_config, "Original runnable was mutated; use with_config instead of attribute assignment"
+
+ def test_same_runnable_reused_across_multiple_subagents(self) -> None:
+ """Same runnable registered under two different names must not cross-contaminate configs."""
+
+ class _Runnable:
+ def __init__(self, config: dict[str, object] | None = None) -> None:
+ self.config = config
+
+ def with_config(self, config: dict[str, object]) -> "_Runnable":
+ return _Runnable(config)
+
+ def invoke(
+ self,
+ state: dict[str, object],
+ config: object = None,
+ ) -> dict[str, object]:
+ del state, config
+ if self.config is None:
+ msg = "Expected configured runnable clone"
+ raise AssertionError(msg)
+ return {"messages": [AIMessage(content=str(self.config["run_name"]))]}
+
+ graph = _Runnable()
+
+ middleware = SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "agent-alpha",
+ "description": "First binding",
+ "runnable": graph,
+ },
+ {
+ "name": "agent-beta",
+ "description": "Second binding",
+ "runnable": graph,
+ },
+ ],
+ )
+
+ task_tool = middleware.tools[0]
+ alpha = task_tool.func(
+ description="Do alpha work.",
+ subagent_type="agent-alpha",
+ runtime=self._task_runtime(task_tool, "call_alpha"),
+ )
+ beta = task_tool.func(
+ description="Do beta work.",
+ subagent_type="agent-beta",
+ runtime=self._task_runtime(task_tool, "call_beta"),
+ )
+
+ assert alpha.update["messages"][0].content == "agent-alpha"
+ assert beta.update["messages"][0].content == "agent-beta"
+ assert graph.config is None
+
+ def test_middleware_delegates_to_create_sub_agent(self, monkeypatch: pytest.MonkeyPatch) -> None:
+ """Middleware should use the shared entrypoint for declarative subagents."""
+ graph = self._make_echo_graph()
+ calls: list[tuple[object, type | None]] = []
+
+ class CustomState(MessagesState):
+ pass
+
+ def fake_create_sub_agent(
+ spec: object,
+ *,
+ state_schema: type | None = None,
+ response_format: object = None,
+ ) -> object:
+ del response_format
+ calls.append((spec, state_schema))
+ return graph
+
+ monkeypatch.setattr("deepagents.middleware.subagents.create_sub_agent", fake_create_sub_agent)
+
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "agent-alpha",
+ "description": "First binding",
+ "system_prompt": "Work on the task.",
+ "model": "test-model",
+ "tools": [],
+ },
+ ],
+ state_schema=CustomState,
+ )
+
+ assert len(calls) == 1
+ assert calls[0][1] is CustomState
+
+ def test_multiple_subagents_with_interrupt_on(self) -> None:
+ """Test creating agent with multiple subagents that have interrupt_on configured."""
+ agent = create_agent(
+ model="claude-sonnet-4-6",
+ system_prompt="Use the task tool to call subagents.",
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ {
+ "name": "subagent1",
+ "description": "First subagent.",
+ "system_prompt": "You are subagent 1.",
+ "model": "claude-sonnet-4-6",
+ "tools": [get_weather],
+ },
+ {
+ "name": "subagent2",
+ "description": "Second subagent.",
+ "system_prompt": "You are subagent 2.",
+ "model": "claude-sonnet-4-6",
+ "tools": [get_weather],
+ },
+ ],
+ )
+ ],
+ )
+ # This would error if the middleware was accumulated incorrectly
+ assert agent is not None
diff --git a/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_subagents.py b/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_subagents.py
new file mode 100644
index 0000000000..7a278a2766
--- /dev/null
+++ b/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_subagents.py
@@ -0,0 +1,2284 @@
+"""Tests for sub-agent middleware functionality.
+
+This module contains tests for the subagent system, focusing on how subagents
+are invoked, how they return results, and how state is managed between parent
+and child agents.
+"""
+
+import dataclasses
+import json
+import re
+import uuid
+from collections.abc import Callable, Iterator, Sequence
+from typing import Annotated, Any, TypedDict, cast
+
+from typing_extensions import override
+from unittest.mock import MagicMock
+
+from langchain.agents.middleware.subagents import CompiledSubAgent, SubAgent, SubAgentMiddleware
+import pytest
+from langchain.agents import create_agent
+from langchain.agents.middleware import TodoListMiddleware
+from langchain.agents.middleware.types import AgentMiddleware, AgentState, PrivateStateAttr
+from langchain.agents.structured_output import ToolStrategy
+from langchain.tools import ToolRuntime
+from langchain_core.callbacks import BaseCallbackHandler, CallbackManagerForLLMRun
+from langchain_core.language_models import BaseChatModel, LanguageModelInput
+from langchain_core.messages import AIMessage, AIMessageChunk, BaseMessage, HumanMessage, ToolMessage
+from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult
+from langchain_core.runnables import Runnable, RunnableConfig, RunnableLambda
+from langchain_core.tools import BaseTool, tool
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph import END, START, MessagesState, StateGraph
+from langgraph.types import Command
+from langsmith import Client
+from langsmith.run_helpers import tracing_context
+from pydantic import BaseModel, Field
+
+
+
+class GenericFakeChatModel(BaseChatModel):
+ """Local fake chat model for subagent tests."""
+
+ messages: Iterator[AIMessage | str] = Field(exclude=True)
+ call_history: list[Any] = Field(default_factory=list)
+ tools: Sequence[dict[str, Any] | type | Callable | BaseTool] = ()
+ stream_delimiter: str | None = None
+
+ def bind_tools(
+ self,
+ tools: Sequence[dict[str, Any] | type | Callable | BaseTool],
+ *,
+ tool_choice: str | None = None,
+ **kwargs: Any,
+ ) -> Runnable[LanguageModelInput, AIMessage]:
+ self.tools = tools
+ return self
+
+ @override
+ def _generate(
+ self,
+ messages: list[BaseMessage],
+ stop: list[str] | None = None,
+ run_manager: CallbackManagerForLLMRun | None = None,
+ **kwargs: Any,
+ ) -> ChatResult:
+ self.call_history.append(
+ {
+ "messages": messages,
+ "kwargs": {"stop": stop, "run_manager": run_manager, **kwargs},
+ "tools": self.tools,
+ }
+ )
+ message = next(self.messages)
+ message_ = AIMessage(content=message) if isinstance(message, str) else message
+ return ChatResult(generations=[ChatGeneration(message=message_)])
+
+ def _stream(
+ self,
+ messages: list[BaseMessage],
+ stop: list[str] | None = None,
+ run_manager: CallbackManagerForLLMRun | None = None,
+ **kwargs: Any,
+ ) -> Iterator[ChatGenerationChunk]:
+ message = self._generate(messages, stop=stop, run_manager=run_manager, **kwargs).generations[0].message
+ if not isinstance(message, AIMessage):
+ msg = f"Expected invoke to return an AIMessage, got {type(message)}."
+ raise ValueError(msg)
+
+ content = message.content
+ tool_calls = message.tool_calls
+ if content:
+ if not isinstance(content, str):
+ msg = "Expected content to be a string."
+ raise ValueError(msg)
+ chunks = [content] if self.stream_delimiter is None else [chunk for chunk in cast("list[str]", re.split(self.stream_delimiter, content)) if chunk]
+ for idx, token in enumerate(chunks):
+ is_last = idx == len(chunks) - 1
+ chunk = ChatGenerationChunk(
+ message=AIMessageChunk(
+ content=token,
+ id=message.id,
+ tool_calls=tool_calls if is_last else [],
+ )
+ )
+ if is_last and not message.additional_kwargs:
+ chunk.message.chunk_position = "last"
+ if run_manager:
+ run_manager.on_llm_new_token(token, chunk=chunk)
+ yield chunk
+ elif tool_calls:
+ chunk = ChatGenerationChunk(
+ message=AIMessageChunk(
+ content="", id=message.id, tool_calls=tool_calls, chunk_position="last"
+ )
+ )
+ if run_manager:
+ run_manager.on_llm_new_token("", chunk=chunk)
+ yield chunk
+
+ for key, value in message.additional_kwargs.items():
+ if key == "function_call":
+ for fkey, fvalue in value.items():
+ value_chunks = (
+ cast("list[str]", re.split(r"(,)", fvalue))
+ if isinstance(fvalue, str)
+ else [fvalue]
+ )
+ for value_chunk in value_chunks:
+ chunk = ChatGenerationChunk(
+ message=AIMessageChunk(
+ id=message.id,
+ content="",
+ additional_kwargs={"function_call": {fkey: value_chunk}},
+ )
+ )
+ if run_manager:
+ run_manager.on_llm_new_token("", chunk=chunk)
+ yield chunk
+ else:
+ chunk = ChatGenerationChunk(
+ message=AIMessageChunk(
+ id=message.id, content="", additional_kwargs={key: value}
+ )
+ )
+ if run_manager:
+ run_manager.on_llm_new_token("", chunk=chunk)
+ yield chunk
+
+ @property
+ def _llm_type(self) -> str:
+ return "generic-fake-chat-model"
+
+
+class _ScriptedChatModel(BaseChatModel):
+ """Fake chat model that returns a fixed scripted sequence of AIMessages.
+
+ Each call to `_generate` returns the next message in `responses`;
+ once exhausted, it repeats the final response. This avoids `StopIteration`
+ bugs that arise with plain iterators under langgraph's generator runner.
+ """
+
+ responses: list[AIMessage] = [] # noqa: RUF012 # Pydantic field, per-instance
+ tools: Sequence[dict[str, Any] | type | Callable | BaseTool] = ()
+ _call_idx: int = 0
+
+ @property
+ def _llm_type(self) -> str:
+ return "scripted"
+
+ def _generate(
+ self,
+ messages: Sequence[Any],
+ stop: list[str] | None = None,
+ run_manager: CallbackManagerForLLMRun | None = None,
+ **kwargs: Any,
+ ) -> ChatResult:
+ idx = min(self._call_idx, len(self.responses) - 1)
+ self._call_idx += 1
+ return ChatResult(generations=[ChatGeneration(message=self.responses[idx])])
+
+ def bind_tools(
+ self,
+ tools: Sequence[dict[str, Any] | type | Callable | BaseTool],
+ *,
+ tool_choice: str | None = None,
+ **kwargs: Any,
+ ) -> Runnable[LanguageModelInput, AIMessage]:
+ self.tools = tools
+ return self
+
+
+class TestSubAgents:
+ """Tests for sub-agent middleware functionality."""
+
+ def test_subagent_returns_final_message_as_tool_result(self) -> None:
+ """Test that a subagent's final message is returned as a ToolMessage.
+
+ This test verifies the core subagent functionality:
+ 1. Parent agent invokes the 'task' tool to launch a subagent
+ 2. Subagent executes and returns a result
+ 3. The subagent's final message is extracted and returned to the parent
+ as a ToolMessage in the parent's message list
+ 4. Only the final message content is included (not the full conversation)
+
+ The response flow is:
+ - Parent receives ToolMessage with content from subagent's last AIMessage
+ - State updates (excluding messages/todos/structured_response) are merged
+ - Parent can then process the subagent's response and continue
+ """
+ # Create the parent agent's chat model that will call the subagent
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ # First response: invoke the task tool to launch subagent
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Calculate the sum of 2 and 3",
+ "subagent_type": "general-purpose",
+ },
+ "id": "call_calculate_sum",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ # Second response: acknowledge the subagent's result
+ AIMessage(content="The calculation has been completed."),
+ ]
+ )
+ )
+
+ # Create the subagent's chat model that will handle the calculation
+ subagent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(content="The sum of 2 and 3 is 5."),
+ ]
+ )
+ )
+
+ # Create the compiled subagent
+ compiled_subagent = create_agent(model=subagent_chat_model)
+
+ # Create the parent agent with subagent support
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ CompiledSubAgent(
+ name="general-purpose",
+ description="A general-purpose agent for various tasks.",
+ runnable=compiled_subagent,
+ ),
+ ]
+ )
+ ]
+ )
+
+ # Invoke the parent agent with an initial message
+ result = parent_agent.invoke(
+ {"messages": [HumanMessage(content="What is 2 + 3?")]},
+ config={"configurable": {"thread_id": "test_thread_calculation"}},
+ )
+
+ # Verify the result contains messages
+ assert "messages" in result, "Result should contain messages key"
+ assert len(result["messages"]) > 0, "Result should have at least one message"
+
+ # Find the ToolMessage that contains the subagent's response
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) > 0, "Should have at least one ToolMessage from subagent"
+
+ # Verify the ToolMessage contains the subagent's final response
+ subagent_tool_message = tool_messages[0]
+ assert "The sum of 2 and 3 is 5." in subagent_tool_message.content, "ToolMessage should contain subagent's final message content"
+
+ def test_multiple_subagents_invoked_in_parallel(self) -> None:
+ """Test that multiple different subagents can be launched in parallel.
+
+ This test verifies parallel execution with distinct subagent types:
+ 1. Parent agent makes a single AIMessage with multiple tool_calls
+ 2. Two different subagents are invoked concurrently (math-adder and math-multiplier)
+ 3. Each specialized subagent completes its task independently
+ 4. Both subagent results are returned as separate ToolMessages
+ 5. Parent agent receives both results and can synthesize them
+
+ The parallel execution pattern is important for:
+ - Reducing latency when tasks are independent
+ - Efficient resource utilization
+ - Handling multi-part user requests with specialized agents
+ """
+ # Create the parent agent's chat model that will call both subagents in parallel
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ # First response: invoke TWO different task tools in parallel
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Calculate the sum of 5 and 7",
+ "subagent_type": "math-adder",
+ },
+ "id": "call_addition",
+ "type": "tool_call",
+ },
+ {
+ "name": "task",
+ "args": {
+ "description": "Calculate the product of 4 and 6",
+ "subagent_type": "math-multiplier",
+ },
+ "id": "call_multiplication",
+ "type": "tool_call",
+ },
+ ],
+ ),
+ # Second response: acknowledge both results
+ AIMessage(content="Both calculations completed successfully."),
+ ]
+ )
+ )
+
+ # Create specialized subagent models - each handles a specific math operation
+ addition_subagent_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(content="The sum of 5 and 7 is 12."),
+ ]
+ )
+ )
+
+ multiplication_subagent_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(content="The product of 4 and 6 is 24."),
+ ]
+ )
+ )
+
+ # Compile the two different specialized subagents
+ addition_subagent = create_agent(model=addition_subagent_model)
+ multiplication_subagent = create_agent(model=multiplication_subagent_model)
+
+ # Create the parent agent with BOTH specialized subagents
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ CompiledSubAgent(
+ name="math-adder",
+ description="Specialized agent for addition operations.",
+ runnable=addition_subagent,
+ ),
+ CompiledSubAgent(
+ name="math-multiplier",
+ description="Specialized agent for multiplication operations.",
+ runnable=multiplication_subagent,
+ ),
+ ]
+ )
+ ],
+ )
+
+ # Invoke the parent agent with a request that triggers parallel subagent calls
+ result = parent_agent.invoke(
+ {"messages": [HumanMessage(content="What is 5+7 and what is 4*6?")]},
+ config={"configurable": {"thread_id": "test_thread_parallel"}},
+ )
+
+ # Verify the result contains messages
+ assert "messages" in result, "Result should contain messages key"
+
+ # Find all ToolMessages - should have one for each subagent invocation
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 2, f"Should have exactly 2 ToolMessages (one per subagent), but got {len(tool_messages)}"
+
+ # Create a lookup map from tool_call_id to ToolMessage for precise verification
+ tool_messages_by_id = {msg.tool_call_id: msg for msg in tool_messages}
+
+ # Verify we have both expected tool call IDs
+ assert "call_addition" in tool_messages_by_id, "Should have response from addition subagent"
+ assert "call_multiplication" in tool_messages_by_id, "Should have response from multiplication subagent"
+
+ # Verify the exact content of each response by looking up the specific tool message
+ addition_tool_message = tool_messages_by_id["call_addition"]
+ assert addition_tool_message.content == "The sum of 5 and 7 is 12.", (
+ f"Addition subagent should return exact message, got: {addition_tool_message.content}"
+ )
+
+ multiplication_tool_message = tool_messages_by_id["call_multiplication"]
+ assert multiplication_tool_message.content == "The product of 4 and 6 is 24.", (
+ f"Multiplication subagent should return exact message, got: {multiplication_tool_message.content}"
+ )
+
+ def test_private_state_does_not_propagate_between_sibling_subagents(self) -> None:
+ """A private state field should not propagate from one sibling subagent to another."""
+
+ class _LocalPrivateState(AgentState):
+ shared_value: Annotated[str | None, PrivateStateAttr]
+
+ class _LocalPrivateMiddleware(AgentMiddleware[_LocalPrivateState, Any, Any]):
+ state_schema = _LocalPrivateState
+
+ def before_agent(self, state: _LocalPrivateState, runtime: object) -> dict[str, Any] | None:
+ if "shared_value" in state:
+ return None
+ return {"shared_value": "seeded"}
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Seed the interpreter state",
+ "subagent_type": "writer",
+ },
+ "id": "call_writer",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Read the interpreter state",
+ "subagent_type": "reader",
+ },
+ "id": "call_reader",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="done"),
+ ]
+ )
+ )
+
+ writer_model = GenericFakeChatModel(messages=iter([AIMessage(content="writer saw seeded")]))
+ reader_model = GenericFakeChatModel(messages=iter([AIMessage(content="reader saw missing")]))
+
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[
+ SubAgentMiddleware(
+ subagents=[
+ SubAgent(
+ name="writer",
+ description="Writes state.",
+ system_prompt="Write the seeded private state value and report completion.",
+ model=writer_model,
+ middleware=[_LocalPrivateMiddleware()],
+ ),
+ SubAgent(
+ name="reader",
+ description="Reads state.",
+ system_prompt="Read the private state value and report what you received.",
+ model=reader_model,
+ ),
+ ],
+ )
+ ]
+ )
+
+ result = parent_agent.invoke(
+ {"messages": [HumanMessage(content="run the two subagents")]},
+ config={"configurable": {"thread_id": "test_shared_quickjs_subagents"}},
+ )
+
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 2
+ assert "seeded" in tool_messages[0].content
+ assert "missing" in tool_messages[1].content
+
+ def test_private_state_does_not_propagate_from_parent_to_subagent(self) -> None:
+ """A private state field on the parent should not be visible to a child subagent."""
+
+ class _ParentPrivateState(AgentState):
+ shared_value: Annotated[str | None, PrivateStateAttr]
+
+ class _ChildCaptureState(AgentState):
+ shared_value: Annotated[str | None, PrivateStateAttr]
+
+ captured_child_states: list[dict[str, Any]] = []
+
+ class _ChildCaptureMiddleware(AgentMiddleware[_ChildCaptureState, Any, Any]):
+ state_schema = _ChildCaptureState
+
+ def before_agent(self, state: _ChildCaptureState, runtime: object) -> dict[str, Any] | None:
+ captured_child_states.append(dict(state))
+ return None
+
+ class _ParentSeedMiddleware(AgentMiddleware[_ParentPrivateState, Any, Any]):
+ state_schema = _ParentPrivateState
+
+ def before_agent(self, state: _ParentPrivateState, runtime: object) -> dict[str, Any] | None:
+ if "shared_value" in state:
+ return None
+ return {"shared_value": "parent-secret"}
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Run the child subagent",
+ "subagent_type": "child",
+ },
+ "id": "call_child",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="done"),
+ ]
+ )
+ )
+
+ child_model = GenericFakeChatModel(messages=iter([AIMessage(content="child done")]))
+
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[_ParentSeedMiddleware(),
+ SubAgentMiddleware(
+ subagents=[
+ SubAgent(
+ name="child",
+ description="Captures its incoming state.",
+ system_prompt="Capture the incoming state and complete the task.",
+ model=child_model,
+ middleware=[_ChildCaptureMiddleware()],
+ ),
+ ]
+ )
+ ],
+ )
+
+ result = parent_agent.invoke(
+ {"messages": [HumanMessage(content="run the child subagent")]},
+ config={
+ "configurable": {"thread_id": "test_private_state_parent_to_child"},
+ "metadata": {"shared_value": "parent-secret"},
+ },
+ )
+
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 1
+ assert "child done" in tool_messages[0].content
+ assert captured_child_states, "Child subagent should have received state"
+ assert "shared_value" not in captured_child_states[0]
+
+ def test_private_state_from_custom_state_schema_does_not_propagate_to_subagent(self) -> None:
+ """Private fields on `create_agent(state_schema=...)` should not reach subagents."""
+
+ class _ParentState(AgentState):
+ parent_secret: Annotated[str | None, PrivateStateAttr]
+ public_value: str | None
+
+ captured_subagent_states: list[dict[str, Any]] = []
+
+ @tool
+ def seed_parent_state(runtime: ToolRuntime) -> Command:
+ """Seed parent state."""
+ return Command(
+ update={
+ "parent_secret": "parent-secret",
+ "public_value": "parent-public",
+ "messages": [ToolMessage(content="seeded", tool_call_id=runtime.tool_call_id)],
+ }
+ )
+
+ @tool
+ def capture_subagent_state(query: str, runtime: ToolRuntime) -> str:
+ """Capture subagent state."""
+ captured_subagent_states.append(dict(runtime.state))
+ return f"captured {query}"
+
+ parent_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "seed_parent_state",
+ "args": {},
+ "id": "call_seed_parent_state",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Capture the incoming state",
+ "subagent_type": "child",
+ },
+ "id": "call_child",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="done"),
+ ]
+ )
+ )
+ child_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "capture_subagent_state",
+ "args": {"query": "state"},
+ "id": "call_capture_subagent_state",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="child done"),
+ ]
+ )
+ )
+
+ parent_agent = create_agent(
+ model=parent_model,
+ tools=[seed_parent_state],
+ checkpointer=InMemorySaver(),
+ state_schema=_ParentState,
+ middleware=[
+ SubAgentMiddleware(
+ state_schema=_ParentState,
+ subagents=[
+ SubAgent(
+ name="child",
+ description="Captures its incoming state.",
+ system_prompt="Capture the incoming state and complete the task.",
+ model=child_model,
+ tools=[capture_subagent_state],
+ )
+ ],
+ )
+ ]
+ )
+
+ parent_agent.invoke(
+ {"messages": [HumanMessage(content="seed state and run the child subagent")]},
+ config={"configurable": {"thread_id": "test_private_state_schema_parent_to_child"}},
+ )
+
+ assert captured_subagent_states, "Child subagent should have captured state"
+ assert captured_subagent_states[0]["public_value"] == "parent-public"
+ assert "parent_secret" not in captured_subagent_states[0]
+
+ def test_agent_with_structured_output_tool_strategy(self) -> None:
+ """Test that an agent with ToolStrategy properly generates structured output.
+
+ This test verifies the structured output setup:
+ 1. Define a Pydantic model as the response schema
+ 2. Configure agent with ToolStrategy for structured output
+ 3. Fake model calls the structured output tool
+ 4. Agent validates and returns the structured response
+ 5. The structured_response key contains the validated Pydantic instance
+
+ This validates our understanding of how to set up structured output
+ correctly using the fake model for testing.
+ """
+
+ # Define the Pydantic model for structured output
+ class WeatherReport(BaseModel):
+ """Structured weather information."""
+
+ location: str = Field(description="The city or location for the weather report")
+ temperature: float = Field(description="Temperature in Celsius")
+ condition: str = Field(description="Weather condition (e.g., sunny, rainy)")
+
+ # Create a fake model that calls the structured output tool
+ # The tool name will be the schema class name: "WeatherReport"
+ fake_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "WeatherReport",
+ "args": {
+ "location": "San Francisco",
+ "temperature": 18.5,
+ "condition": "sunny",
+ },
+ "id": "call_weather_report",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ ]
+ )
+ )
+
+ # Create agent with ToolStrategy for structured output
+ agent = create_agent(
+ model=fake_model,
+ response_format=ToolStrategy(schema=WeatherReport),
+ )
+
+ # Invoke the agent
+ result = agent.invoke({"messages": [HumanMessage(content="What's the weather in San Francisco?")]})
+
+ # Verify the structured_response key exists in the result
+ assert "structured_response" in result, "Result should contain structured_response key"
+
+ # Verify the structured response is the correct type
+ structured_response = result["structured_response"]
+ assert isinstance(structured_response, WeatherReport), f"Expected WeatherReport instance, got {type(structured_response)}"
+
+ # Verify the structured response has the correct values
+ expected_response = WeatherReport(location="San Francisco", temperature=18.5, condition="sunny")
+ assert structured_response == expected_response, f"Expected {expected_response}, got {structured_response}"
+
+ def test_parallel_subagents_with_todo_lists(self) -> None:
+ """Test that multiple subagents can manage their own isolated todo lists.
+
+ This test verifies that:
+ 1. Multiple subagents can be invoked in parallel
+ 2. Each subagent can use write_todos to manage its own todo list
+ 3. Todo lists are properly isolated to each subagent (not merged into parent)
+ 4. Parent receives clean ToolMessages from each subagent
+ 5. The 'todos' key is excluded from parent state per _EXCLUDED_STATE_KEYS
+
+ This validates that todo list state isolation works correctly in parallel execution.
+ """
+ # Create parent agent's chat model that calls two subagents in parallel
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ # First response: invoke TWO subagents in parallel
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Research the history of Python programming language",
+ "subagent_type": "python-researcher",
+ },
+ "id": "call_research_python",
+ "type": "tool_call",
+ },
+ {
+ "name": "task",
+ "args": {
+ "description": "Research the history of JavaScript programming language",
+ "subagent_type": "javascript-researcher",
+ },
+ "id": "call_research_javascript",
+ "type": "tool_call",
+ },
+ ],
+ ),
+ # Second response: acknowledge both results
+ AIMessage(content="Both research tasks completed successfully."),
+ ]
+ )
+ )
+
+ # Create first subagent that uses write_todos and returns a result
+ python_subagent_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ # First: write some todos
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "write_todos",
+ "args": {
+ "todos": [
+ {
+ "content": "Search for Python history",
+ "status": "in_progress",
+ "activeForm": "Searching for Python history",
+ },
+ {"content": "Summarize findings", "status": "pending", "activeForm": "Summarizing findings"},
+ ]
+ },
+ "id": "call_write_todos_python_1",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ # Second: update todos and return final message
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "write_todos",
+ "args": {
+ "todos": [
+ {"content": "Search for Python history", "status": "completed", "activeForm": "Searching for Python history"},
+ {"content": "Summarize findings", "status": "completed", "activeForm": "Summarizing findings"},
+ ]
+ },
+ "id": "call_write_todos_python_2",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ # Final result message
+ AIMessage(content="Python was created by Guido van Rossum and released in 1991."),
+ ]
+ )
+ )
+
+ # Create second subagent that uses write_todos and returns a result
+ javascript_subagent_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ # First: write some todos
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "write_todos",
+ "args": {
+ "todos": [
+ {
+ "content": "Search for JavaScript history",
+ "status": "in_progress",
+ "activeForm": "Searching for JavaScript history",
+ },
+ {"content": "Compile summary", "status": "pending", "activeForm": "Compiling summary"},
+ ]
+ },
+ "id": "call_write_todos_js_1",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ # Second: update todos and return final message
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "write_todos",
+ "args": {
+ "todos": [
+ {
+ "content": "Search for JavaScript history",
+ "status": "completed",
+ "activeForm": "Searching for JavaScript history",
+ },
+ {"content": "Compile summary", "status": "completed", "activeForm": "Compiling summary"},
+ ]
+ },
+ "id": "call_write_todos_js_2",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ # Final result message
+ AIMessage(content="JavaScript was created by Brendan Eich at Netscape in 1995."),
+ ]
+ )
+ )
+
+ python_research_agent = create_agent(
+ model=python_subagent_model,
+ middleware=[TodoListMiddleware()],
+ )
+
+ javascript_research_agent = create_agent(
+ model=javascript_subagent_model,
+ middleware=[TodoListMiddleware()],
+ )
+
+ # Create parent agent with both specialized subagents
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="python-researcher",
+ description="Agent specialized in Python research.",
+ runnable=python_research_agent,
+ ),
+ CompiledSubAgent(
+ name="javascript-researcher",
+ description="Agent specialized in JavaScript research.",
+ runnable=javascript_research_agent,
+ ),
+ ])],
+ )
+
+ # Invoke the parent agent
+ result = parent_agent.invoke(
+ {"messages": [HumanMessage(content="Research Python and JavaScript history")]},
+ config={"configurable": {"thread_id": "test_thread_todos"}},
+ )
+
+ # Verify the result contains messages
+ assert "messages" in result, "Result should contain messages key"
+
+ # Find all ToolMessages from the subagents
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 2, f"Should have exactly 2 ToolMessages, got {len(tool_messages)}"
+
+ # Create lookup map by tool_call_id
+ tool_messages_by_id = {msg.tool_call_id: msg for msg in tool_messages}
+
+ # Verify both expected tool call IDs are present
+ assert "call_research_python" in tool_messages_by_id, "Should have response from Python researcher"
+ assert "call_research_javascript" in tool_messages_by_id, "Should have response from JavaScript researcher"
+
+ # Verify that todos are NOT in the parent agent's final state
+ # (they should be excluded per _EXCLUDED_STATE_KEYS)
+ assert "todos" not in result, "Parent agent state should not contain todos key (it should be excluded per _EXCLUDED_STATE_KEYS)"
+
+ # Verify the final messages contain the research results
+ python_tool_message = tool_messages_by_id["call_research_python"]
+ assert "Python was created by Guido van Rossum" in python_tool_message.content, (
+ f"Expected Python research result in message, got: {python_tool_message.content}"
+ )
+
+ javascript_tool_message = tool_messages_by_id["call_research_javascript"]
+ assert "JavaScript was created by Brendan Eich" in javascript_tool_message.content, (
+ f"Expected JavaScript research result in message, got: {javascript_tool_message.content}"
+ )
+
+ def test_subagent_propagates_recursion_limit_to_tool_runtime(self) -> None:
+ """Test that subagent tools receive the parent's recursion limit via `ToolRuntime.config`."""
+ captured_config: Any = None
+
+ @tool
+ def capture_recursion_limit(runtime: ToolRuntime) -> str:
+ """Capture the recursion limit from runtime config."""
+ nonlocal captured_config
+ captured_config = runtime.config
+ return "OK"
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Check the recursion limit and report it.",
+ "subagent_type": "general-purpose",
+ },
+ "id": "call_subagent_recursion_limit",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="The subagent finished successfully."),
+ ]
+ )
+ )
+
+ subagent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "capture_recursion_limit",
+ "args": {},
+ "id": "call_capture_recursion_limit",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="done"),
+ ]
+ )
+ )
+
+ compiled_subagent = create_agent(
+ model=subagent_chat_model,
+ tools=[capture_recursion_limit],
+ name="subagent-runtime-check",
+ ).with_config({"recursion_limit": 5000})
+
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="general-purpose",
+ description="A general-purpose agent for various tasks.",
+ runnable=compiled_subagent,
+ )
+ ])],
+ )
+
+ result = parent_agent.invoke(
+ {"messages": [HumanMessage(content="Run the recursion limit check.")]},
+ config={
+ "configurable": {"thread_id": str(uuid.uuid4())},
+ "tags": ["hello"],
+ },
+ durability="exit",
+ )
+
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 1
+ assert captured_config is not None
+ assert captured_config["recursion_limit"] == 5000
+ # Pregel merges the runtime recursion_limit patch with the subagent's own
+ # config instead of replacing it wholesale.
+ assert captured_config["tags"] == ["hello"]
+ # CompiledSubAgent.name takes precedence over the name set in create_agent()
+ # so that lc_agent_name in streamed chunks reflects the declared subagent name.
+ assert captured_config["metadata"]["lc_agent_name"] == "general-purpose"
+
+ def test_subagent_inherits_parent_user_metadata(self) -> None:
+ """User metadata set on the parent invoke reaches subagent runs (deepagents#3634).
+
+ `langgraph`'s `ensure_config` seeds each run's metadata from the ambient
+ parent config and merges it per-key (langgraph#7926). A user key like
+ `customer_id` therefore propagates into subagent runs, while the
+ subagent's bound `lc_agent_name` wins the key collision and is preserved.
+
+ Requires a `langgraph` that includes langgraph#7926's merge semantics;
+ with the older overwrite behaviour the parent metadata is dropped.
+ """
+ captured_config: Any = None
+
+ @tool
+ def capture_metadata(runtime: ToolRuntime) -> str:
+ """Capture the runtime config from inside the subagent."""
+ nonlocal captured_config
+ captured_config = runtime.config
+ return "OK"
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Capture metadata and report it.",
+ "subagent_type": "general-purpose",
+ },
+ "id": "call_subagent_metadata",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="The subagent finished successfully."),
+ ]
+ )
+ )
+
+ subagent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "capture_metadata",
+ "args": {},
+ "id": "call_capture_metadata",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="done"),
+ ]
+ )
+ )
+
+ compiled_subagent = create_agent(
+ model=subagent_chat_model,
+ tools=[capture_metadata],
+ name="subagent-runtime-check",
+ )
+
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="general-purpose",
+ description="A general-purpose agent for various tasks.",
+ runnable=compiled_subagent,
+ )
+ ])],
+ )
+
+ parent_agent.invoke(
+ {"messages": [HumanMessage(content="Run the metadata check.")]},
+ config={
+ "configurable": {"thread_id": str(uuid.uuid4())},
+ # `lc_agent_name` collides with the subagent's bound identity (it
+ # must keep its own value); `customer_id` is a non-colliding user
+ # key that must survive the merge into the subagent's runs.
+ "metadata": {"customer_id": "abc-123", "lc_agent_name": "parent-agent"},
+ },
+ durability="exit",
+ )
+
+ assert captured_config is not None
+ subagent_metadata = captured_config["metadata"]
+ # User-set parent metadata propagated into the subagent run.
+ assert subagent_metadata["customer_id"] == "abc-123"
+ # The subagent's bound identity won the `lc_agent_name` collision.
+ assert subagent_metadata["lc_agent_name"] == "general-purpose"
+
+ @pytest.mark.xfail(
+ reason="callbacks in parent config are not forwarded to subagent invocations (see #2315)",
+ strict=True,
+ )
+ def test_subagent_propagates_callbacks_to_model_calls(self) -> None:
+ """Test that callbacks in parent config are forwarded to subagent model invocations.
+
+ Regression test for https://github.com/langchain-ai/deepagents/issues/2315.
+ """
+ llm_start_agent_names: list[str] = []
+
+ class CapturingCallback(BaseCallbackHandler):
+ def on_llm_start(self, serialized: dict, prompts: list, **kwargs: Any) -> None:
+ llm_start_agent_names.append(kwargs.get("name", "unknown"))
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Do something.",
+ "subagent_type": "general-purpose",
+ },
+ "id": "call_subagent_callback",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done."),
+ ]
+ )
+ )
+
+ subagent_chat_model = GenericFakeChatModel(messages=iter([AIMessage(content="Subagent done.")]))
+
+ compiled_subagent = create_agent(
+ model=subagent_chat_model,
+ name="callback-check-subagent",
+ )
+
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="general-purpose",
+ description="A general-purpose agent.",
+ runnable=compiled_subagent,
+ )
+ ])],
+ )
+
+ callback = CapturingCallback()
+
+ parent_agent.invoke(
+ {"messages": [HumanMessage(content="Run the callback check.")]},
+ config={
+ "configurable": {"thread_id": str(uuid.uuid4())},
+ "callbacks": [callback],
+ },
+ durability="exit",
+ )
+
+ # All three LLM calls (2 parent + 1 subagent) should trigger the callback
+ assert len(llm_start_agent_names) == 3, (
+ f"Expected callbacks from 2 parent + 1 subagent LLM calls, but only got {len(llm_start_agent_names)}: {llm_start_agent_names}"
+ )
+ # The subagent name should be identifiable in at least one callback
+ assert any(name == "callback-check-subagent" for name in llm_start_agent_names), (
+ f"Subagent LLM call should have triggered callback with correct name, got: {llm_start_agent_names}"
+ )
+
+ def test_parallel_subagents_with_different_structured_outputs(self) -> None:
+ """Test that multiple subagents with different structured outputs work correctly.
+
+ This test verifies that:
+ 1. Two different subagents can be invoked in parallel
+ 2. Each subagent has its own structured output schema
+ 3. Structured responses are properly excluded from parent state (per _EXCLUDED_STATE_KEYS)
+ 4. Parent receives clean ToolMessages from each subagent
+ 5. Each subagent's structured_response stays isolated to that subagent
+
+ This validates that structured_response exclusion prevents schema conflicts
+ between parent and subagent agents.
+ """
+
+ # Define structured output schemas for the two specialized subagents
+ class CityWeather(BaseModel):
+ """Weather information for a city."""
+
+ city: str = Field(description="Name of the city")
+ temperature_celsius: float = Field(description="Temperature in Celsius")
+ humidity_percent: int = Field(description="Humidity percentage")
+
+ class CityPopulation(BaseModel):
+ """Population statistics for a city."""
+
+ city: str = Field(description="Name of the city")
+ population: int = Field(description="Total population")
+ metro_area_population: int = Field(description="Metropolitan area population")
+
+ # Create parent agent's chat model that calls both subagents in parallel
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ # First response: invoke TWO different subagents in parallel
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Get weather information for Tokyo",
+ "subagent_type": "weather-analyzer",
+ },
+ "id": "call_weather",
+ "type": "tool_call",
+ },
+ {
+ "name": "task",
+ "args": {
+ "description": "Get population statistics for Tokyo",
+ "subagent_type": "population-analyzer",
+ },
+ "id": "call_population",
+ "type": "tool_call",
+ },
+ ],
+ ),
+ # Second response: acknowledge both results
+ AIMessage(content="I've gathered weather and population data for Tokyo."),
+ ]
+ )
+ )
+
+ # Create weather subagent with structured output
+ weather_subagent_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "CityWeather",
+ "args": {
+ "city": "Tokyo",
+ "temperature_celsius": 22.5,
+ "humidity_percent": 65,
+ },
+ "id": "call_weather_struct",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ ]
+ )
+ )
+
+ weather_subagent = create_agent(
+ model=weather_subagent_model,
+ response_format=ToolStrategy(schema=CityWeather),
+ )
+
+ # Create population subagent with structured output
+ population_subagent_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "CityPopulation",
+ "args": {
+ "city": "Tokyo",
+ "population": 14000000,
+ "metro_area_population": 37400000,
+ },
+ "id": "call_population_struct",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ ]
+ )
+ )
+
+ population_subagent = create_agent(
+ model=population_subagent_model,
+ response_format=ToolStrategy(schema=CityPopulation),
+ )
+
+ # Create parent agent with both specialized subagents
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="weather-analyzer",
+ description="Specialized agent for weather analysis.",
+ runnable=weather_subagent,
+ ),
+ CompiledSubAgent(
+ name="population-analyzer",
+ description="Specialized agent for population analysis.",
+ runnable=population_subagent,
+ ),
+ ])],
+ )
+
+ # Invoke the parent agent
+ result = parent_agent.invoke(
+ {"messages": [HumanMessage(content="Tell me about Tokyo's weather and population")]},
+ config={"configurable": {"thread_id": "test_thread_structured"}},
+ )
+
+ # Verify the result contains messages
+ assert "messages" in result, "Result should contain messages key"
+
+ # Find all ToolMessages from the subagents
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 2, f"Should have exactly 2 ToolMessages, got {len(tool_messages)}"
+
+ # Create lookup map by tool_call_id
+ tool_messages_by_id = {msg.tool_call_id: msg for msg in tool_messages}
+
+ # Verify both expected tool call IDs are present
+ assert "call_weather" in tool_messages_by_id, "Should have response from weather subagent"
+ assert "call_population" in tool_messages_by_id, "Should have response from population subagent"
+
+ # Verify that structured_response is NOT in the parent agent's final state
+ # (it should be excluded per _EXCLUDED_STATE_KEYS)
+ assert "structured_response" not in result, (
+ "Parent agent state should not contain structured_response key (it should be excluded per _EXCLUDED_STATE_KEYS)"
+ )
+
+ # When a subagent produces a structured_response, the ToolMessage content is
+ # the JSON-serialized structured data (not the last message text).
+ weather_tool_message = tool_messages_by_id["call_weather"]
+ weather_parsed = CityWeather.model_validate_json(weather_tool_message.content)
+ assert weather_parsed == CityWeather(city="Tokyo", temperature_celsius=22.5, humidity_percent=65), (
+ f"Expected JSON-serialized weather data, got: {weather_tool_message.content}"
+ )
+
+ population_tool_message = tool_messages_by_id["call_population"]
+ population_parsed = CityPopulation.model_validate_json(population_tool_message.content)
+ assert population_parsed == CityPopulation(city="Tokyo", population=14000000, metro_area_population=37400000), (
+ f"Expected JSON-serialized population data, got: {population_tool_message.content}"
+ )
+
+ def test_structured_response_serialized_as_tool_message(self) -> None:
+ """Test that structured_response is JSON-serialized as ToolMessage content.
+
+ When a subagent produces a `structured_response`, the middleware should
+ JSON-serialize it as the ToolMessage content instead of extracting the
+ last message text.
+ """
+ structured_data = {
+ "findings": "Renewable energy adoption is accelerating",
+ "confidence": 0.92,
+ "sources": 3,
+ }
+
+ mock_subagent = RunnableLambda(
+ lambda _: {
+ "messages": [AIMessage(content="Here are my findings about renewable energy.")],
+ "structured_response": structured_data,
+ }
+ )
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Analyze renewable energy trends",
+ "subagent_type": "analyzer",
+ },
+ "id": "call_structured",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done"),
+ ]
+ )
+ )
+
+ agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="analyzer",
+ description="An analysis agent",
+ runnable=mock_subagent,
+ ),
+ ])],
+ )
+
+ result = agent.invoke(
+ {"messages": [HumanMessage(content="Analyze renewable energy")]},
+ config={"configurable": {"thread_id": f"test-structured-{uuid.uuid4().hex}"}},
+ )
+
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 1
+ task_tool_message = tool_messages[0]
+ assert task_tool_message.content == json.dumps(structured_data)
+
+ parsed = json.loads(task_tool_message.content)
+ assert parsed == structured_data
+
+ def test_structured_response_dataclass_serialized_as_tool_message(self) -> None:
+ """Test that a dataclass structured_response is JSON-serialized correctly.
+
+ Dataclass instances don't have `model_dump_json` and aren't natively
+ JSON-serializable, so the middleware must convert them via
+ `dataclasses.asdict` before calling `json.dumps`.
+ """
+
+ @dataclasses.dataclass
+ class AnalysisResult:
+ findings: str
+ confidence: float
+ sources: int
+
+ structured_instance = AnalysisResult(
+ findings="Renewable energy adoption is accelerating",
+ confidence=0.92,
+ sources=3,
+ )
+
+ mock_subagent = RunnableLambda(
+ lambda _: {
+ "messages": [AIMessage(content="Here are my findings.")],
+ "structured_response": structured_instance,
+ }
+ )
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Analyze trends",
+ "subagent_type": "analyzer",
+ },
+ "id": "call_dc",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done"),
+ ]
+ )
+ )
+
+ agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="analyzer",
+ description="An analysis agent",
+ runnable=mock_subagent,
+ ),
+ ])],
+ )
+
+ result = agent.invoke(
+ {"messages": [HumanMessage(content="Analyze")]},
+ config={"configurable": {"thread_id": f"test-dc-structured-{uuid.uuid4().hex}"}},
+ )
+
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 1
+ task_tool_message = tool_messages[0]
+
+ parsed = json.loads(task_tool_message.content)
+ assert parsed == {
+ "findings": "Renewable energy adoption is accelerating",
+ "confidence": 0.92,
+ "sources": 3,
+ }
+
+ def test_structured_response_pydantic_serialized_as_tool_message(self) -> None:
+ """Test that a Pydantic model structured_response uses model_dump_json."""
+
+ class AnalysisResult(BaseModel):
+ findings: str
+ confidence: float
+
+ structured_instance = AnalysisResult(
+ findings="Solar is growing fast",
+ confidence=0.95,
+ )
+
+ mock_subagent = RunnableLambda(
+ lambda _: {
+ "messages": [AIMessage(content="Here are my findings.")],
+ "structured_response": structured_instance,
+ }
+ )
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Analyze trends",
+ "subagent_type": "analyzer",
+ },
+ "id": "call_pydantic",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done"),
+ ]
+ )
+ )
+
+ agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="analyzer",
+ description="An analysis agent",
+ runnable=mock_subagent,
+ ),
+ ])],
+ )
+
+ result = agent.invoke(
+ {"messages": [HumanMessage(content="Analyze")]},
+ config={"configurable": {"thread_id": f"test-pydantic-structured-{uuid.uuid4().hex}"}},
+ )
+
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 1
+ task_tool_message = tool_messages[0]
+
+ parsed = AnalysisResult.model_validate_json(task_tool_message.content)
+ assert parsed == AnalysisResult(findings="Solar is growing fast", confidence=0.95)
+
+ def test_fallback_to_last_message_without_structured_response(self) -> None:
+ """Test fallback to last message when no structured_response is present.
+
+ When a subagent does not produce a `structured_response`, the middleware
+ should fall back to extracting the last message text.
+ """
+ mock_subagent = RunnableLambda(
+ lambda _: {
+ "messages": [AIMessage(content="Plain text result without structured response")],
+ }
+ )
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Do work",
+ "subagent_type": "worker",
+ },
+ "id": "call_plain",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done"),
+ ]
+ )
+ )
+
+ agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="worker",
+ description="A worker agent",
+ runnable=mock_subagent,
+ ),
+ ])],
+ )
+
+ result = agent.invoke(
+ {"messages": [HumanMessage(content="Test")]},
+ config={"configurable": {"thread_id": f"test-no-structured-{uuid.uuid4().hex}"}},
+ )
+
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 1
+ task_tool_message = tool_messages[0]
+ assert task_tool_message.content == "Plain text result without structured response"
+
+ def test_fallback_skips_trailing_empty_ai_message(self) -> None:
+ """Skip a trailing empty AIMessage and use the last AIMessage with text.
+
+ Anthropic/Bedrock occasionally emits an empty `end_turn` AIMessage after
+ a successful final tool call. The middleware should walk back to the
+ prior AIMessage carrying the real answer instead of forwarding an empty
+ ToolMessage.
+ """
+ mock_subagent = RunnableLambda(
+ lambda _: {
+ "messages": [
+ AIMessage(content="The real answer from the subagent."),
+ AIMessage(content=""),
+ ],
+ }
+ )
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Do work",
+ "subagent_type": "worker",
+ },
+ "id": "call_trailing_empty",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done"),
+ ]
+ )
+ )
+
+ agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="worker",
+ description="A worker agent",
+ runnable=mock_subagent,
+ ),
+ ])],
+ )
+
+ result = agent.invoke(
+ {"messages": [HumanMessage(content="Test")]},
+ config={"configurable": {"thread_id": f"test-trailing-empty-{uuid.uuid4().hex}"}},
+ )
+
+ tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+ assert len(tool_messages) == 1
+ assert tool_messages[0].content == "The real answer from the subagent."
+
+ def test_subagent_streaming_emits_messages_and_updates_from_subgraph(self) -> None:
+ """Test end-to-end subagent streaming with `subgraphs=True`.
+
+ Verifies:
+ 1. Parent and subagent message chunks are both streamed in `messages` mode.
+ 2. Parent and subagent completed messages are both streamed in `updates` mode.
+ 3. Subagent message metadata includes its `lc_agent_name`, inherited tags, and config metadata.
+ 4. The subagent's tool result is surfaced back through the parent tools update.
+ """
+ parent_content = "PARENT_RESPONSE"
+ subagent_content = "SUBAGENT_RESPONSE"
+ test_tags = ["test-tag", "session-123"]
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {"description": "Do task", "subagent_type": "worker"},
+ "id": "call_worker",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content=parent_content),
+ ]
+ ),
+ stream_delimiter="_",
+ )
+ subagent_chat_model = GenericFakeChatModel(
+ messages=iter([AIMessage(content=subagent_content)]),
+ stream_delimiter="_",
+ )
+
+ compiled_subagent = create_agent(model=subagent_chat_model, name="worker")
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ name="supervisor",
+ middleware=[SubAgentMiddleware(subagents=[CompiledSubAgent(name="worker", description="Does work.", runnable=compiled_subagent)])],
+ )
+
+ saw_parent_message_chunk = False
+ saw_subagent_message_chunk = False
+ saw_subagent_update = False
+ saw_parent_tools_update = False
+ saw_parent_model_update = False
+
+ seen_agent_names: set[str | None] = set()
+
+ for ns, stream_mode, data in parent_agent.stream(
+ {"messages": [HumanMessage(content="Do something")]},
+ stream_mode=["messages", "updates"],
+ subgraphs=True,
+ config={"configurable": {"thread_id": "test_thread"}, "tags": test_tags},
+ ):
+ if stream_mode == "messages":
+ message_chunk, metadata = data
+ agent_name = metadata.get("lc_agent_name")
+ seen_agent_names.add(agent_name)
+ tags = metadata.get("tags", [])
+
+ if parent_content.split("_", maxsplit=1)[0] in message_chunk.content and agent_name == "supervisor":
+ saw_parent_message_chunk = True
+
+ if subagent_content.split("_", maxsplit=1)[0] in message_chunk.content and agent_name == "worker":
+ assert all(t in tags for t in test_tags), f"Subagent chunk missing tags. Expected {test_tags}, got {tags}"
+ saw_subagent_message_chunk = True
+
+ elif stream_mode == "updates":
+ update = data
+ if "model" in update and ns and ns[-1].startswith("tools:"):
+ subagent_message = update["model"]["messages"][-1]
+ assert subagent_message.content == subagent_content.replace("_", "")
+ saw_subagent_update = True
+ elif "tools" in update and ns == ():
+ tool_message = update["tools"]["messages"][-1]
+ assert tool_message.content == subagent_content.replace("_", "")
+ saw_parent_tools_update = True
+ elif "model" in update and ns == ():
+ parent_message = update["model"]["messages"][-1]
+ if parent_message.content == parent_content.replace("_", ""):
+ saw_parent_model_update = True
+
+ assert saw_parent_message_chunk, "Should have seen parent message chunks in the stream"
+ assert saw_subagent_message_chunk, "Should have seen subagent message chunks in the stream"
+ assert saw_subagent_update, "Should have seen a subagent model update in the stream"
+ assert saw_parent_tools_update, "Should have seen the parent tools update with the subagent result"
+ assert saw_parent_model_update, "Should have seen the parent final model update in the stream"
+ assert seen_agent_names == {"supervisor", "worker"}
+
+ def test_compiled_subagent_lc_agent_name_in_stream_metadata(self) -> None:
+ """lc_agent_name in streamed chunks must reflect the CompiledSubAgent's declared name.
+
+ Regression test for #2925: when a raw StateGraph (not created via create_agent)
+ is passed as a CompiledSubAgent, streamed chunks must carry the declared name in
+ metadata, not the parent agent's name.
+ """
+ subagent_content = "RAW_GRAPH_RESPONSE"
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {"description": "Do task", "subagent_type": "raw-worker"},
+ "id": "call_raw_worker",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done."),
+ ]
+ ),
+ stream_delimiter="_",
+ )
+ subagent_chat_model = GenericFakeChatModel(
+ messages=iter([AIMessage(content=subagent_content)]),
+ stream_delimiter="_",
+ )
+
+ # Raw StateGraph — NOT created via create_agent, so no lc_agent_name pre-set.
+ builder = StateGraph(MessagesState)
+ builder.add_node("model", create_agent(model=subagent_chat_model))
+ builder.add_edge(START, "model")
+ raw_graph = builder.compile()
+
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ name="supervisor",
+ middleware=[SubAgentMiddleware(subagents=[CompiledSubAgent(name="raw-worker", description="Raw graph subagent.", runnable=raw_graph)])],
+ )
+
+ seen_subagent_names: set[str | None] = set()
+
+ for _ns, stream_mode, data in parent_agent.stream(
+ {"messages": [HumanMessage(content="Do something")]},
+ stream_mode=["messages"],
+ subgraphs=True,
+ config={"configurable": {"thread_id": "test_raw_graph_lc_agent_name"}},
+ ):
+ if stream_mode == "messages":
+ message_chunk, metadata = data
+ if message_chunk.content:
+ seen_subagent_names.add(metadata.get("lc_agent_name"))
+
+ assert "raw-worker" in seen_subagent_names, f"Expected 'raw-worker' in streamed lc_agent_name metadata, got: {seen_subagent_names}"
+
+ async def test_compiled_subagent_lc_agent_name_in_astream_metadata(self) -> None:
+ """Async variant of the #2925 streaming regression test.
+
+ The fix relies on `with_config` being symmetric across sync/async, but the
+ symptom in #2925 also shows up in `astream` — covering both paths guards
+ against an async-only regression.
+ """
+ subagent_content = "RAW_GRAPH_RESPONSE"
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {"description": "Do task", "subagent_type": "raw-worker"},
+ "id": "call_raw_worker_async",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done."),
+ ]
+ ),
+ stream_delimiter="_",
+ )
+ subagent_chat_model = GenericFakeChatModel(
+ messages=iter([AIMessage(content=subagent_content)]),
+ stream_delimiter="_",
+ )
+
+ builder = StateGraph(MessagesState)
+ builder.add_node("model", create_agent(model=subagent_chat_model))
+ builder.add_edge(START, "model")
+ raw_graph = builder.compile()
+
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ name="supervisor",
+ middleware=[SubAgentMiddleware(subagents=[CompiledSubAgent(name="raw-worker", description="Raw graph subagent.", runnable=raw_graph)])],
+ )
+
+ seen_subagent_names: set[str | None] = set()
+
+ async for _ns, stream_mode, data in parent_agent.astream(
+ {"messages": [HumanMessage(content="Do something")]},
+ stream_mode=["messages"],
+ subgraphs=True,
+ config={"configurable": {"thread_id": "test_raw_graph_lc_agent_name_async"}},
+ ):
+ if stream_mode == "messages":
+ message_chunk, metadata = data
+ if message_chunk.content:
+ seen_subagent_names.add(metadata.get("lc_agent_name"))
+
+ assert "raw-worker" in seen_subagent_names, f"Expected 'raw-worker' in async streamed lc_agent_name metadata, got: {seen_subagent_names}"
+
+ def test_compiled_subagent_name_overrides_inner_runnable_name_in_stream(self) -> None:
+ """CompiledSubAgent.name takes precedence over the inner runnable's lc_agent_name in streamed chunks.
+
+ When the inner runnable was itself created via `create_agent(name=...)`, the
+ registered `CompiledSubAgent.name` is what the parent uses to reference the
+ subagent and what tracing consumers display. This precedence is verified at
+ the tool-runtime layer in `test_subagent_propagates_recursion_limit_to_tool_runtime`;
+ this test pins the same precedence in the streamed-chunk metadata surface
+ from #2925 so a future "fix" that swaps merge order can't silently regress it.
+ """
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {"description": "Do task", "subagent_type": "outer-name"},
+ "id": "call_named_inner",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done."),
+ ]
+ ),
+ stream_delimiter="_",
+ )
+ subagent_chat_model = GenericFakeChatModel(
+ messages=iter([AIMessage(content="NAMED_INNER_RESPONSE")]),
+ stream_delimiter="_",
+ )
+
+ named_inner = create_agent(model=subagent_chat_model, name="inner-name")
+
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ name="supervisor",
+ middleware=[SubAgentMiddleware(subagents=[CompiledSubAgent(name="outer-name", description="Subagent with a different inner name.", runnable=named_inner)])],
+ )
+
+ seen_subagent_names: set[str | None] = set()
+
+ for _ns, stream_mode, data in parent_agent.stream(
+ {"messages": [HumanMessage(content="Do something")]},
+ stream_mode=["messages"],
+ subgraphs=True,
+ config={"configurable": {"thread_id": "test_outer_name_wins"}},
+ ):
+ if stream_mode == "messages":
+ message_chunk, metadata = data
+ if message_chunk.content:
+ seen_subagent_names.add(metadata.get("lc_agent_name"))
+
+ assert "outer-name" in seen_subagent_names, f"Expected 'outer-name' in streamed lc_agent_name metadata, got: {seen_subagent_names}"
+ assert "inner-name" not in seen_subagent_names, f"Inner runnable's lc_agent_name leaked into stream metadata: {seen_subagent_names}"
+
+ def test_config_passed_to_runnable_lambda_subagent(self) -> None:
+ """Test that config (including tags) is passed to a RunnableLambda subagent.
+
+ RunnableLambda doesn't have a 'config' attribute, so this tests the safe getattr fallback.
+ """
+ received_configs: list[RunnableConfig] = []
+
+ def lambda_subagent(state: dict[str, Any], config: RunnableConfig) -> dict[str, Any]: # noqa: ARG001
+ received_configs.append(config)
+ return {"messages": [AIMessage(content="Lambda response")]}
+
+ runnable_lambda = RunnableLambda(lambda_subagent)
+ assert not hasattr(runnable_lambda, "config")
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {"description": "Do something", "subagent_type": "lambda-agent"},
+ "id": "call_lambda",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done."),
+ ]
+ )
+ )
+
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ name="parent",
+ middleware=[SubAgentMiddleware(subagents=[CompiledSubAgent(name="lambda-agent", description="Lambda subagent.", runnable=runnable_lambda)])],
+ )
+
+ test_tags = ["lambda-tag", "config-test"]
+ parent_agent.invoke(
+ {"messages": [HumanMessage(content="Do something")]},
+ config={"configurable": {"thread_id": "test_lambda"}, "tags": test_tags},
+ )
+
+ assert len(received_configs) > 0, "Lambda should have been invoked"
+ assert all(t in received_configs[0].get("tags", []) for t in test_tags), f"Missing tags in config: {received_configs[0].get('tags')}"
+
+ @pytest.mark.filterwarnings("ignore:Pydantic serializer warnings:UserWarning")
+ def test_context_passed_to_subagent_tool_runtime(self) -> None:
+ """Test that context passed to main agent is available in subagent's ToolRuntime.context."""
+ received_contexts: list[Any] = []
+
+ @tool
+ def capture_context(query: str, runtime: ToolRuntime) -> str:
+ """Captures runtime context."""
+ received_contexts.append(runtime.context)
+ return f"Processed: {query}"
+
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {"description": "Use capture_context", "subagent_type": "ctx-agent"},
+ "id": "call_ctx",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done."),
+ ]
+ )
+ )
+ subagent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "capture_context",
+ "args": {"query": "test"},
+ "id": "call_tool",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Captured."),
+ ]
+ )
+ )
+
+ compiled_subagent = create_agent(model=subagent_chat_model, tools=[capture_context], name="ctx-agent")
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ name="orchestrator",
+ middleware=[SubAgentMiddleware(subagents=[CompiledSubAgent(name="ctx-agent", description="Context-aware subagent.", runnable=compiled_subagent)])],
+ )
+
+ test_context = {"user_id": "user-123", "session_id": "session-456"}
+ parent_agent.invoke(
+ {"messages": [HumanMessage(content="Process")]},
+ config={"configurable": {"thread_id": "test_context"}},
+ context=test_context,
+ )
+
+ assert len(received_contexts) > 0, "Subagent tool should have been invoked"
+ assert received_contexts[0] == test_context, f"Expected {test_context}, got {received_contexts[0]}"
+
+ def test_compiled_subagent_without_messages_raises_error(self) -> None:
+ """Test that a CompiledSubAgent without 'messages' in state raises a clear error.
+
+ This test verifies that when a custom StateGraph is used with CompiledSubAgent
+ and doesn't include a 'messages' key in its state, a helpful ValueError is raised
+ explaining the requirement.
+ """
+
+ # Define a custom state without 'messages' key
+ class CustomState(TypedDict):
+ custom_field: str
+
+ def custom_node(_state: CustomState) -> CustomState:
+ return {"custom_field": "processed"}
+
+ # Build a custom graph that doesn't use messages
+ graph_builder = StateGraph(CustomState)
+ graph_builder.add_node("process", custom_node)
+ graph_builder.add_edge(START, "process")
+ graph_builder.add_edge("process", END)
+ custom_graph = graph_builder.compile()
+
+ # Create parent agent with this custom subagent
+ parent_chat_model = GenericFakeChatModel(
+ messages=iter(
+ [
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Process something",
+ "subagent_type": "custom-processor",
+ },
+ "id": "call_custom",
+ }
+ ],
+ ),
+ ]
+ )
+ )
+
+ parent_agent = create_agent(
+ model=parent_chat_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ CompiledSubAgent(
+ name="custom-processor",
+ description="A custom processor",
+ runnable=custom_graph,
+ )
+ ])],
+ )
+
+ # Attempting to invoke should raise a clear error about missing 'messages' key
+ with pytest.raises(
+ ValueError,
+ match="CompiledSubAgent must return a dict containing a 'messages' key",
+ ):
+ parent_agent.invoke(
+ {"messages": [HumanMessage(content="Process this")]},
+ config={"configurable": {"thread_id": "test_thread_no_messages"}},
+ )
+
+ def test_ls_agent_type_is_trace_only_metadata(self) -> None:
+ """`ls_agent_type` must reach LangSmith but not streamed callback metadata.
+
+ The task tool wraps each subagent invocation in a langsmith
+ `tracing_context` with `metadata={"ls_agent_type": "subagent"}` so
+ downstream LangSmith tracing can distinguish subagent runs from
+ root-agent runs. Because this metadata is set via langsmith's tracing
+ contextvar (not via RunnableConfig), it only reaches the
+ `LangChainTracer` — it is not added to the callback manager's
+ metadata and therefore does not leak into streamed callback events.
+ """
+ # Root model: first call emits a task tool call that dispatches to the
+ # subagent; subsequent calls return a final response.
+ root_model = _ScriptedChatModel(
+ responses=[
+ AIMessage(
+ content="",
+ tool_calls=[
+ {
+ "name": "task",
+ "args": {
+ "description": "Do some work",
+ "subagent_type": "test-worker",
+ },
+ "id": "call_test_worker",
+ "type": "tool_call",
+ }
+ ],
+ ),
+ AIMessage(content="Done."),
+ ]
+ )
+ subagent_model = _ScriptedChatModel(responses=[AIMessage(content="Subagent completed the task.")])
+
+ # Capture streamed callback metadata per-run.
+ captured_callbacks: list[dict[str, Any]] = []
+
+ class CaptureHandler(BaseCallbackHandler):
+ def on_chain_start(
+ self,
+ serialized: dict[str, Any],
+ inputs: dict[str, Any],
+ *,
+ run_id: str,
+ parent_run_id: str | None = None,
+ tags: list[str] | None = None,
+ metadata: dict[str, Any] | None = None,
+ **kwargs: Any,
+ ) -> None:
+ captured_callbacks.append(
+ {
+ "name": kwargs.get("name") or (serialized or {}).get("name"),
+ "metadata": metadata or {},
+ }
+ )
+
+ mock_session = MagicMock()
+ mock_client = Client(session=mock_session, api_key="test", auto_batch_tracing=False)
+
+ agent = create_agent(
+ model=root_model,
+ checkpointer=InMemorySaver(),
+ middleware=[SubAgentMiddleware(subagents=[
+ SubAgent(
+ name="test-worker",
+ description="A test worker subagent.",
+ system_prompt="You are a test worker.",
+ model=subagent_model,
+ )
+ ])],
+ )
+
+ with tracing_context(client=mock_client, enabled=True):
+ agent.invoke(
+ {"messages": [HumanMessage(content="Please do some work")]},
+ config={
+ "configurable": {"thread_id": "test_ls_agent_type"},
+ "callbacks": [CaptureHandler()],
+ },
+ )
+
+ # The subagent path must have run (otherwise the trace-only check below
+ # is trivially true).
+ subagent_callback = next((c for c in captured_callbacks if c["name"] == "test-worker"), None)
+ assert subagent_callback is not None, f"Expected a 'test-worker' subagent callback, got names: {[c['name'] for c in captured_callbacks]}"
+
+ # (1) ls_agent_type must not leak into any streamed callback metadata.
+ for captured in captured_callbacks:
+ assert "ls_agent_type" not in captured["metadata"], (
+ f"ls_agent_type leaked into callback metadata for run {captured['name']!r}: {captured['metadata']}"
+ )
+
+ # (2) ls_agent_type='subagent' must reach the LangSmith tracer.
+ posts: list[dict[str, Any]] = []
+ for call in mock_session.request.mock_calls:
+ if call.args and call.args[0] == "POST":
+ body = json.loads(call.kwargs["data"])
+ posts.extend(body.get("post", []) if "post" in body else [body])
+
+ subagent_tracer_metadatas = [
+ post.get("extra", {}).get("metadata", {})
+ for post in posts
+ if post.get("extra", {}).get("metadata", {}).get("ls_agent_type") == "subagent"
+ ]
+ assert subagent_tracer_metadatas, (
+ f"Expected at least one LangSmith post with ls_agent_type='subagent'. "
+ f"Got tracer metadatas: "
+ f"{[p.get('extra', {}).get('metadata', {}) for p in posts]}"
+ )
diff --git a/libs/langchain_v1/uv.lock b/libs/langchain_v1/uv.lock
index f38a1a570b..17decb4b55 100644
--- a/libs/langchain_v1/uv.lock
+++ b/libs/langchain_v1/uv.lock
@@ -2123,7 +2123,7 @@ typing = [
[[package]]
name = "langchain-anthropic"
-version = "1.4.8"
+version = "1.5.0"
source = { editable = "../partners/anthropic" }
dependencies = [
{ name = "anthropic" },
@@ -2465,7 +2465,7 @@ wheels = [
[[package]]
name = "langchain-openai"
-version = "1.3.5"
+version = "1.4.0"
source = { editable = "../partners/openai" }
dependencies = [
{ name = "langchain-core" },