From 9ada2f5448c02552a61006a6117940c01d22077d Mon Sep 17 00:00:00 2001 From: bochencwx Date: Fri, 10 Jul 2026 11:34:10 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=8A=A8=E6=80=81=E5=88=9B=E5=BB=BA?= =?UTF-8?q?=E5=AD=90Agent=E6=94=AF=E6=8C=81=E5=9C=A8=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E6=97=B6=E4=BA=A7=E7=94=9F=E7=9A=84=E4=BA=8B=E4=BB=B6=E7=9B=B4?= =?UTF-8?q?=E6=8E=A5=E8=BD=AC=E5=8F=91=E5=88=B0=E7=88=B6=E6=B5=81=E7=A8=8B?= =?UTF-8?q?=E7=9A=84=E4=BA=8B=E4=BB=B6=E6=B5=81=E4=B8=AD=E7=9A=84=E8=83=BD?= =?UTF-8?q?=E5=8A=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/mkdocs/en/sub_agent.md | 25 ++ docs/mkdocs/zh/sub_agent.md | 23 ++ examples/dynamic_subagent/agent/agent.py | 9 +- examples/dynamic_subagent/run_agent.py | 45 ++- examples/spawn_subagent/agent/agent.py | 19 +- examples/spawn_subagent/run_agent.py | 44 +++ .../sub_agent/test_dynamic_sub_agent_tool.py | 51 +++ .../sub_agent/test_spawn_sub_agent_tool.py | 53 +++ .../test_subagent_event_forwarding.py | 317 ++++++++++++++++++ .../sub_agent/_dynamic_sub_agent_tool.py | 90 ++++- trpc_agent_sdk/agents/sub_agent/_runner.py | 156 ++++++++- .../agents/sub_agent/_spawn_sub_agent_tool.py | 89 ++++- .../agents/sub_agent/_sub_agent_config.py | 6 + 13 files changed, 885 insertions(+), 42 deletions(-) create mode 100644 tests/agents/sub_agent/test_subagent_event_forwarding.py diff --git a/docs/mkdocs/en/sub_agent.md b/docs/mkdocs/en/sub_agent.md index 00e8d79f..f9ebce63 100644 --- a/docs/mkdocs/en/sub_agent.md +++ b/docs/mkdocs/en/sub_agent.md @@ -21,6 +21,19 @@ A **short-lived sub-agent** is a natural fit for these problems: a fresh context The difference is *who defines the role*: the developer (Spawn) or the LLM (Dynamic). +### How this differs from other multi-agent mechanisms + +The framework already offers several ways to compose agents (see [Multi Agents](multi_agents.md)). Spawned Sub-Agents solve a different problem: + +| Mechanism | Agents involved | Who decides when to invoke | Context | Typical use | +| --- | --- | --- | --- | --- | +| **Chain / Parallel / Cycle Agent** | **Pre-built** fixed agent instances | **Deterministic** orchestration — run in list order / in parallel / in a loop, regardless of input | Each agent independent | Fixed multi-step workflows | +| **Sub Agents (transfer)** | Pre-registered agents | Parent **transfers control** at runtime; the sub-agent then takes over the conversation | Shared session | **Hand off** the whole conversation to a better-suited agent | +| **AgentTool** | Wraps **an existing agent instance** as a tool | Parent LLM calls it on demand | Shares/syncs state & artifacts back to parent | Reuse a **specific, already-built** agent | +| **Spawned Sub-Agents** | **Created on the fly** per call, destroyed after | Parent LLM calls it on demand | **Strictly isolated**: fresh ephemeral session, history/state not shared by default | Delegate a **one-off** subtask while keeping the parent context clean | + +In one line: **Chain/Parallel/Cycle** deterministically orchestrate a fixed set of agents; **transfer** hands the conversation off; **AgentTool** reuses one existing agent as a tool; while **Spawned Sub-Agents** create an **isolated, short-lived** sub-agent on the spot for a single task and discard it afterward — the emphasis is on **on-demand runtime creation** and **context isolation**, not reusing an existing agent or transferring control. + ## Quick Start ```python @@ -173,8 +186,19 @@ class SubAgentConfig: max_turns: int | None = None """Max LLM calls the sub-agent may make. None = unlimited.""" + + forward_events: bool = False + """Whether to forward the sub-agent's execution events to the parent + runner's consumer as progress updates. + + True: the orchestrator can display the sub-agent's execution live (model + output, tool calls, tool results); the parent agent's LLM still receives + only the sub-agent's final result. False (default): the sub-agent runs + silently and only its final result is returned.""" ``` +Forwarded events reach the consumer as progress events; they are **not** written to the parent session and **never** enter the parent agent's LLM context. Consumers identify them via `tool_progress=True` on `event.custom_metadata` and read the execution from `payload` (`author` / `partial` / `content`, plus optional `error` / `usage`). + ## Usage ### SpawnSubAgentTool @@ -270,3 +294,4 @@ orchestrator = LlmAgent( - **Session isolation**: sub-agents run in a fresh ephemeral session. Parent history is not shared by default; opt in via `include_parent_history=True`. - **Nesting**: 1-level hard cap. Sub-agents cannot spawn further sub-agents. - **Result shape**: the sub-agent's final text is returned as the tool result string. +- **Live execution (`forward_events`)**: set `SubAgentConfig(forward_events=True)` to stream the sub-agent's execution to the parent runner's consumer for display. Forwarded events are progress events — they never enter the parent LLM's context, which still receives only the final result. Consumers detect them via `tool_progress=True` on `event.custom_metadata` and read `payload`. See `examples/dynamic_subagent` and `examples/spawn_subagent` for a working consumer. diff --git a/docs/mkdocs/zh/sub_agent.md b/docs/mkdocs/zh/sub_agent.md index 568bbf11..09df29a4 100644 --- a/docs/mkdocs/zh/sub_agent.md +++ b/docs/mkdocs/zh/sub_agent.md @@ -21,6 +21,19 @@ 区别在于**谁定义角色**：开发者（Spawn）还是 LLM（Dynamic）。 +### 与框架其他多 agent 机制的区别 + +框架已有几种组合 agent 的方式（详见 [Multi Agents](multi_agents.md)）。Spawned Sub-Agents 与它们解决的是不同问题： + +| 机制 | 参与的 agent | 谁决定何时调用 | 上下文 | 典型用途 | +| --- | --- | --- | --- | --- | +| **Chain / Parallel / Cycle Agent** | **预先构建**的固定 agent 实例 | **确定性**编排——按列表顺序/并行/循环执行，与输入无关 | 各 agent 独立 | 固定的多步工作流 | +| **Sub Agents（transfer）** | 预先注册的 agent | 父 agent 运行时**转移控制权**，之后由子 agent 接管对话 | 共享同一会话 | 把整段对话**移交**给更合适的 agent | +| **AgentTool** | 把**某个已有 agent 实例**包成工具 | 父 LLM 按需调用 | 会共享/同步 state 与 artifact 回父 | 复用一个**具体的、已存在的** agent | +| **Spawned Sub-Agents** | **调用时临时创建**、用完即销毁 | 父 LLM 按需调用 | **严格隔离**：全新临时会话，默认不共享历史/state | 委派**一次性**子任务，保持父上下文干净 | + +一句话概括：**Chain/Parallel/Cycle** 是"确定性编排一组固定 agent"；**transfer** 是"把对话交出去"；**AgentTool** 是"把一个已有 agent 当工具复用"；而 **Spawned Sub-Agents** 是"为单次任务**现场造一个隔离的、短命的**子 agent，跑完就丢"——强调的是**运行时按需创建**与**上下文隔离**，而非复用既有 agent 或转移控制。 + ## Quick Start ```python @@ -173,8 +186,17 @@ class SubAgentConfig: max_turns: int | None = None """子 agent 最多可发起的 LLM 调用次数。None = 不限制。""" + + forward_events: bool = False + """是否将子 agent 的执行事件转发给父 runner 的消费者，作为进度更新。 + + True：编排层可实时展示子 agent 的执行（模型输出、工具调用、工具结果）， + 父 agent 的 LLM 仍只收到子 agent 的最终结果。 + False（默认）：子 agent 静默执行，只回传最终结果。""" ``` +转发的事件以进度事件形式到达消费者，**不会**写入父会话、也**不会**进入父 agent 的 LLM 上下文；消费者通过 `event.custom_metadata` 上的 `tool_progress=True` 识别它们，并从 `payload` 读取执行内容（`author` / `partial` / `content` / 可选 `error` / `usage`）。 + ## 使用方式 ### SpawnSubAgentTool @@ -270,3 +292,4 @@ orchestrator = LlmAgent( - **会话隔离**：子 agent 在全新临时会话中运行，默认不共享父会话历史。通过 `include_parent_history=True` 可注入。 - **嵌套限制**：1 层硬限，子 agent 无法再次 spawn。 - **结果形态**：子 agent 的最终文本作为 tool result 字符串返回。 +- **实时执行（`forward_events`）**：设置 `SubAgentConfig(forward_events=True)` 可将子 agent 的执行流转发给父 runner 的消费者用于展示。转发事件为进度事件——它们不会进入父 agent 的 LLM 上下文（父仍只收到最终结果）。消费者通过 `event.custom_metadata` 上的 `tool_progress=True` 识别它们并读取 `payload`。可运行示例见 `examples/dynamic_subagent` 与 `examples/spawn_subagent`。 diff --git a/examples/dynamic_subagent/agent/agent.py b/examples/dynamic_subagent/agent/agent.py index af661a84..fc8e10b1 100644 --- a/examples/dynamic_subagent/agent/agent.py +++ b/examples/dynamic_subagent/agent/agent.py @@ -45,7 +45,12 @@ def create_minimal_agent() -> LlmAgent: temperature=0.7, max_output_tokens=2000, ), - tools=workspace_tools + [DynamicSubAgentTool()], + tools=workspace_tools + [ + DynamicSubAgentTool( + # Stream the sub-agent's execution to the parent consumer. + agent_config=SubAgentConfig(forward_events=True), + ), + ], ) @@ -69,6 +74,8 @@ def create_bounded_agent() -> LlmAgent: temperature=0.3, max_output_tokens=1000, ), + # Stream the sub-agent's execution to the parent consumer. + forward_events=True, ), ), ], diff --git a/examples/dynamic_subagent/run_agent.py b/examples/dynamic_subagent/run_agent.py index d0350007..18031dd5 100644 --- a/examples/dynamic_subagent/run_agent.py +++ b/examples/dynamic_subagent/run_agent.py @@ -33,7 +33,6 @@ def _truncate(text: str, max_len: int = 200) -> str: """Truncate long tool output for display.""" - return text if not isinstance(text, str): text = str(text) if len(text) <= max_len: @@ -41,6 +40,38 @@ def _truncate(text: str, max_len: int = 200) -> str: return text[:max_len] + f"\n... (truncated, total {len(text)} chars)" +def _print_subagent_progress(payload: dict) -> None: + """Render one forwarded sub-agent execution event. + + ``payload`` is a :class:`SubAgentProgress` dict: ``author`` / ``partial``, + the framework-native ``content`` dump (``parts`` with ``function_call`` / + ``function_response`` / ``text`` / ``thought``), and optional ``error`` / + ``usage``. Indented under the parent output so the sub-agent's steps are + visually distinct from the orchestrator's. + """ + name = payload.get("author") or "subagent" + # Errors first — always surface them, even on an otherwise-partial event. + err = payload.get("error") + if err: + print(f"\n \U0001F9E9 [{name}] !! error {err.get('code')}: {err.get('message')}") + if payload.get("partial"): + # Skip streaming text deltas to keep the demo output readable; the + # non-partial steps below already summarize the sub-agent's work. + return + parts = (payload.get("content") or {}).get("parts") or [] + has_calls = any(p.get("function_call") or p.get("function_response") for p in parts) + for p in parts: + fc = p.get("function_call") + if fc: + print(f"\n \U0001F9E9 [{name}] -> tool {fc.get('name')}({_truncate(fc.get('args'))})") + fr = p.get("function_response") + if fr: + print(f" \U0001F9E9 [{name}] <- {_truncate(fr.get('response'))}") + text = p.get("text") + if text and not p.get("thought") and not has_calls: + print(f"\n \U0001F9E9 [{name}] {_truncate(text)}") + + _QUERIES = { "minimal": [ # Simple task: orchestrator may call word_count directly. @@ -93,6 +124,18 @@ async def run_demo(mode: str): session_id=current_session_id, new_message=user_content, ): + # Forwarded sub-agent execution events (SubAgentConfig + # forward_events=True). These are partial progress events carrying + # the sub-agent's own steps under custom_metadata.payload; they + # never reach the parent LLM's context. This demo registers only the + # sub-agent tool, so tool_progress alone identifies them; an app with + # several progress tools would also branch on meta["tool_name"]. + meta = event.custom_metadata or {} + payload = meta.get("payload") + if meta.get("tool_progress") and isinstance(payload, dict): + _print_subagent_progress(payload) + continue + if event.content and event.content.parts and event.author != "user": if event.partial: for part in event.content.parts: diff --git a/examples/spawn_subagent/agent/agent.py b/examples/spawn_subagent/agent/agent.py index e5e7ebc7..7fd99360 100644 --- a/examples/spawn_subagent/agent/agent.py +++ b/examples/spawn_subagent/agent/agent.py @@ -25,6 +25,7 @@ from trpc_agent_sdk.agents.sub_agent import EXPLORE_AGENT from trpc_agent_sdk.agents.sub_agent import PLAN_AGENT from trpc_agent_sdk.agents.sub_agent import SubAgentArchetype +from trpc_agent_sdk.agents.sub_agent import SubAgentConfig from trpc_agent_sdk.tools import SpawnSubAgentTool from trpc_agent_sdk.models import LLMModel from trpc_agent_sdk.models import OpenAIModel @@ -52,7 +53,11 @@ def create_default_agent() -> LlmAgent: description="Coding assistant with spawn_subagent in zero-config mode.", model=_create_model(), instruction=INSTRUCTION, - tools=[ReadTool(), GlobTool(), GrepTool(), SpawnSubAgentTool()], + tools=[ReadTool(), GlobTool(), GrepTool(), + SpawnSubAgentTool( + # Stream the sub-agent's execution to the parent consumer. + agent_config=SubAgentConfig(forward_events=True), + )], ) @@ -88,7 +93,11 @@ def create_code_agent() -> LlmAgent: instruction=INSTRUCTION, tools=[ ReadTool(), GlobTool(), GrepTool(), - SpawnSubAgentTool(agents=[_SECURITY_AUDITOR, EXPLORE_AGENT, PLAN_AGENT]), + SpawnSubAgentTool( + agents=[_SECURITY_AUDITOR, EXPLORE_AGENT, PLAN_AGENT], + # Stream the sub-agent's execution to the parent consumer. + agent_config=SubAgentConfig(forward_events=True), + ), ], ) @@ -109,7 +118,11 @@ def create_md_agent() -> LlmAgent: instruction=INSTRUCTION, tools=[ ReadTool(), GlobTool(), GrepTool(), - SpawnSubAgentTool(agents=[EXPLORE_AGENT, PLAN_AGENT], agent_paths=[_AGENTS_PATH]), + SpawnSubAgentTool( + agents=[EXPLORE_AGENT, PLAN_AGENT], agent_paths=[_AGENTS_PATH], + # Stream the sub-agent's execution to the parent consumer. + agent_config=SubAgentConfig(forward_events=True), + ), ], ) diff --git a/examples/spawn_subagent/run_agent.py b/examples/spawn_subagent/run_agent.py index ca72b96a..90dc2519 100644 --- a/examples/spawn_subagent/run_agent.py +++ b/examples/spawn_subagent/run_agent.py @@ -45,6 +45,38 @@ def _truncate(text: str, max_len: int = 200) -> str: return text[:max_len] + f"\n... (truncated, total {len(text)} chars)" +def _print_subagent_progress(payload: dict) -> None: + """Render one forwarded sub-agent execution event. + + ``payload`` is a :class:`SubAgentProgress` dict: ``author`` / ``partial``, + the framework-native ``content`` dump (``parts`` with ``function_call`` / + ``function_response`` / ``text`` / ``thought``), and optional ``error`` / + ``usage``. Indented under the parent output so the sub-agent's steps are + visually distinct from the assistant's. + """ + name = payload.get("author") or "subagent" + # Errors first — always surface them, even on an otherwise-partial event. + err = payload.get("error") + if err: + print(f"\n \U0001F9E9 [{name}] !! error {err.get('code')}: {err.get('message')}") + if payload.get("partial"): + # Skip streaming text deltas to keep the demo output readable; the + # non-partial steps below already summarize the sub-agent's work. + return + parts = (payload.get("content") or {}).get("parts") or [] + has_calls = any(p.get("function_call") or p.get("function_response") for p in parts) + for p in parts: + fc = p.get("function_call") + if fc: + print(f"\n \U0001F9E9 [{name}] -> tool {fc.get('name')}({_truncate(fc.get('args'))})") + fr = p.get("function_response") + if fr: + print(f" \U0001F9E9 [{name}] <- {_truncate(fr.get('response'))}") + text = p.get("text") + if text and not p.get("thought") and not has_calls: + print(f"\n \U0001F9E9 [{name}] {_truncate(text)}") + + # Queries per mode — simple tasks (parent handles directly) vs complex # tasks (delegated to a sub-agent). _SHARED_AGENT_QUERIES = [ @@ -111,6 +143,18 @@ async def run_demo(mode: str): session_id=current_session_id, new_message=user_content, ): + # Forwarded sub-agent execution events (SubAgentConfig + # forward_events=True). These are partial progress events carrying + # the sub-agent's own steps under custom_metadata.payload; they + # never reach the parent LLM's context. This demo registers only the + # sub-agent tool, so tool_progress alone identifies them; an app with + # several progress tools would also branch on meta["tool_name"]. + meta = event.custom_metadata or {} + payload = meta.get("payload") + if meta.get("tool_progress") and isinstance(payload, dict): + _print_subagent_progress(payload) + continue + if event.content and event.content.parts and event.author != "user": if event.partial: for part in event.content.parts: diff --git a/tests/agents/sub_agent/test_dynamic_sub_agent_tool.py b/tests/agents/sub_agent/test_dynamic_sub_agent_tool.py index a48f0d87..6d25450c 100644 --- a/tests/agents/sub_agent/test_dynamic_sub_agent_tool.py +++ b/tests/agents/sub_agent/test_dynamic_sub_agent_tool.py @@ -39,6 +39,57 @@ def test_constructor_with_config() -> None: def test_constructor_skip_summarization() -> None: t = DynamicSubAgentTool(skip_summarization=True) assert t._skip_summarization is True + # Exposed as a property for the progress-streaming execution path. + assert t.skip_summarization is True + + +def test_is_progress_streaming_default_off() -> None: + """Without forward_events, the tool runs on the non-streaming path.""" + assert DynamicSubAgentTool().is_progress_streaming is False + assert DynamicSubAgentTool(agent_config=SubAgentConfig()).is_progress_streaming is False + + +def test_is_progress_streaming_on_when_forwarding() -> None: + t = DynamicSubAgentTool(agent_config=SubAgentConfig(forward_events=True)) + assert t.is_progress_streaming is True + + +@pytest.mark.asyncio +async def test_run_streaming_empty_prompt_yields_error() -> None: + """run_streaming surfaces the prompt validation error as its only value.""" + t = DynamicSubAgentTool(agent_config=SubAgentConfig(forward_events=True)) + ctx = MagicMock() + yielded = [v async for v in t.run_streaming(tool_context=ctx, args={"prompt": " "})] + assert len(yielded) == 1 + assert yielded[0]["status"] == "error" + assert "prompt" in yielded[0]["message"] + + +@pytest.mark.asyncio +async def test_run_streaming_forwards_projections_then_result() -> None: + """run_streaming delegates to run_subagent_streaming, yielding its values in order.""" + from unittest.mock import patch + + t = DynamicSubAgentTool(agent_config=SubAgentConfig(forward_events=True)) + ctx = MagicMock() + + async def _fake_stream(**kwargs): + yield {"author": "subagent_dynamic", "partial": True, "content": {"parts": [{"text": "step 1"}]}} + yield "final result" + + with patch( + "trpc_agent_sdk.agents.sub_agent._dynamic_sub_agent_tool.run_subagent_streaming", + _fake_stream, + ): + yielded = [v async for v in t.run_streaming( + tool_context=ctx, + args={"instruction": "You are a helper.", "prompt": "do it"}, + )] + + assert yielded == [ + {"author": "subagent_dynamic", "partial": True, "content": {"parts": [{"text": "step 1"}]}}, + "final result", + ] def test_constructor_custom_name() -> None: diff --git a/tests/agents/sub_agent/test_spawn_sub_agent_tool.py b/tests/agents/sub_agent/test_spawn_sub_agent_tool.py index 6758b428..e71aa3b7 100644 --- a/tests/agents/sub_agent/test_spawn_sub_agent_tool.py +++ b/tests/agents/sub_agent/test_spawn_sub_agent_tool.py @@ -267,3 +267,56 @@ async def test_skip_summarization_sets_event_action() -> None: args={"subagent_type": "nope", "prompt": "hi", "description": "x"}, ) assert ctx.event_actions.skip_summarization is True + + +def test_is_progress_streaming_default_off() -> None: + """Without forward_events, spawn runs on the non-streaming path.""" + assert SpawnSubAgentTool().is_progress_streaming is False + assert SpawnSubAgentTool(agent_config=SubAgentConfig()).is_progress_streaming is False + + +def test_is_progress_streaming_on_when_forwarding() -> None: + t = SpawnSubAgentTool(agent_config=SubAgentConfig(forward_events=True)) + assert t.is_progress_streaming is True + # skip_summarization is exposed as a property for the streaming path. + assert SpawnSubAgentTool(skip_summarization=True).skip_summarization is True + + +@pytest.mark.asyncio +async def test_run_streaming_unknown_type_no_default_yields_error() -> None: + """run_streaming surfaces the resolve error as its only value.""" + t = SpawnSubAgentTool(with_default=False, agent_config=SubAgentConfig(forward_events=True)) + ctx = _make_tool_context() + yielded = [v async for v in t.run_streaming( + tool_context=ctx, + args={"subagent_type": "nope", "prompt": "hi", "description": "x"}, + )] + assert len(yielded) == 1 + assert yielded[0]["status"] == "error" + + +@pytest.mark.asyncio +async def test_run_streaming_forwards_projections_then_result() -> None: + """run_streaming delegates to run_subagent_streaming, yielding its values in order.""" + from unittest.mock import patch + + t = SpawnSubAgentTool(agent_config=SubAgentConfig(forward_events=True)) + ctx = _make_tool_context() + + async def _fake_stream(**kwargs): + yield {"author": "subagent_default", "partial": True, "content": {"parts": [{"text": "step 1"}]}} + yield "final result" + + with patch( + "trpc_agent_sdk.agents.sub_agent._spawn_sub_agent_tool.run_subagent_streaming", + _fake_stream, + ): + yielded = [v async for v in t.run_streaming( + tool_context=ctx, + args={"subagent_type": "default", "prompt": "do it", "description": "x"}, + )] + + assert yielded == [ + {"author": "subagent_default", "partial": True, "content": {"parts": [{"text": "step 1"}]}}, + "final result", + ] diff --git a/tests/agents/sub_agent/test_subagent_event_forwarding.py b/tests/agents/sub_agent/test_subagent_event_forwarding.py new file mode 100644 index 00000000..fc85cdfc --- /dev/null +++ b/tests/agents/sub_agent/test_subagent_event_forwarding.py @@ -0,0 +1,317 @@ +# Tencent is pleased to support the open source community by making trpc-agent-python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# trpc-agent-python is licensed under the Apache License Version 2.0. +# +"""Tests for sub-agent event forwarding (progress-streaming path). + +Covers the projection helper ``_project_subagent_event`` and the streaming +generator ``run_subagent_streaming``, plus the fact that ``run_subagent`` +delegates to it. The Runner is mocked so no real LLM is required — this mirrors +the mocking style in ``test_runner.py``. +""" + +from __future__ import annotations + +from typing import List +from unittest.mock import AsyncMock +from unittest.mock import MagicMock +from unittest.mock import patch + +import pytest + +from trpc_agent_sdk.agents.sub_agent import GENERAL_PURPOSE_AGENT +from trpc_agent_sdk.agents.sub_agent import SubAgentConfig +from trpc_agent_sdk.agents.sub_agent._runner import _project_subagent_event +from trpc_agent_sdk.agents.sub_agent._runner import run_subagent +from trpc_agent_sdk.agents.sub_agent._runner import run_subagent_streaming +from trpc_agent_sdk.exceptions import RunCancelledException +from trpc_agent_sdk.models import LLMModel +from trpc_agent_sdk.models import LlmResponse +from trpc_agent_sdk.models import ModelRegistry +from trpc_agent_sdk.tools import ReadTool +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import Part + + +class MockLLMModel(LLMModel): + @classmethod + def supported_models(cls) -> List[str]: + return [r"test-dynamic-.*"] + + async def _generate_async_impl(self, request, stream=False, ctx=None): + yield LlmResponse(content=None) + + def validate_request(self, request): + pass + + +@pytest.fixture(scope="module", autouse=True) +def _register_mock_model(): + original = ModelRegistry._registry.copy() + ModelRegistry.register(MockLLMModel) + yield + ModelRegistry._registry = original + + +def _parent_ctx_with_model(model: str) -> MagicMock: + parent_ctx = MagicMock() + parent_agent = MagicMock() + parent_agent.model = model + parent_agent.generate_content_config = None + parent_agent.parallel_tool_calls = False + parent_ctx.agent = parent_agent + parent_ctx.agent.tools = [ReadTool()] + parent_ctx.session.app_name = "test_app" + parent_ctx.artifact_service = None + return parent_ctx + + +# --- _project_subagent_event ------------------------------------------------- + + +def test_project_subagent_event_text_and_metadata() -> None: + event = MagicMock() + event.author = "subagent_general_purpose" + event.partial = True + event.error_code = None + event.error_message = None + event.usage_metadata = None + event.content = Content(role="model", parts=[Part.from_text(text="thinking...")]) + + payload = _project_subagent_event(event) + + assert payload["author"] == "subagent_general_purpose" + assert payload["partial"] is True + # content is the framework-native Content dump, not a bespoke shape. + assert payload["content"]["role"] == "model" + assert payload["content"]["parts"][0]["text"] == "thinking..." + assert "error" not in payload + assert "usage" not in payload + + +def test_project_subagent_event_captures_tool_calls_and_responses() -> None: + call_part = Part.from_function_call(name="calculator", args={"expr": "1+1"}) + call_part.function_call.id = "call-1" + resp_part = Part.from_function_response(name="calculator", response={"result": 2}) + resp_part.function_response.id = "call-1" + + event = MagicMock() + event.author = "subagent_general_purpose" + event.partial = False + event.error_code = None + event.error_message = None + event.usage_metadata = None + event.content = Content(role="model", parts=[call_part, resp_part]) + + payload = _project_subagent_event(event) + + parts = payload["content"]["parts"] + assert parts[0]["function_call"] == {"id": "call-1", "args": {"expr": "1+1"}, "name": "calculator"} + assert parts[1]["function_response"] == {"id": "call-1", "name": "calculator", "response": {"result": 2}} + + +def test_project_subagent_event_keeps_thought_in_content() -> None: + """Thought parts are preserved in content (with thought=True) for the consumer to render or hide.""" + thought = Part(text="internal reasoning", thought=True) + visible = Part.from_text(text="answer") + + event = MagicMock() + event.author = "sub" + event.partial = False + event.error_code = None + event.error_message = None + event.usage_metadata = None + event.content = Content(role="model", parts=[thought, visible]) + + payload = _project_subagent_event(event) + parts = payload["content"]["parts"] + assert parts[0] == {"text": "internal reasoning", "thought": True} + assert parts[1]["text"] == "answer" + + +def test_project_subagent_event_no_content() -> None: + event = MagicMock() + event.author = "sub" + event.partial = False + event.error_code = None + event.error_message = None + event.usage_metadata = None + event.content = None + + payload = _project_subagent_event(event) + assert "content" not in payload + assert payload["author"] == "sub" + + +def test_project_subagent_event_surfaces_error() -> None: + """When the event carries an error, it appears under the ``error`` key.""" + event = MagicMock() + event.author = "sub" + event.partial = False + event.error_code = "MAX_TOKENS" + event.error_message = "context length exceeded" + event.usage_metadata = None + event.content = None + + payload = _project_subagent_event(event) + assert payload["error"] == {"code": "MAX_TOKENS", "message": "context length exceeded"} + + +def test_project_subagent_event_no_error_key_when_clean() -> None: + """A clean event omits the ``error`` key entirely (payload stays lean).""" + event = MagicMock() + event.author = "sub" + event.partial = False + event.error_code = None + event.error_message = None + event.usage_metadata = None + event.content = Content(role="model", parts=[Part.from_text(text="ok")]) + + payload = _project_subagent_event(event) + assert "error" not in payload + + +def test_project_subagent_event_captures_usage() -> None: + """Token counts are lifted out of usage_metadata under the ``usage`` key.""" + usage = MagicMock() + usage.prompt_token_count = 120 + usage.candidates_token_count = 45 + usage.total_token_count = 165 + + event = MagicMock() + event.author = "sub" + event.partial = False + event.error_code = None + event.error_message = None + event.usage_metadata = usage + event.content = Content(role="model", parts=[Part.from_text(text="done")]) + + payload = _project_subagent_event(event) + assert payload["usage"] == {"prompt_tokens": 120, "completion_tokens": 45, "total_tokens": 165} + + +# --- run_subagent_streaming helpers ------------------------------------------ + + +def _make_model_event(text: str, *, partial: bool = False) -> MagicMock: + event = MagicMock() + event.content = Content(role="model", parts=[Part.from_text(text=text)]) + event.author = "subagent_general-purpose" + event.partial = partial + event.error_code = None + event.error_message = None + event.usage_metadata = None + event.is_error = MagicMock(return_value=False) + return event + + +def _mock_streaming_runner(event_stream: list) -> MagicMock: + async def _fake_run_async(*args, **kwargs): + for event in event_stream: + yield event + + mock_runner_instance = MagicMock() + mock_runner_instance.run_async = _fake_run_async + mock_runner_instance.session_service = MagicMock() + mock_runner_instance.session_service.create_session = AsyncMock() + mock_runner_instance.session_service.append_event = AsyncMock() + mock_runner_instance.artifact_service = None + mock_runner_instance.close = AsyncMock() + return mock_runner_instance + + +# --- run_subagent_streaming -------------------------------------------------- + + +@pytest.mark.asyncio +async def test_streaming_yields_projections_then_result() -> None: + """Each sub-agent event yields a projection dict; the last value is the result.""" + parent_ctx = _parent_ctx_with_model("test-dynamic-parent") + events = [_make_model_event("partial", partial=True), _make_model_event("Final answer.")] + mock_runner_instance = _mock_streaming_runner(events) + + with patch("trpc_agent_sdk.runners.Runner", MagicMock(return_value=mock_runner_instance)): + yielded = [v async for v in run_subagent_streaming( + parent_ctx=parent_ctx, + archetype=GENERAL_PURPOSE_AGENT, + prompt="Do something.", + )] + + # 2 projection dicts (one per event) + 1 final string result. + assert len(yielded) == 3 + assert all(isinstance(v, dict) and "content" in v for v in yielded[:2]) + assert yielded[-1] == "Final answer." + mock_runner_instance.close.assert_called_once() + + +@pytest.mark.asyncio +async def test_streaming_build_error_yields_error_dict_only() -> None: + """A build failure yields a single error dict as the final value (no projections).""" + parent_ctx = _parent_ctx_with_model("test-dynamic-parent") + parent_ctx.agent.model = None # force resolve_model to raise + parent_ctx.agent.tools = [] + + yielded = [v async for v in run_subagent_streaming( + parent_ctx=parent_ctx, + archetype=GENERAL_PURPOSE_AGENT, + prompt="Do something.", + )] + + assert len(yielded) == 1 + assert yielded[0]["status"] == "error" + + +@pytest.mark.asyncio +async def test_streaming_cancelled_final_value() -> None: + """Cancellation surfaces the marker as the final yielded value, not raised.""" + parent_ctx = _parent_ctx_with_model("test-dynamic-parent") + mock_runner_instance = _mock_streaming_runner([]) + mock_runner_instance.run_async = MagicMock(side_effect=RunCancelledException()) + + with patch("trpc_agent_sdk.runners.Runner", MagicMock(return_value=mock_runner_instance)): + yielded = [v async for v in run_subagent_streaming( + parent_ctx=parent_ctx, + archetype=GENERAL_PURPOSE_AGENT, + prompt="Do something.", + )] + + assert yielded[-1] == "[sub-agent cancelled]" + mock_runner_instance.close.assert_called_once() + + +@pytest.mark.asyncio +async def test_streaming_max_turns_note_in_final() -> None: + """max_turns stops streaming and folds the note into the final value.""" + parent_ctx = _parent_ctx_with_model("test-dynamic-parent") + events = [_make_model_event("Iteration 1."), _make_model_event("Iteration 2.")] + mock_runner_instance = _mock_streaming_runner(events) + + with patch("trpc_agent_sdk.runners.Runner", MagicMock(return_value=mock_runner_instance)): + yielded = [v async for v in run_subagent_streaming( + parent_ctx=parent_ctx, + archetype=GENERAL_PURPOSE_AGENT, + prompt="Do something.", + agent_config=SubAgentConfig(max_turns=1), + )] + + assert "[sub-agent stopped: max turns reached]" in yielded[-1] + mock_runner_instance.close.assert_called_once() + + +@pytest.mark.asyncio +async def test_run_subagent_delegates_to_streaming() -> None: + """run_subagent returns the same final value the streaming generator ends on.""" + parent_ctx = _parent_ctx_with_model("test-dynamic-parent") + events = [_make_model_event("partial", partial=True), _make_model_event("Final answer.")] + mock_runner_instance = _mock_streaming_runner(events) + + with patch("trpc_agent_sdk.runners.Runner", MagicMock(return_value=mock_runner_instance)): + result = await run_subagent( + parent_ctx=parent_ctx, + archetype=GENERAL_PURPOSE_AGENT, + prompt="Do something.", + ) + + assert result == "Final answer." diff --git a/trpc_agent_sdk/agents/sub_agent/_dynamic_sub_agent_tool.py b/trpc_agent_sdk/agents/sub_agent/_dynamic_sub_agent_tool.py index d1dec7f9..1f0a82f0 100644 --- a/trpc_agent_sdk/agents/sub_agent/_dynamic_sub_agent_tool.py +++ b/trpc_agent_sdk/agents/sub_agent/_dynamic_sub_agent_tool.py @@ -9,8 +9,11 @@ from __future__ import annotations from typing import Any +from typing import AsyncIterator from typing import List from typing import Optional +from typing import Tuple +from typing import Union from typing_extensions import override from trpc_agent_sdk.context import InvocationContext @@ -23,6 +26,7 @@ from ._archetype import SubAgentArchetype from ._runner import run_subagent +from ._runner import run_subagent_streaming from ._sub_agent_config import SubAgentConfig _DESCRIPTION = ("Run one short-lived sub-agent for a single focused task and return " @@ -86,8 +90,35 @@ def __init__( self._agent_config = agent_config self._skip_summarization = skip_summarization self._expose_tool_selection = expose_tool_selection + # When the config asks to forward the sub-agent's events, this tool + # behaves as a progress-streaming tool (see is_progress_streaming / + # run_streaming). Resolved once at construction because + # is_progress_streaming is read before execution to route the call + # onto the streaming path. + self._forward_events = bool(agent_config is not None and agent_config.forward_events) super().__init__(name=name, description=description or _DESCRIPTION, filters_name=filters_name, filters=filters) + @property + @override + def is_progress_streaming(self) -> bool: + """Route through the progress-streaming path when event forwarding is on. + + Statically determined by ``SubAgentConfig.forward_events`` so + the tools processor can partition this call onto the streaming path + (which drives :meth:`run_streaming`) before execution begins. + """ + return self._forward_events + + @property + def skip_summarization(self) -> bool: + """Whether the streamed sub-agent result is the parent's final answer. + + Read by the progress-streaming execution path (which bypasses + ``_run_async_impl``) to set ``skip_summarization`` on the final + ``function_response`` event. + """ + return self._skip_summarization + @override def _get_declaration(self) -> FunctionDeclaration: properties: dict = { @@ -155,16 +186,12 @@ async def process_request( "or constraints for this run.") llm_request.append_instructions([instruction]) - @override - async def _run_async_impl( - self, - *, - tool_context: InvocationContext, - args: dict[str, Any], - ) -> Any: - if self._skip_summarization: - tool_context.event_actions.skip_summarization = True + def _resolve_call(self, args: dict[str, Any]) -> Union[Tuple[SubAgentArchetype, str, Optional[list]], dict]: + """Parse and validate call args into ``(archetype, prompt, tool_filter)``. + Returns an error dict when ``prompt`` is missing/empty so both the + streaming and non-streaming paths surface the same failure shape. + """ instruction = args.get("instruction") prompt = args.get("prompt") @@ -192,6 +219,23 @@ async def _run_async_impl( instruction=instruction, tools=self._tools, # None=inherit, tuple=fixed set ) + return synthetic, prompt, tool_filter + + @override + async def _run_async_impl( + self, + *, + tool_context: InvocationContext, + args: dict[str, Any], + ) -> Any: + if self._skip_summarization: + tool_context.event_actions.skip_summarization = True + + resolved = self._resolve_call(args) + if isinstance(resolved, dict): + return resolved + synthetic, prompt, tool_filter = resolved + return await run_subagent( parent_ctx=tool_context, archetype=synthetic, @@ -200,6 +244,34 @@ async def _run_async_impl( tool_filter=tool_filter, ) + async def run_streaming( + self, + *, + tool_context: InvocationContext, + args: dict[str, Any], + ) -> AsyncIterator[Any]: + """Progress-streaming entrypoint used when event forwarding is enabled. + + Yields one progress projection per sub-agent event and, as the final + value, the sub-agent's result (which the tools processor turns into the + ``function_response`` fed back to the parent LLM). See + :func:`run_subagent_streaming` for the per-value contract. + """ + resolved = self._resolve_call(args) + if isinstance(resolved, dict): + yield resolved + return + synthetic, prompt, tool_filter = resolved + + async for value in run_subagent_streaming( + parent_ctx=tool_context, + archetype=synthetic, + prompt=prompt, + agent_config=self._agent_config, + tool_filter=tool_filter, + ): + yield value + def _tool_names(tools: tuple) -> list[str]: """Extract declaration names from a tuple of tool items. diff --git a/trpc_agent_sdk/agents/sub_agent/_runner.py b/trpc_agent_sdk/agents/sub_agent/_runner.py index 79302901..d35bd215 100644 --- a/trpc_agent_sdk/agents/sub_agent/_runner.py +++ b/trpc_agent_sdk/agents/sub_agent/_runner.py @@ -19,7 +19,9 @@ from __future__ import annotations from typing import Any +from typing import AsyncIterator from typing import Optional +from typing import TypedDict from typing import Union from trpc_agent_sdk.abc import ArtifactId @@ -246,21 +248,99 @@ def _extract_final_text(last_event) -> str: return "\n".join(p.text for p in last_event.content.parts if getattr(p, "text", None)) -async def run_subagent( +class SubAgentProgress(TypedDict, total=False): + """Wire contract for a forwarded sub-agent progress event. + + This is the ``payload`` dict a consumer receives for each ``partial=True`` + progress event when ``forward_events`` is enabled. + + ``content`` is the sub-agent event's :class:`Content` dumped with + ``model_dump(exclude_none=True)`` — the same shape used everywhere else in + the framework (``parts[i].function_call.name``, ``parts[i].text`` / + ``thought``), so consumers reuse the structure they already know instead of + a bespoke schema. ``error`` and ``usage`` are lifted from the ``Event`` + itself (they do not live on ``content``). ``total=False`` because + ``content`` / ``error`` / ``usage`` are only present when the underlying + event carries them. + + Only ``content`` — never the whole ``Event`` — crosses the boundary, so + ``actions`` / ``state_delta`` (parent-context state) are not leaked. + """ + + author: Optional[str] + partial: bool + content: dict + error: dict + usage: dict + + +def _project_subagent_event(event: Any) -> SubAgentProgress: + """Project a sub-agent event into a lightweight, JSON-serializable dict. + + Forwarded to the parent runner's consumer as the ``payload`` of a progress + event when ``forward_events`` is enabled. See :class:`SubAgentProgress` for + the shape. Uses the framework-native ``Content`` dump for the event body + rather than a custom projection, and deliberately dumps only ``content`` + (not the whole ``Event``) so parent-context state never crosses the + isolation boundary. + """ + payload: SubAgentProgress = { + "author": getattr(event, "author", None), + "partial": bool(getattr(event, "partial", False)), + } + + # Framework-native content shape (parts / function_call / text / thought). + # Dump only content — actions / state_delta live on the Event and must not + # cross the isolation boundary. + content = getattr(event, "content", None) + dump = getattr(content, "model_dump", None) if content is not None else None + if callable(dump): + payload["content"] = dump(exclude_none=True) + + # Surface sub-agent run errors so the consumer can render them instead of + # showing an empty event. Added conditionally to keep the payload clean on + # the common (no-error) path. See Event.is_error(): error_code drives it. + error_code = getattr(event, "error_code", None) + error_message = getattr(event, "error_message", None) + if error_code is not None or error_message is not None: + payload["error"] = {"code": error_code, "message": error_message} + + # Token usage for cost observability. Only the headline counts are lifted + # out of the usage_metadata object (which is not JSON-serializable as-is). + usage = getattr(event, "usage_metadata", None) + if usage is not None: + payload["usage"] = { + "prompt_tokens": getattr(usage, "prompt_token_count", None), + "completion_tokens": getattr(usage, "candidates_token_count", None), + "total_tokens": getattr(usage, "total_token_count", None), + } + + return payload + + +async def run_subagent_streaming( *, parent_ctx: InvocationContext, archetype: SubAgentArchetype, prompt: str, agent_config=None, tool_filter: Optional[list] = None, -) -> Union[str, dict]: - """Run an isolated sub-agent and return its final assistant text. - - Returns: - Final assistant text on success, ``"[sub-agent cancelled]"`` if the - run was cancelled, or ``{"status": "error", "message": ...}`` on - unexpected exceptions. Errors are not raised back to the parent so - the orchestrator can decide how to react. +) -> AsyncIterator[Union[str, dict]]: + """Run an isolated sub-agent, yielding progress projections then the result. + + Yields one projection ``dict`` (see :func:`_project_subagent_event`) per + sub-agent event, and finally the sub-agent's final result as the **last** + yielded value. The final value is the assistant text on success, + ``"[sub-agent cancelled]"`` on cancellation, or + ``{"status": "error", "message": ...}`` on unexpected exceptions — errors are + surfaced as the final value rather than raised, matching + :func:`run_subagent`'s graceful-degradation contract. + + Contract with the progress-streaming tool path: every yielded value except + the last becomes a ``partial=True`` progress event surfaced to the parent + runner's consumer; the last value becomes the tool's ``function_response`` + fed back to the parent LLM. The projection dicts are never persisted into + the parent session nor seen by the parent LLM. """ # Imported lazily to mirror AgentTool and avoid a circular import at module load. from trpc_agent_sdk.runners import Runner @@ -269,7 +349,8 @@ async def run_subagent( sub_agent = _build_sub_agent(archetype, parent_ctx, agent_config=agent_config, tool_filter=tool_filter) except Exception as ex: # noqa: BLE001 logger.error("sub-agent build failed: %s", ex, exc_info=True) - return {"status": "error", "message": str(ex)} + yield {"status": "error", "message": str(ex)} + return parent_app_name = getattr(parent_ctx.session, "app_name", "trpc_app") sub_app_name = f"{parent_app_name}{SUBAGENT_APP_NAME_SUFFIX}{archetype.name}" @@ -285,6 +366,7 @@ async def run_subagent( last_event = None max_turns_reached = False + final_value: Union[str, dict, None] = None try: sub_session = await sub_runner.session_service.create_session( app_name=sub_app_name, @@ -308,6 +390,10 @@ async def run_subagent( new_message=content, ): last_event = event + # Forward this sub-agent event to the parent consumer as a progress + # projection. This is the only divergence from run_subagent's silent + # drain; the projection never reaches the parent LLM. + yield _project_subagent_event(event) # Count LLM calls (one non-partial event per request, including # those with tool calls). Aligns with claw-code-agent. if event.content and not event.partial and not event.is_error(): @@ -322,21 +408,55 @@ async def run_subagent( await _forward_artifacts(sub_runner, sub_session, parent_ctx) except RunCancelledException: - return "[sub-agent cancelled]" + final_value = "[sub-agent cancelled]" except Exception as ex: # noqa: BLE001 logger.error("sub-agent run failed: %s", ex, exc_info=True) - return {"status": "error", "message": str(ex)} + final_value = {"status": "error", "message": str(ex)} finally: try: await sub_runner.close() except Exception as close_ex: # noqa: BLE001 logger.warning("sub-agent runner close failed: %s", close_ex) - result = _extract_final_text(last_event) - if max_turns_reached: - note = "[sub-agent stopped: max turns reached]" - return f"{result}\n\n{note}" if result else note - return result + if final_value is None: + result = _extract_final_text(last_event) + if max_turns_reached: + note = "[sub-agent stopped: max turns reached]" + final_value = f"{result}\n\n{note}" if result else note + else: + final_value = result + yield final_value + + +async def run_subagent( + *, + parent_ctx: InvocationContext, + archetype: SubAgentArchetype, + prompt: str, + agent_config=None, + tool_filter: Optional[list] = None, +) -> Union[str, dict]: + """Run an isolated sub-agent and return its final assistant text. + Non-streaming wrapper around :func:`run_subagent_streaming`: it drains the + progress projections and returns only the final value. -__all__ = ["run_subagent"] + Returns: + Final assistant text on success, ``"[sub-agent cancelled]"`` if the + run was cancelled, or ``{"status": "error", "message": ...}`` on + unexpected exceptions. Errors are not raised back to the parent so + the orchestrator can decide how to react. + """ + final: Union[str, dict] = "" + async for value in run_subagent_streaming( + parent_ctx=parent_ctx, + archetype=archetype, + prompt=prompt, + agent_config=agent_config, + tool_filter=tool_filter, + ): + final = value + return final + + +__all__ = ["run_subagent", "run_subagent_streaming", "SubAgentProgress"] diff --git a/trpc_agent_sdk/agents/sub_agent/_spawn_sub_agent_tool.py b/trpc_agent_sdk/agents/sub_agent/_spawn_sub_agent_tool.py index b49f2e3c..ee54088b 100644 --- a/trpc_agent_sdk/agents/sub_agent/_spawn_sub_agent_tool.py +++ b/trpc_agent_sdk/agents/sub_agent/_spawn_sub_agent_tool.py @@ -10,8 +10,10 @@ import os from typing import Any +from typing import AsyncIterator from typing import List from typing import Optional +from typing import Tuple from typing import Union from typing_extensions import override @@ -29,6 +31,7 @@ from ._loader import load_archetypes_from_dir from ._registry import SubAgentRegistry from ._runner import run_subagent +from ._runner import run_subagent_streaming from ._sub_agent_config import SubAgentConfig @@ -106,6 +109,11 @@ def __init__( self._registry = registry self._skip_summarization = skip_summarization self._agent_config = agent_config + # When the config asks to forward the sub-agent's events, this tool + # behaves as a progress-streaming tool. Resolved once at construction + # because is_progress_streaming is read before execution to route the + # call onto the streaming path. + self._forward_events = bool(agent_config is not None and agent_config.forward_events) rendered = render_tool_description(registry) super().__init__(name="spawn_subagent", description=rendered, filters_name=filters_name, filters=filters) @@ -113,6 +121,27 @@ def __init__( def registry(self) -> SubAgentRegistry: return self._registry + @property + @override + def is_progress_streaming(self) -> bool: + """Route through the progress-streaming path when event forwarding is on. + + Statically determined by ``SubAgentConfig.forward_events`` so + the tools processor can partition this call onto the streaming path + (which drives :meth:`run_streaming`) before execution begins. + """ + return self._forward_events + + @property + def skip_summarization(self) -> bool: + """Whether the streamed sub-agent result is the parent's final answer. + + Read by the progress-streaming execution path (which bypasses + ``_run_async_impl``) to set ``skip_summarization`` on the final + ``function_response`` event. + """ + return self._skip_summarization + @override def _get_declaration(self) -> FunctionDeclaration: return FunctionDeclaration( @@ -169,16 +198,13 @@ async def process_request( "Put everything it needs in `prompt`.") llm_request.append_instructions([instruction]) - @override - async def _run_async_impl( - self, - *, - tool_context: InvocationContext, - args: dict[str, Any], - ) -> Any: - if self._skip_summarization: - tool_context.event_actions.skip_summarization = True + def _resolve_call(self, args: dict[str, Any]) -> Union[Tuple[SubAgentArchetype, str], dict]: + """Parse and validate call args into ``(archetype, prompt)``. + Returns an error dict on an unknown ``subagent_type`` (with no + ``default`` fallback) or a missing/empty ``prompt`` so both the + streaming and non-streaming paths surface the same failure shape. + """ subagent_type = args.get("subagent_type") prompt = args.get("prompt") @@ -196,7 +222,23 @@ async def _run_async_impl( if not isinstance(prompt, str) or not prompt.strip(): return {"status": "error", "message": "prompt must be a non-empty string"} - archetype = self._registry.get(resolved_type) + return self._registry.get(resolved_type), prompt + + @override + async def _run_async_impl( + self, + *, + tool_context: InvocationContext, + args: dict[str, Any], + ) -> Any: + if self._skip_summarization: + tool_context.event_actions.skip_summarization = True + + resolved = self._resolve_call(args) + if isinstance(resolved, dict): + return resolved + archetype, prompt = resolved + return await run_subagent( parent_ctx=tool_context, archetype=archetype, @@ -204,5 +246,32 @@ async def _run_async_impl( agent_config=self._agent_config, ) + async def run_streaming( + self, + *, + tool_context: InvocationContext, + args: dict[str, Any], + ) -> AsyncIterator[Any]: + """Progress-streaming entrypoint used when event forwarding is enabled. + + Yields one progress projection per sub-agent event and, as the final + value, the sub-agent's result (which the tools processor turns into the + ``function_response`` fed back to the parent LLM). See + :func:`run_subagent_streaming` for the per-value contract. + """ + resolved = self._resolve_call(args) + if isinstance(resolved, dict): + yield resolved + return + archetype, prompt = resolved + + async for value in run_subagent_streaming( + parent_ctx=tool_context, + archetype=archetype, + prompt=prompt, + agent_config=self._agent_config, + ): + yield value + __all__ = ["SpawnSubAgentTool"] diff --git a/trpc_agent_sdk/agents/sub_agent/_sub_agent_config.py b/trpc_agent_sdk/agents/sub_agent/_sub_agent_config.py index 075c9871..1d7af099 100644 --- a/trpc_agent_sdk/agents/sub_agent/_sub_agent_config.py +++ b/trpc_agent_sdk/agents/sub_agent/_sub_agent_config.py @@ -42,5 +42,11 @@ class SubAgentConfig: """Max LLM calls the sub-agent may make. ``None`` = unlimited. Each LLM request counts as one turn, including those with tool calls.""" + forward_events: bool = False + """Stream sub-agent events to the parent consumer as progress updates. + + ``True``: orchestrator can display sub-agent execution live. + ``False`` (default): sub-agent runs silently.""" + __all__ = ["SubAgentConfig"]