feat: 增强版的SubAgent功能 - #7108
Closed
elecvoid243 wants to merge 83 commits into
Closed
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces an enhanced dynamic subagent management system, allowing the main agent to create, manage, and communicate with specialized subagents. Key additions include a DynamicSubAgentManager for lifecycle and shared context handling, a dedicated SubAgentLogger, and integration into the main agent's toolset and prompt construction. Feedback focuses on improving error handling by avoiding broad exception silences, ensuring safe string formatting for system prompts, and refining the logic for detecting subagent creation failures. A minor typo in the subagent capability prompt was also identified.
elecvoid243
marked this pull request as draft
May 4, 2026 07:14
elecvoid243
marked this pull request as ready for review
May 4, 2026 11:58
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 6 issues, and left some high level feedback:
- In
RemoveSubagentTool.call, the checkif remove_status == "__SUBAGENT_REMOVED__":will never be true becauseSubAgentManager.remove_subagentreturns a string with that token plus a message suffix; consider switching to a prefix check (e.g.,startswith) or returning a structured result instead of a raw string. - In
_build_handoff_toolsetyou importSEND_SHARED_CONTEXT_TOOLfromastrbot.core.subagent_manager, but that constant is defined insubagent_tools.py; this import path will raise at runtime and should be updated to import from the correct module. - Several subagent-management paths assume session and status entries exist (e.g.,
remove_subagentindexingsession.subagent_status[agent_name], andWaitForSubagentToolusingresult.errorwhenresultmay beNone), which can raise KeyError/AttributeError; add defensive checks for missing session/agent/status/result before accessing these fields.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `RemoveSubagentTool.call`, the check `if remove_status == "__SUBAGENT_REMOVED__":` will never be true because `SubAgentManager.remove_subagent` returns a string with that token plus a message suffix; consider switching to a prefix check (e.g., `startswith`) or returning a structured result instead of a raw string.
- In `_build_handoff_toolset` you import `SEND_SHARED_CONTEXT_TOOL` from `astrbot.core.subagent_manager`, but that constant is defined in `subagent_tools.py`; this import path will raise at runtime and should be updated to import from the correct module.
- Several subagent-management paths assume session and status entries exist (e.g., `remove_subagent` indexing `session.subagent_status[agent_name]`, and `WaitForSubagentTool` using `result.error` when `result` may be `None`), which can raise KeyError/AttributeError; add defensive checks for missing session/agent/status/result before accessing these fields.
## Individual Comments
### Comment 1
<location path="astrbot/core/subagent_manager.py" line_range="962-931" />
<code_context>
+ return task_id
+
+ @classmethod
+ def _ensure_task_store(
+ cls, session: SubAgentSession, agent_name: str
+ ) -> dict[str, SubAgentExecutionResult]:
+ if agent_name not in session.subagent_background_results:
+ session.subagent_background_results[agent_name] = {}
+ return session.subagent_background_results[agent_name]
</code_context>
<issue_to_address>
**issue (bug_risk):** _ensure_task_store is called before checking for a session, which will crash when session is None.
`_ensure_task_store` assumes `session` is a valid `SubAgentSession`, but some callers do `session = cls.get_session(session_id)` and call `_ensure_task_store(session, ...)` before checking `session`:
- `has_subagent_result`
- `clear_subagent_result`
If no session exists, these will raise instead of returning `False`/no-op. Please either move the `if not session` guard before the `_ensure_task_store` call, or make `_ensure_task_store` safely handle `session is None` (e.g., by returning an empty dict).
</issue_to_address>
### Comment 2
<location path="astrbot/core/subagent_tools.py" line_range="130-139" />
<code_context>
+ }
+ )
+
+ async def call(self, context, **kwargs) -> str:
+ name = kwargs.get("name", "")
+ if not name:
+ return "Error: name required"
+ session_id = context.context.event.unified_msg_origin
+ remove_status = SubAgentManager.remove_subagent(session_id, name)
+ if remove_status == "__SUBAGENT_REMOVED__":
+ return f"Cleaned {name} Subagent"
+ else:
</code_context>
<issue_to_address>
**issue (bug_risk):** RemoveSubagentTool checks for an exact marker string that is never returned, so the success branch is dead code.
`SubAgentManager.remove_subagent` returns messages that *start with* `"__SUBAGENT_REMOVED__"` but include extra text (e.g. `"__SUBAGENT_REMOVED__: Subagent foo has been removed."`). Because of this, the equality check never passes and the friendly `"Cleaned ..."` response is never used. Consider checking `remove_status.startswith("__SUBAGENT_REMOVED__")` or having `remove_subagent` return a structured result with a status field so you can branch on that instead.
</issue_to_address>
### Comment 3
<location path="astrbot/core/subagent_tools.py" line_range="546-555" />
<code_context>
+ elif status == "FAILED":
</code_context>
<issue_to_address>
**issue (bug_risk):** wait_for_subagent may access result.error when result is None, causing a crash instead of a clean error.
In the `FAILED` branch of `WaitForSubagentTool.call`, we have:
```python
result = SubAgentManager.get_subagent_result(...)
if result and (result.result != "" or result.completed_at > 0):
return ...
else:
return f"... Error: {result.error or 'Unknown error'}"
```
If `get_subagent_result` returns `None`, the `if` condition fails and we still access `result.error`, which will raise. Please handle the `None` case explicitly, e.g.:
```python
if not result:
return f"SubAgent '{subagent_name}' failed task {task_id} with no stored result."
...
```
or split the branch into `if result is None` / `else` so we never access attributes on `None`.
</issue_to_address>
### Comment 4
<location path="astrbot/core/astr_agent_tool_exec.py" line_range="371" />
<code_context>
- tool_call_timeout=run_context.tool_call_timeout,
- stream=stream,
+
+ # 获取子代理的历史上下文
+ subagent_history, agent_name = cls._load_subagent_history(umo, tool)
+ # 如果有历史上下文，合并到 contexts 中
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the subagent prep/timeout logic and background-handling modes into small helpers to flatten control flow and make the handoff code paths clearer.
You can keep all functionality but significantly flatten the control flow with a couple of focused extractions.
### 1. Flatten `_execute_handoff` by extracting “prep” + “run with timeout”
Right now `_execute_handoff` builds history, system prompt, timeout, the `runner_messages` accumulator, and then defines `_run_subagent` that closes over a lot of locals and embeds timeout handling.
You can keep behavior while separating:
* a “pure” preparation step (`system_prompt`, `contexts`, `timeout`), and
* a “run + timeout + history persistence” helper.
This keeps `_execute_handoff` readable and makes the subagent runner reusable.
**Example refactor:**
```python
@classmethod
def _prepare_subagent_run(
cls,
*,
umo: str,
tool: HandoffTool,
prov_settings: dict,
contexts: list[Message] | None,
) -> tuple[str, list[Message] | None, float, str]:
subagent_history, agent_name = cls._load_subagent_history(umo, tool)
if subagent_history:
contexts = subagent_history + (contexts or [])
subagent_system_prompt = cls._build_subagent_system_prompt(umo, tool, prov_settings)
execution_timeout = cls._get_subagent_execution_timeout()
return subagent_system_prompt, contexts, execution_timeout, agent_name
```
```python
@classmethod
async def _run_subagent_with_timeout(
cls,
*,
run_coro: T.Callable[[], T.Awaitable],
umo: str,
agent_name: str,
execution_timeout: float,
runner_messages: list[Message],
) -> T.Any:
if execution_timeout <= 0:
return await run_coro()
try:
return await asyncio.wait_for(run_coro(), timeout=execution_timeout)
except asyncio.TimeoutError:
cls._save_subagent_history(umo, runner_messages, agent_name)
error_msg = (
f"SubAgent '{agent_name}' execution timeout after "
f"{execution_timeout:.1f} seconds."
)
logger.warning(f"[SubAgent:Timeout] {error_msg}")
cls._handle_subagent_timeout(umo=umo, agent_name=agent_name)
raise # let caller decide how to build the CallToolResult
```
Then `_execute_handoff` shrinks to:
```python
@classmethod
async def _execute_handoff(...):
# ... existing setup until prov_settings / contexts ...
subagent_system_prompt, contexts, execution_timeout, agent_name = (
cls._prepare_subagent_run(
umo=umo,
tool=tool,
prov_settings=prov_settings,
contexts=contexts,
)
)
runner_messages: list[Message] = []
async def _run_subagent():
return await ctx.tool_loop_agent(
event=event,
chat_provider_id=prov_id,
prompt=input_,
image_urls=image_urls,
system_prompt=subagent_system_prompt,
tools=toolset,
contexts=contexts,
max_steps=agent_max_step,
tool_call_timeout=run_context.tool_call_timeout,
stream=stream,
runner_messages=runner_messages,
)
try:
llm_resp = await cls._run_subagent_with_timeout(
run_coro=_run_subagent,
umo=umo,
agent_name=agent_name,
execution_timeout=execution_timeout,
runner_messages=runner_messages,
)
except asyncio.TimeoutError:
yield mcp.types.CallToolResult(
content=[
mcp.types.TextContent(
type="text",
text=(
f"error: SubAgent '{agent_name}' execution timeout after "
f"{execution_timeout:.1f} seconds."
),
)
]
)
return
cls._save_subagent_history(umo, runner_messages, agent_name)
yield mcp.types.CallToolResult(
content=[mcp.types.TextContent(type="text", text=llm_resp.completion_text)]
)
```
This removes the “inner coroutine with many captured variables” from the main method and cleanly isolates timeout behavior.
### 2. Split `_do_handoff_background` into explicit managed vs legacy paths
Currently `_do_handoff_background`:
* measures execution,
* collects tool args,
* runs handoff with timeout,
* then decides managed vs legacy behavior via `_is_managed_subagent`,
* then dispatches to either `_handle_subagent_background_result` or `_wake_main_agent_for_background_result`.
The “mode switch” is buried in the middle of the function, which increases cognitive load.
You can keep all behavior but make `_do_handoff_background` only responsible for:
* collecting common inputs,
* deciding “managed vs legacy” once,
* delegating to two small helpers.
**Example refactor (structure only):**
```python
@classmethod
async def _do_handoff_background(
cls,
tool: HandoffTool,
run_context: ContextWrapper[AstrAgentContext],
task_id: str,
**tool_args,
) -> None:
start_time = time.time()
tool_args = dict(tool_args)
tool_args["image_urls"] = await cls._collect_handoff_image_urls(
run_context,
tool_args.get("image_urls"),
)
event = run_context.context.event
umo = event.unified_msg_origin
agent_name = getattr(tool.agent, "name", None)
execution_timeout = cls._get_subagent_execution_timeout()
result_text = ""
error_text: str | None = None
async def _run():
nonlocal result_text
async for r in cls._execute_handoff(
tool,
run_context,
image_urls_prepared=True,
**tool_args,
):
if isinstance(r, mcp.types.CallToolResult):
for content in r.content:
if isinstance(content, mcp.types.TextContent):
result_text += content.text + "\n"
try:
if execution_timeout > 0:
await asyncio.wait_for(_run(), timeout=execution_timeout)
else:
await _run()
except asyncio.TimeoutError:
error_text = f"Execution timeout after {execution_timeout:.1f} seconds."
result_text = (
f"error: Background SubAgent '{agent_name}' {error_text}"
)
logger.warning(f"[SubAgent:BackgroundTask] {error_text}")
except Exception as e:
error_text = str(e)
result_text = (
f"error: Background task execution failed, internal error: {e!s}"
)
execution_time = time.time() - start_time
if cls._is_managed_subagent(umo, agent_name):
await cls._do_handoff_background_managed(
umo=umo,
agent_name=agent_name,
task_id=tool_args.get("subagent_task_id"),
result_text=result_text,
error_text=error_text,
execution_time=execution_time,
run_context=run_context,
tool=tool,
tool_args=tool_args,
)
else:
await cls._do_handoff_background_legacy(
run_context=run_context,
tool=tool,
task_id=task_id,
agent_name=agent_name,
result_text=result_text,
tool_args=tool_args,
)
```
New helpers stay tiny and linear:
```python
@classmethod
async def _do_handoff_background_managed(
cls,
*,
umo: str,
agent_name: str,
task_id: str | None,
result_text: str,
error_text: str | None,
execution_time: float,
run_context: ContextWrapper[AstrAgentContext],
tool: HandoffTool,
tool_args: dict,
) -> None:
await cls._handle_subagent_background_result(
umo=umo,
agent_name=agent_name,
task_id=task_id,
result_text=result_text,
error_text=error_text,
execution_time=execution_time,
run_context=run_context,
tool=tool,
tool_args=tool_args,
)
```
```python
@classmethod
async def _do_handoff_background_legacy(
cls,
*,
run_context: ContextWrapper[AstrAgentContext],
tool: HandoffTool,
task_id: str,
agent_name: str | None,
result_text: str,
tool_args: dict,
) -> None:
event = run_context.context.event
await cls._wake_main_agent_for_background_result(
run_context=run_context,
task_id=task_id,
tool_name=tool.name,
result_text=result_text,
tool_args=tool_args,
note=(
event.get_extra("background_note")
or f"Background task for subagent '{agent_name}' finished."
),
summary_name=f"Dedicated to subagent `{agent_name}`",
extra_result_fields={"subagent_name": agent_name},
)
```
This makes it obvious there are two paths, and `_do_handoff_background` becomes a narrow coordinator instead of a deep branching function.
### 3. Collapse thin `SubAgentManager` wrappers
Some class-level helpers are essentially pass-throughs to `SubAgentManager`:
* `_get_subagent_execution_timeout`
* `_is_managed_subagent`
* `_handle_subagent_timeout`
They increase surface area without adding much logic. Even if you keep them, you can group them into a tiny “adapter” to make their role clear and avoid scattering subagent lifecycle concerns across many methods.
**Example:**
```python
class _SubAgentRuntime:
@staticmethod
def get_timeout() -> float:
try:
return SubAgentManager.get_execution_timeout()
except Exception:
return -1
@staticmethod
def is_managed(umo: str, agent_name: str | None) -> bool:
if not agent_name:
return False
session = SubAgentManager.get_session(umo)
return bool(session and agent_name in session.subagents)
@staticmethod
def mark_timeout(umo: str, agent_name: str) -> None:
SubAgentManager.set_subagent_status(
session_id=umo,
agent_name=agent_name,
status="FAILED",
)
```
Then the class helpers become thin aliases or disappear:
```python
@staticmethod
def _get_subagent_execution_timeout() -> float:
return _SubAgentRuntime.get_timeout()
@staticmethod
def _is_managed_subagent(umo: str, agent_name: str | None) -> bool:
return _SubAgentRuntime.is_managed(umo, agent_name)
@staticmethod
def _handle_subagent_timeout(umo: str, agent_name: str) -> None:
_SubAgentRuntime.mark_timeout(umo, agent_name)
```
This at least centralizes the “protocol” with `SubAgentManager` into one small concept instead of many scattered helpers, making it easier to reason about and evolve.
</issue_to_address>
### Comment 5
<location path="astrbot/core/agent/runners/tool_loop_agent_runner.py" line_range="1018" />
<code_context>
if not req.func_tool:
return
</code_context>
<issue_to_address>
**issue (complexity):** Consider encapsulating tool resolution, dynamic tool protocol handling, and abort reasons into focused helpers/enums so the runner’s main control flow stays linear and simpler.
You can reduce the added complexity without changing behavior by isolating the cross‑cutting concerns (tool resolution/protocol and abort reasons) behind small helpers/objects, so the runner stays linear.
### 1. Centralize tool resolution into a single helper
Right now the tool resolution path is:
- inline dynamic resolution (`_resolve_dynamic_subagent_tool`)
- then schema-mode (`skills_like` vs normal)
- plus the old `req.func_tool.get_tool(...)`
This is duplicated/overlapping and makes the main execution branch harder to follow. You can hide this behind a single `_get_func_tool(...)` that encodes the resolution strategy in one place.
**Before (simplified):**
```python
if not req.func_tool:
return
# Prefer dynamic tools when available
func_tool = self._resolve_dynamic_subagent_tool(func_tool_name)
# If not found in dynamic tools, check regular tool sets
if func_tool is None:
if (
self.tool_schema_mode == "skills_like"
and self._skill_like_raw_tool_set
):
func_tool = self._skill_like_raw_tool_set.get_tool(func_tool_name)
else:
func_tool = req.func_tool.get_tool(func_tool_name)
if (
self.tool_schema_mode == "skills_like"
and self._skill_like_raw_tool_set
):
func_tool = self._skill_like_raw_tool_set.get_tool(func_tool_name)
available_tools = self._skill_like_raw_tool_set.names()
else:
func_tool = req.func_tool.get_tool(func_tool_name)
available_tools = req.func_tool.names()
```
**After (main path stays simple):**
```python
if not req.func_tool:
return
func_tool, available_tools = self._get_func_tool(
func_tool_name=func_tool_name,
req=req,
)
```
**Helper implementation (localized complexity):**
```python
def _get_func_tool(
self,
func_tool_name: str,
req: ProviderRequest,
) -> tuple[FunctionTool, list[str]]:
# 1) dynamic subagent / handoff tools first
dynamic_tool = self._resolve_dynamic_subagent_tool(func_tool_name)
if dynamic_tool is not None:
# In case dynamic tools need their own "names" set:
available = (
self._skill_like_raw_tool_set.names()
if (self.tool_schema_mode == "skills_like"
and self._skill_like_raw_tool_set)
else req.func_tool.names()
)
return dynamic_tool, available
# 2) fall back to existing skills_like vs full logic
if self.tool_schema_mode == "skills_like" and self._skill_like_raw_tool_set:
func_tool = self._skill_like_raw_tool_set.get_tool(func_tool_name)
available_tools = self._skill_like_raw_tool_set.names()
else:
func_tool = req.func_tool.get_tool(func_tool_name)
available_tools = req.func_tool.names()
return func_tool, available_tools
```
This keeps the behavior the same but removes branching from the core loop and makes the lookup strategy explicit and testable.
### 2. Isolate the “dynamic tool protocol” into a separate helper
`_maybe_register_dynamic_tool_from_result` is currently:
- parsing a magic prefix
- doing string splitting and validation
- reaching into `run_context.context.event.unified_msg_origin`
- mutating `self.req.func_tool`
You can make this easier to reason about and reuse by pushing protocol parsing into a separate static/helper, so the runner just orchestrates.
**Before (runner does everything):**
```python
def _maybe_register_dynamic_tool_from_result(self, result_content: str) -> None:
if not result_content.startswith("__DYNAMIC_TOOL_CREATED__:"):
return
parts = result_content.split(":", 3)
if len(parts) < 4:
return
new_tool_name = parts[1]
new_tool_obj_name = parts[2]
...
session_id = getattr(event, "unified_msg_origin", None) if event else None
...
handoffs = SubAgentManager.get_handoff_tools_for_session(session_id)
for handoff in handoffs:
if (
handoff.name == new_tool_obj_name
or handoff.name == new_tool_name.replace("transfer_to_", "")
):
if self.req.func_tool:
self.req.func_tool.add_tool(handoff)
...
break
```
**After (runner calls into a dedicated protocol adapter):**
```python
def _maybe_register_dynamic_tool_from_result(self, result_content: str) -> None:
run_context_context = getattr(self.run_context, "context", None)
event = getattr(run_context_context, "event", None) if run_context_context else None
session_id = getattr(event, "unified_msg_origin", None) if event else None
if not session_id or not self.req.func_tool:
return
tool = DynamicToolProtocol.parse_and_resolve(
result_content=result_content,
session_id=session_id,
)
if tool:
self.req.func_tool.add_tool(tool)
```
**Protocol helper (can live in `SubAgentManager` or a small module):**
```python
class DynamicToolProtocol:
MARKER = "__DYNAMIC_TOOL_CREATED__:"
@classmethod
def parse_and_resolve(cls, result_content: str, session_id: str):
if not result_content.startswith(cls.MARKER):
return None
parts = result_content.split(":", 3)
if len(parts) < 4:
return None
new_tool_name = parts[1]
new_tool_obj_name = parts[2]
handoffs = SubAgentManager.get_handoff_tools_for_session(session_id)
for handoff in handoffs:
if (
handoff.name == new_tool_obj_name
or handoff.name == new_tool_name.replace("transfer_to_", "")
):
logger.info(f"[SubAgent] Tool created: {new_tool_name}")
logger.info(f"[SubAgent] Resolved handoff: {handoff.name}")
return handoff
return None
```
This keeps the protocol semantics exactly as-is, but the runner becomes a thin client of `DynamicToolProtocol`, and you can unit test that class independently of the runner.
### 3. Make abort reasons explicit instead of a boolean flag
`_finalize_aborted_step` currently uses a `manual_stop: bool` to distinguish “user stop” vs “main agent stopped SubAgent”. That’s a hidden “reason enum” encoded as a boolean, and every caller has to remember which value to pass.
You can make this more self‑documenting by introducing a tiny `Enum` and a local helper, without changing logging/messages.
**Before:**
```python
async def _finalize_aborted_step(
self,
llm_resp: LLMResponse | None = None,
manual_stop: bool = False,
) -> AgentResponse:
if manual_stop:
logger.info("SubAgent execution was manually stopped by main agent.")
else:
logger.info("Agent execution was requested to stop by user.")
if llm_resp is None:
llm_resp = LLMResponse(role="assistant", completion_text="")
if llm_resp.role != "assistant":
if manual_stop:
interruption_msg = (
"[SYSTEM: SubAgent was manually stopped by main agent. "
"Partial output before interruption is preserved.]"
)
else:
interruption_msg = self.USER_INTERRUPTION_MESSAGE
llm_resp = LLMResponse(
role="assistant",
completion_text=interruption_msg,
)
...
```
**After (more explicit, same behavior):**
```python
from enum import Enum, auto
class AbortReason(Enum):
USER_REQUEST = auto()
MAIN_AGENT_MANUAL_STOP = auto()
```
```python
def _build_interruption_message(self, reason: AbortReason) -> str:
if reason is AbortReason.MAIN_AGENT_MANUAL_STOP:
return (
"[SYSTEM: SubAgent was manually stopped by main agent. "
"Partial output before interruption is preserved.]"
)
return self.USER_INTERRUPTION_MESSAGE
```
```python
async def _finalize_aborted_step(
self,
llm_resp: LLMResponse | None = None,
reason: AbortReason = AbortReason.USER_REQUEST,
) -> AgentResponse:
if reason is AbortReason.MAIN_AGENT_MANUAL_STOP:
logger.info("SubAgent execution was manually stopped by main agent.")
else:
logger.info("Agent execution was requested to stop by user.")
if llm_resp is None:
llm_resp = LLMResponse(role="assistant", completion_text="")
if llm_resp.role != "assistant":
llm_resp = LLMResponse(
role="assistant",
completion_text=self._build_interruption_message(reason),
)
...
```
Existing callers that passed `manual_stop=True/False` become more self‑describing:
```python
await self._finalize_aborted_step(
llm_resp=resp,
reason=AbortReason.MAIN_AGENT_MANUAL_STOP,
)
```
This keeps all functionality intact but removes the “boolean protocol” and makes stop logic easier to reason about and test.
</issue_to_address>
### Comment 6
<location path="astrbot/core/subagent_tools.py" line_range="25" />
<code_context>
+
+
+@dataclass
+class CreateSubAgentTool(FunctionTool):
+ name: str = "create_subagent"
+ description: str = "Create a subagent. After creation, use transfer_to_{name} tool."
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared helpers and splitting this large module by concern so each subagent tool focuses only on orchestration rather than low-level logic and protocol details.
You can reduce the new complexity without changing behavior by extracting a few helpers and centralizing the protocol handling. Here are focused refactors that keep the tools thin and coordinated.
---
### 1. Move `_check_path_safety` out of `CreateSubAgentTool`
Right now the tool mixes validation, security, and orchestration. You can move path-safety logic into a shared helper (or `SubAgentManager` / `astrbot_path`), so the tool stays as a coordinator:
```python
# subagent_utils.py (or subagent_manager.py)
def is_safe_workdir(path_str: str) -> bool:
if not path_str or not isinstance(path_str, str):
return False
if not os.path.isabs(path_str):
return False
try:
resolved = os.path.realpath(path_str)
except (OSError, ValueError):
return False
# ... existing OS-specific checks here ...
# (copy the existing logic, unchanged)
return True
```
Then in the tool:
```python
from .subagent_utils import is_safe_workdir
async def call(self, context, **kwargs) -> str:
...
workdir = kwargs.get("workdir")
if workdir is None or not is_safe_workdir(workdir):
workdir = get_astrbot_temp_path()
...
```
This keeps the tool focused on wiring config and leaves security policy in a reusable, testable function.
---
### 2. Centralize magic protocol strings
Multiple tools depend on string constants like `__DYNAMIC_TOOL_CREATED__`, `__SUBAGENT_REMOVED__`, `__HISTORY_CLEARED__`, `__SHARED_CONTEXT_ADDED__`. To reduce coupling and scattered string comparisons, introduce a small result type and a common mapper.
Example: define structured results and constants:
```python
# subagent_protocol.py
from enum import Enum
from dataclasses import dataclass
from typing import Optional
class SubAgentAction(Enum):
CREATED = "created"
CREATE_FAILED = "create_failed"
REMOVED = "removed"
HISTORY_CLEARED = "history_cleared"
SHARED_CONTEXT_ADDED = "shared_context_added"
ERROR = "error"
@dataclass
class SubAgentResult:
action: SubAgentAction
message: Optional[str] = None
tool_name: Optional[str] = None
handoff_tool_name: Optional[str] = None
```
Change `SubAgentManager.create_subagent` (and others) to return `SubAgentResult` instead of protocol strings (or in addition, if you need backward compatibility). Then keep protocol string formatting in one helper:
```python
# subagent_protocol.py
def encode_dynamic_tool_created(result: SubAgentResult) -> str:
if result.action == SubAgentAction.CREATED:
return f"__DYNAMIC_TOOL_CREATED__:{result.tool_name}:{result.handoff_tool_name}:{result.message}"
if result.action == SubAgentAction.CREATE_FAILED:
return f"__DYNAMIC_TOOL_CREATE_FAILED__:{result.tool_name}"
return result.message or "Unknown result"
```
Then the tool becomes:
```python
from .subagent_protocol import encode_dynamic_tool_created
async def call(self, context, **kwargs) -> str:
...
result = await SubAgentManager.create_subagent(session_id=session_id, config=config)
return encode_dynamic_tool_created(result)
```
Similarly, `RemoveSubagentTool`, `ResetSubAgentTool`, and shared-context tools can consume structured results and call shared format/encode helpers instead of locally comparing magic constants.
---
### 3. Extract the polling state machine from `WaitForSubagentTool`
`WaitForSubagentTool.call` is doing a lot (task selection + polling + formatting). You can extract the polling logic and status handling into a reusable helper, leaving the tool to argument validation and presentation.
For instance:
```python
# subagent_waiter.py
import time
import asyncio
from typing import Optional
async def wait_for_task_completion(
session_id: str,
subagent_name: str,
task_id: Optional[str],
timeout: float,
poll_interval: float,
):
start_time = time.time()
while time.time() - start_time < timeout:
session = SubAgentManager.get_session(session_id)
if not session:
return {"type": "error", "reason": "session_not_found"}
if subagent_name not in session.subagents:
return {"type": "error", "reason": "agent_removed"}
status = SubAgentManager.get_subagent_status(session_id, subagent_name)
if status in ("COMPLETED", "FAILED"):
result = SubAgentManager.get_subagent_result(session_id, subagent_name, task_id)
return {
"type": status.lower(),
"result": result,
}
if status == "IDLE":
return {"type": "idle"}
await asyncio.sleep(poll_interval)
return {"type": "timeout", "task_id": task_id}
```
Then the tool:
```python
from .subagent_waiter import wait_for_task_completion
async def call(self, context, **kwargs) -> str:
...
# keep the logic that selects/infers task_id as-is
outcome = await wait_for_task_completion(
session_id, subagent_name, task_id, timeout, poll_interval
)
if outcome["type"] == "completed":
result = outcome["result"]
return (
f"SubAgent '{result.agent_name}' execution completed\n"
f" Task id: {result.task_id}\n"
f" Execution time: {result.execution_time:.1f}s\n--- Result ---\n{result.result}\n"
)
if outcome["type"] == "failed":
result = outcome["result"]
return (
f"SubAgent '{result.agent_name}' execution failed\n"
f"Task id: {result.task_id}\n"
f"Execution time: {result.execution_time:.1f}s\n"
f"Error: {result.error or 'Unknown error'}\n"
)
if outcome["type"] == "idle":
return f"Error: SubAgent '{subagent_name}' is running no tasks."
if outcome["type"] == "timeout":
target = f"Task {outcome['task_id']}"
return (
f" Timeout! \nSubAgent '{subagent_name}' has not finished '{target}' in {timeout}s. "
"The task may be still running. You can continue waiting by `wait_for_subagent` again."
)
if outcome["type"] == "error":
if outcome["reason"] == "session_not_found":
return "Error: Session Not Found"
if outcome["reason"] == "agent_removed":
return f"Error: SubAgent '{subagent_name}' not found. It may be removed."
```
This keeps the state machine reusable (for other background tasks) and makes the tool’s control flow straightforward.
---
### 4. Split the module by concern
Given the size and mix of responsibilities, you can improve cohesion by moving tools into small, focused modules while keeping the public API unchanged:
```text
subagent_tools/
__init__.py # re-exports current constants
lifecycle.py # Create/Remove/Protect/Unprotect/Reset/List tools
shared_context.py # Send/View shared context tools
waiters.py # WaitForSubagentTool and polling helpers
utils.py # is_safe_workdir, protocol helpers
```
`__init__.py` can keep the existing names:
```python
from .lifecycle import (
CREATE_SUBAGENT_TOOL,
REMOVE_SUBAGENT_TOOL,
LIST_SUBAGENTS_TOOL,
RESET_SUBAGENT_TOOL,
PROTECT_SUBAGENT_TOOL,
UNPROTECT_SUBAGENT_TOOL,
)
from .shared_context import (
SEND_SHARED_CONTEXT_TOOL,
SEND_SHARED_CONTEXT_TOOL_FOR_MAIN_AGENT,
VIEW_SHARED_CONTEXT_TOOL,
)
from .waiters import WAIT_FOR_SUBAGENT_TOOL
```
This keeps current imports working while reducing per-file complexity and clarifying boundaries between lifecycle, shared-context, and background-task behavior.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| session = cls._get_or_create_session(session_id) | ||
|
|
||
| # 初始化 | ||
| if agent_name not in session.subagent_background_results: |
Contributor
There was a problem hiding this comment.
issue (bug_risk): _ensure_task_store is called before checking for a session, which will crash when session is None.
_ensure_task_store assumes session is a valid SubAgentSession, but some callers do session = cls.get_session(session_id) and call _ensure_task_store(session, ...) before checking session:
has_subagent_resultclear_subagent_result
If no session exists, these will raise instead of returning False/no-op. Please either move the if not session guard before the _ensure_task_store call, or make _ensure_task_store safely handle session is None (e.g., by returning an empty dict).
Comment on lines
+130
to
+139
| async def call(self, context, **kwargs) -> str: | ||
| name = kwargs.get("name", "") | ||
|
|
||
| if not name: | ||
| return "Error: subagent name required" | ||
| # 验证名称格式：只允许英文字母、数字和下划线，长度限制 | ||
| if not re.match(r"^[a-zA-Z][a-zA-Z0-9_]{0,31}$", name): | ||
| return "Error: SubAgent name must start with letter, contain only letters/numbers/underscores, max 32 characters" | ||
|
|
||
| if name.startswith("__") and name.endswith("__"): |
Contributor
There was a problem hiding this comment.
issue (bug_risk): RemoveSubagentTool checks for an exact marker string that is never returned, so the success branch is dead code.
SubAgentManager.remove_subagent returns messages that start with "__SUBAGENT_REMOVED__" but include extra text (e.g. "__SUBAGENT_REMOVED__: Subagent foo has been removed."). Because of this, the equality check never passes and the friendly "Cleaned ..." response is never used. Consider checking remove_status.startswith("__SUBAGENT_REMOVED__") or having remove_subagent return a structured result with a status field so you can branch on that instead.
Comment on lines
+546
to
+555
| elif status == "FAILED": | ||
| result = SubAgentManager.get_subagent_result( | ||
| session_id, subagent_name, task_id | ||
| ) | ||
| if result and (result.result != "" or result.completed_at > 0): | ||
| return ( | ||
| f"SubAgent '{result.agent_name}' execution failed\n" | ||
| f"Task id: {result.task_id}\n" | ||
| f"Execution time: {result.execution_time:.1f}s\n" | ||
| f"Error: {result.error or 'Unknown error'}\n" |
Contributor
There was a problem hiding this comment.
issue (bug_risk): wait_for_subagent may access result.error when result is None, causing a crash instead of a clean error.
In the FAILED branch of WaitForSubagentTool.call, we have:
result = SubAgentManager.get_subagent_result(...)
if result and (result.result != "" or result.completed_at > 0):
return ...
else:
return f"... Error: {result.error or 'Unknown error'}"If get_subagent_result returns None, the if condition fails and we still access result.error, which will raise. Please handle the None case explicitly, e.g.:
if not result:
return f"SubAgent '{subagent_name}' failed task {task_id} with no stored result."
...or split the branch into if result is None / else so we never access attributes on None.
| tool_call_timeout=run_context.tool_call_timeout, | ||
| stream=stream, | ||
|
|
||
| # 获取子代理的历史上下文 |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting the subagent prep/timeout logic and background-handling modes into small helpers to flatten control flow and make the handoff code paths clearer.
You can keep all functionality but significantly flatten the control flow with a couple of focused extractions.
1. Flatten _execute_handoff by extracting “prep” + “run with timeout”
Right now _execute_handoff builds history, system prompt, timeout, the runner_messages accumulator, and then defines _run_subagent that closes over a lot of locals and embeds timeout handling.
You can keep behavior while separating:
- a “pure” preparation step (
system_prompt,contexts,timeout), and - a “run + timeout + history persistence” helper.
This keeps _execute_handoff readable and makes the subagent runner reusable.
Example refactor:
@classmethod
def _prepare_subagent_run(
cls,
*,
umo: str,
tool: HandoffTool,
prov_settings: dict,
contexts: list[Message] | None,
) -> tuple[str, list[Message] | None, float, str]:
subagent_history, agent_name = cls._load_subagent_history(umo, tool)
if subagent_history:
contexts = subagent_history + (contexts or [])
subagent_system_prompt = cls._build_subagent_system_prompt(umo, tool, prov_settings)
execution_timeout = cls._get_subagent_execution_timeout()
return subagent_system_prompt, contexts, execution_timeout, agent_name@classmethod
async def _run_subagent_with_timeout(
cls,
*,
run_coro: T.Callable[[], T.Awaitable],
umo: str,
agent_name: str,
execution_timeout: float,
runner_messages: list[Message],
) -> T.Any:
if execution_timeout <= 0:
return await run_coro()
try:
return await asyncio.wait_for(run_coro(), timeout=execution_timeout)
except asyncio.TimeoutError:
cls._save_subagent_history(umo, runner_messages, agent_name)
error_msg = (
f"SubAgent '{agent_name}' execution timeout after "
f"{execution_timeout:.1f} seconds."
)
logger.warning(f"[SubAgent:Timeout] {error_msg}")
cls._handle_subagent_timeout(umo=umo, agent_name=agent_name)
raise # let caller decide how to build the CallToolResultThen _execute_handoff shrinks to:
@classmethod
async def _execute_handoff(...):
# ... existing setup until prov_settings / contexts ...
subagent_system_prompt, contexts, execution_timeout, agent_name = (
cls._prepare_subagent_run(
umo=umo,
tool=tool,
prov_settings=prov_settings,
contexts=contexts,
)
)
runner_messages: list[Message] = []
async def _run_subagent():
return await ctx.tool_loop_agent(
event=event,
chat_provider_id=prov_id,
prompt=input_,
image_urls=image_urls,
system_prompt=subagent_system_prompt,
tools=toolset,
contexts=contexts,
max_steps=agent_max_step,
tool_call_timeout=run_context.tool_call_timeout,
stream=stream,
runner_messages=runner_messages,
)
try:
llm_resp = await cls._run_subagent_with_timeout(
run_coro=_run_subagent,
umo=umo,
agent_name=agent_name,
execution_timeout=execution_timeout,
runner_messages=runner_messages,
)
except asyncio.TimeoutError:
yield mcp.types.CallToolResult(
content=[
mcp.types.TextContent(
type="text",
text=(
f"error: SubAgent '{agent_name}' execution timeout after "
f"{execution_timeout:.1f} seconds."
),
)
]
)
return
cls._save_subagent_history(umo, runner_messages, agent_name)
yield mcp.types.CallToolResult(
content=[mcp.types.TextContent(type="text", text=llm_resp.completion_text)]
)This removes the “inner coroutine with many captured variables” from the main method and cleanly isolates timeout behavior.
2. Split _do_handoff_background into explicit managed vs legacy paths
Currently _do_handoff_background:
- measures execution,
- collects tool args,
- runs handoff with timeout,
- then decides managed vs legacy behavior via
_is_managed_subagent, - then dispatches to either
_handle_subagent_background_resultor_wake_main_agent_for_background_result.
The “mode switch” is buried in the middle of the function, which increases cognitive load.
You can keep all behavior but make _do_handoff_background only responsible for:
- collecting common inputs,
- deciding “managed vs legacy” once,
- delegating to two small helpers.
Example refactor (structure only):
@classmethod
async def _do_handoff_background(
cls,
tool: HandoffTool,
run_context: ContextWrapper[AstrAgentContext],
task_id: str,
**tool_args,
) -> None:
start_time = time.time()
tool_args = dict(tool_args)
tool_args["image_urls"] = await cls._collect_handoff_image_urls(
run_context,
tool_args.get("image_urls"),
)
event = run_context.context.event
umo = event.unified_msg_origin
agent_name = getattr(tool.agent, "name", None)
execution_timeout = cls._get_subagent_execution_timeout()
result_text = ""
error_text: str | None = None
async def _run():
nonlocal result_text
async for r in cls._execute_handoff(
tool,
run_context,
image_urls_prepared=True,
**tool_args,
):
if isinstance(r, mcp.types.CallToolResult):
for content in r.content:
if isinstance(content, mcp.types.TextContent):
result_text += content.text + "\n"
try:
if execution_timeout > 0:
await asyncio.wait_for(_run(), timeout=execution_timeout)
else:
await _run()
except asyncio.TimeoutError:
error_text = f"Execution timeout after {execution_timeout:.1f} seconds."
result_text = (
f"error: Background SubAgent '{agent_name}' {error_text}"
)
logger.warning(f"[SubAgent:BackgroundTask] {error_text}")
except Exception as e:
error_text = str(e)
result_text = (
f"error: Background task execution failed, internal error: {e!s}"
)
execution_time = time.time() - start_time
if cls._is_managed_subagent(umo, agent_name):
await cls._do_handoff_background_managed(
umo=umo,
agent_name=agent_name,
task_id=tool_args.get("subagent_task_id"),
result_text=result_text,
error_text=error_text,
execution_time=execution_time,
run_context=run_context,
tool=tool,
tool_args=tool_args,
)
else:
await cls._do_handoff_background_legacy(
run_context=run_context,
tool=tool,
task_id=task_id,
agent_name=agent_name,
result_text=result_text,
tool_args=tool_args,
)New helpers stay tiny and linear:
@classmethod
async def _do_handoff_background_managed(
cls,
*,
umo: str,
agent_name: str,
task_id: str | None,
result_text: str,
error_text: str | None,
execution_time: float,
run_context: ContextWrapper[AstrAgentContext],
tool: HandoffTool,
tool_args: dict,
) -> None:
await cls._handle_subagent_background_result(
umo=umo,
agent_name=agent_name,
task_id=task_id,
result_text=result_text,
error_text=error_text,
execution_time=execution_time,
run_context=run_context,
tool=tool,
tool_args=tool_args,
)@classmethod
async def _do_handoff_background_legacy(
cls,
*,
run_context: ContextWrapper[AstrAgentContext],
tool: HandoffTool,
task_id: str,
agent_name: str | None,
result_text: str,
tool_args: dict,
) -> None:
event = run_context.context.event
await cls._wake_main_agent_for_background_result(
run_context=run_context,
task_id=task_id,
tool_name=tool.name,
result_text=result_text,
tool_args=tool_args,
note=(
event.get_extra("background_note")
or f"Background task for subagent '{agent_name}' finished."
),
summary_name=f"Dedicated to subagent `{agent_name}`",
extra_result_fields={"subagent_name": agent_name},
)This makes it obvious there are two paths, and _do_handoff_background becomes a narrow coordinator instead of a deep branching function.
3. Collapse thin SubAgentManager wrappers
Some class-level helpers are essentially pass-throughs to SubAgentManager:
_get_subagent_execution_timeout_is_managed_subagent_handle_subagent_timeout
They increase surface area without adding much logic. Even if you keep them, you can group them into a tiny “adapter” to make their role clear and avoid scattering subagent lifecycle concerns across many methods.
Example:
class _SubAgentRuntime:
@staticmethod
def get_timeout() -> float:
try:
return SubAgentManager.get_execution_timeout()
except Exception:
return -1
@staticmethod
def is_managed(umo: str, agent_name: str | None) -> bool:
if not agent_name:
return False
session = SubAgentManager.get_session(umo)
return bool(session and agent_name in session.subagents)
@staticmethod
def mark_timeout(umo: str, agent_name: str) -> None:
SubAgentManager.set_subagent_status(
session_id=umo,
agent_name=agent_name,
status="FAILED",
)Then the class helpers become thin aliases or disappear:
@staticmethod
def _get_subagent_execution_timeout() -> float:
return _SubAgentRuntime.get_timeout()
@staticmethod
def _is_managed_subagent(umo: str, agent_name: str | None) -> bool:
return _SubAgentRuntime.is_managed(umo, agent_name)
@staticmethod
def _handle_subagent_timeout(umo: str, agent_name: str) -> None:
_SubAgentRuntime.mark_timeout(umo, agent_name)This at least centralizes the “protocol” with SubAgentManager into one small concept instead of many scattered helpers, making it easier to reason about and evolve.
| @@ -1010,6 +1018,22 @@ def _append_tool_call_result(tool_call_id: str, content: str) -> None: | |||
| if not req.func_tool: | |||
Contributor
There was a problem hiding this comment.
issue (complexity): Consider encapsulating tool resolution, dynamic tool protocol handling, and abort reasons into focused helpers/enums so the runner’s main control flow stays linear and simpler.
You can reduce the added complexity without changing behavior by isolating the cross‑cutting concerns (tool resolution/protocol and abort reasons) behind small helpers/objects, so the runner stays linear.
1. Centralize tool resolution into a single helper
Right now the tool resolution path is:
- inline dynamic resolution (
_resolve_dynamic_subagent_tool) - then schema-mode (
skills_likevs normal) - plus the old
req.func_tool.get_tool(...)
This is duplicated/overlapping and makes the main execution branch harder to follow. You can hide this behind a single _get_func_tool(...) that encodes the resolution strategy in one place.
Before (simplified):
if not req.func_tool:
return
# Prefer dynamic tools when available
func_tool = self._resolve_dynamic_subagent_tool(func_tool_name)
# If not found in dynamic tools, check regular tool sets
if func_tool is None:
if (
self.tool_schema_mode == "skills_like"
and self._skill_like_raw_tool_set
):
func_tool = self._skill_like_raw_tool_set.get_tool(func_tool_name)
else:
func_tool = req.func_tool.get_tool(func_tool_name)
if (
self.tool_schema_mode == "skills_like"
and self._skill_like_raw_tool_set
):
func_tool = self._skill_like_raw_tool_set.get_tool(func_tool_name)
available_tools = self._skill_like_raw_tool_set.names()
else:
func_tool = req.func_tool.get_tool(func_tool_name)
available_tools = req.func_tool.names()After (main path stays simple):
if not req.func_tool:
return
func_tool, available_tools = self._get_func_tool(
func_tool_name=func_tool_name,
req=req,
)Helper implementation (localized complexity):
def _get_func_tool(
self,
func_tool_name: str,
req: ProviderRequest,
) -> tuple[FunctionTool, list[str]]:
# 1) dynamic subagent / handoff tools first
dynamic_tool = self._resolve_dynamic_subagent_tool(func_tool_name)
if dynamic_tool is not None:
# In case dynamic tools need their own "names" set:
available = (
self._skill_like_raw_tool_set.names()
if (self.tool_schema_mode == "skills_like"
and self._skill_like_raw_tool_set)
else req.func_tool.names()
)
return dynamic_tool, available
# 2) fall back to existing skills_like vs full logic
if self.tool_schema_mode == "skills_like" and self._skill_like_raw_tool_set:
func_tool = self._skill_like_raw_tool_set.get_tool(func_tool_name)
available_tools = self._skill_like_raw_tool_set.names()
else:
func_tool = req.func_tool.get_tool(func_tool_name)
available_tools = req.func_tool.names()
return func_tool, available_toolsThis keeps the behavior the same but removes branching from the core loop and makes the lookup strategy explicit and testable.
2. Isolate the “dynamic tool protocol” into a separate helper
_maybe_register_dynamic_tool_from_result is currently:
- parsing a magic prefix
- doing string splitting and validation
- reaching into
run_context.context.event.unified_msg_origin - mutating
self.req.func_tool
You can make this easier to reason about and reuse by pushing protocol parsing into a separate static/helper, so the runner just orchestrates.
Before (runner does everything):
def _maybe_register_dynamic_tool_from_result(self, result_content: str) -> None:
if not result_content.startswith("__DYNAMIC_TOOL_CREATED__:"):
return
parts = result_content.split(":", 3)
if len(parts) < 4:
return
new_tool_name = parts[1]
new_tool_obj_name = parts[2]
...
session_id = getattr(event, "unified_msg_origin", None) if event else None
...
handoffs = SubAgentManager.get_handoff_tools_for_session(session_id)
for handoff in handoffs:
if (
handoff.name == new_tool_obj_name
or handoff.name == new_tool_name.replace("transfer_to_", "")
):
if self.req.func_tool:
self.req.func_tool.add_tool(handoff)
...
breakAfter (runner calls into a dedicated protocol adapter):
def _maybe_register_dynamic_tool_from_result(self, result_content: str) -> None:
run_context_context = getattr(self.run_context, "context", None)
event = getattr(run_context_context, "event", None) if run_context_context else None
session_id = getattr(event, "unified_msg_origin", None) if event else None
if not session_id or not self.req.func_tool:
return
tool = DynamicToolProtocol.parse_and_resolve(
result_content=result_content,
session_id=session_id,
)
if tool:
self.req.func_tool.add_tool(tool)Protocol helper (can live in SubAgentManager or a small module):
class DynamicToolProtocol:
MARKER = "__DYNAMIC_TOOL_CREATED__:"
@classmethod
def parse_and_resolve(cls, result_content: str, session_id: str):
if not result_content.startswith(cls.MARKER):
return None
parts = result_content.split(":", 3)
if len(parts) < 4:
return None
new_tool_name = parts[1]
new_tool_obj_name = parts[2]
handoffs = SubAgentManager.get_handoff_tools_for_session(session_id)
for handoff in handoffs:
if (
handoff.name == new_tool_obj_name
or handoff.name == new_tool_name.replace("transfer_to_", "")
):
logger.info(f"[SubAgent] Tool created: {new_tool_name}")
logger.info(f"[SubAgent] Resolved handoff: {handoff.name}")
return handoff
return NoneThis keeps the protocol semantics exactly as-is, but the runner becomes a thin client of DynamicToolProtocol, and you can unit test that class independently of the runner.
3. Make abort reasons explicit instead of a boolean flag
_finalize_aborted_step currently uses a manual_stop: bool to distinguish “user stop” vs “main agent stopped SubAgent”. That’s a hidden “reason enum” encoded as a boolean, and every caller has to remember which value to pass.
You can make this more self‑documenting by introducing a tiny Enum and a local helper, without changing logging/messages.
Before:
async def _finalize_aborted_step(
self,
llm_resp: LLMResponse | None = None,
manual_stop: bool = False,
) -> AgentResponse:
if manual_stop:
logger.info("SubAgent execution was manually stopped by main agent.")
else:
logger.info("Agent execution was requested to stop by user.")
if llm_resp is None:
llm_resp = LLMResponse(role="assistant", completion_text="")
if llm_resp.role != "assistant":
if manual_stop:
interruption_msg = (
"[SYSTEM: SubAgent was manually stopped by main agent. "
"Partial output before interruption is preserved.]"
)
else:
interruption_msg = self.USER_INTERRUPTION_MESSAGE
llm_resp = LLMResponse(
role="assistant",
completion_text=interruption_msg,
)
...After (more explicit, same behavior):
from enum import Enum, auto
class AbortReason(Enum):
USER_REQUEST = auto()
MAIN_AGENT_MANUAL_STOP = auto()def _build_interruption_message(self, reason: AbortReason) -> str:
if reason is AbortReason.MAIN_AGENT_MANUAL_STOP:
return (
"[SYSTEM: SubAgent was manually stopped by main agent. "
"Partial output before interruption is preserved.]"
)
return self.USER_INTERRUPTION_MESSAGEasync def _finalize_aborted_step(
self,
llm_resp: LLMResponse | None = None,
reason: AbortReason = AbortReason.USER_REQUEST,
) -> AgentResponse:
if reason is AbortReason.MAIN_AGENT_MANUAL_STOP:
logger.info("SubAgent execution was manually stopped by main agent.")
else:
logger.info("Agent execution was requested to stop by user.")
if llm_resp is None:
llm_resp = LLMResponse(role="assistant", completion_text="")
if llm_resp.role != "assistant":
llm_resp = LLMResponse(
role="assistant",
completion_text=self._build_interruption_message(reason),
)
...Existing callers that passed manual_stop=True/False become more self‑describing:
await self._finalize_aborted_step(
llm_resp=resp,
reason=AbortReason.MAIN_AGENT_MANUAL_STOP,
)This keeps all functionality intact but removes the “boolean protocol” and makes stop logic easier to reason about and test.
|
|
||
|
|
||
| @dataclass | ||
| class CreateSubAgentTool(FunctionTool): |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting shared helpers and splitting this large module by concern so each subagent tool focuses only on orchestration rather than low-level logic and protocol details.
You can reduce the new complexity without changing behavior by extracting a few helpers and centralizing the protocol handling. Here are focused refactors that keep the tools thin and coordinated.
1. Move _check_path_safety out of CreateSubAgentTool
Right now the tool mixes validation, security, and orchestration. You can move path-safety logic into a shared helper (or SubAgentManager / astrbot_path), so the tool stays as a coordinator:
# subagent_utils.py (or subagent_manager.py)
def is_safe_workdir(path_str: str) -> bool:
if not path_str or not isinstance(path_str, str):
return False
if not os.path.isabs(path_str):
return False
try:
resolved = os.path.realpath(path_str)
except (OSError, ValueError):
return False
# ... existing OS-specific checks here ...
# (copy the existing logic, unchanged)
return TrueThen in the tool:
from .subagent_utils import is_safe_workdir
async def call(self, context, **kwargs) -> str:
...
workdir = kwargs.get("workdir")
if workdir is None or not is_safe_workdir(workdir):
workdir = get_astrbot_temp_path()
...This keeps the tool focused on wiring config and leaves security policy in a reusable, testable function.
2. Centralize magic protocol strings
Multiple tools depend on string constants like __DYNAMIC_TOOL_CREATED__, __SUBAGENT_REMOVED__, __HISTORY_CLEARED__, __SHARED_CONTEXT_ADDED__. To reduce coupling and scattered string comparisons, introduce a small result type and a common mapper.
Example: define structured results and constants:
# subagent_protocol.py
from enum import Enum
from dataclasses import dataclass
from typing import Optional
class SubAgentAction(Enum):
CREATED = "created"
CREATE_FAILED = "create_failed"
REMOVED = "removed"
HISTORY_CLEARED = "history_cleared"
SHARED_CONTEXT_ADDED = "shared_context_added"
ERROR = "error"
@dataclass
class SubAgentResult:
action: SubAgentAction
message: Optional[str] = None
tool_name: Optional[str] = None
handoff_tool_name: Optional[str] = NoneChange SubAgentManager.create_subagent (and others) to return SubAgentResult instead of protocol strings (or in addition, if you need backward compatibility). Then keep protocol string formatting in one helper:
# subagent_protocol.py
def encode_dynamic_tool_created(result: SubAgentResult) -> str:
if result.action == SubAgentAction.CREATED:
return f"__DYNAMIC_TOOL_CREATED__:{result.tool_name}:{result.handoff_tool_name}:{result.message}"
if result.action == SubAgentAction.CREATE_FAILED:
return f"__DYNAMIC_TOOL_CREATE_FAILED__:{result.tool_name}"
return result.message or "Unknown result"Then the tool becomes:
from .subagent_protocol import encode_dynamic_tool_created
async def call(self, context, **kwargs) -> str:
...
result = await SubAgentManager.create_subagent(session_id=session_id, config=config)
return encode_dynamic_tool_created(result)Similarly, RemoveSubagentTool, ResetSubAgentTool, and shared-context tools can consume structured results and call shared format/encode helpers instead of locally comparing magic constants.
3. Extract the polling state machine from WaitForSubagentTool
WaitForSubagentTool.call is doing a lot (task selection + polling + formatting). You can extract the polling logic and status handling into a reusable helper, leaving the tool to argument validation and presentation.
For instance:
# subagent_waiter.py
import time
import asyncio
from typing import Optional
async def wait_for_task_completion(
session_id: str,
subagent_name: str,
task_id: Optional[str],
timeout: float,
poll_interval: float,
):
start_time = time.time()
while time.time() - start_time < timeout:
session = SubAgentManager.get_session(session_id)
if not session:
return {"type": "error", "reason": "session_not_found"}
if subagent_name not in session.subagents:
return {"type": "error", "reason": "agent_removed"}
status = SubAgentManager.get_subagent_status(session_id, subagent_name)
if status in ("COMPLETED", "FAILED"):
result = SubAgentManager.get_subagent_result(session_id, subagent_name, task_id)
return {
"type": status.lower(),
"result": result,
}
if status == "IDLE":
return {"type": "idle"}
await asyncio.sleep(poll_interval)
return {"type": "timeout", "task_id": task_id}Then the tool:
from .subagent_waiter import wait_for_task_completion
async def call(self, context, **kwargs) -> str:
...
# keep the logic that selects/infers task_id as-is
outcome = await wait_for_task_completion(
session_id, subagent_name, task_id, timeout, poll_interval
)
if outcome["type"] == "completed":
result = outcome["result"]
return (
f"SubAgent '{result.agent_name}' execution completed\n"
f" Task id: {result.task_id}\n"
f" Execution time: {result.execution_time:.1f}s\n--- Result ---\n{result.result}\n"
)
if outcome["type"] == "failed":
result = outcome["result"]
return (
f"SubAgent '{result.agent_name}' execution failed\n"
f"Task id: {result.task_id}\n"
f"Execution time: {result.execution_time:.1f}s\n"
f"Error: {result.error or 'Unknown error'}\n"
)
if outcome["type"] == "idle":
return f"Error: SubAgent '{subagent_name}' is running no tasks."
if outcome["type"] == "timeout":
target = f"Task {outcome['task_id']}"
return (
f" Timeout! \nSubAgent '{subagent_name}' has not finished '{target}' in {timeout}s. "
"The task may be still running. You can continue waiting by `wait_for_subagent` again."
)
if outcome["type"] == "error":
if outcome["reason"] == "session_not_found":
return "Error: Session Not Found"
if outcome["reason"] == "agent_removed":
return f"Error: SubAgent '{subagent_name}' not found. It may be removed."This keeps the state machine reusable (for other background tasks) and makes the tool’s control flow straightforward.
4. Split the module by concern
Given the size and mix of responsibilities, you can improve cohesion by moving tools into small, focused modules while keeping the public API unchanged:
subagent_tools/
__init__.py # re-exports current constants
lifecycle.py # Create/Remove/Protect/Unprotect/Reset/List tools
shared_context.py # Send/View shared context tools
waiters.py # WaitForSubagentTool and polling helpers
utils.py # is_safe_workdir, protocol helpers
__init__.py can keep the existing names:
from .lifecycle import (
CREATE_SUBAGENT_TOOL,
REMOVE_SUBAGENT_TOOL,
LIST_SUBAGENTS_TOOL,
RESET_SUBAGENT_TOOL,
PROTECT_SUBAGENT_TOOL,
UNPROTECT_SUBAGENT_TOOL,
)
from .shared_context import (
SEND_SHARED_CONTEXT_TOOL,
SEND_SHARED_CONTEXT_TOOL_FOR_MAIN_AGENT,
VIEW_SHARED_CONTEXT_TOOL,
)
from .waiters import WAIT_FOR_SUBAGENT_TOOLThis keeps current imports working while reducing per-file complexity and clarifying boundaries between lifecycle, shared-context, and background-task behavior.
Contributor
Author
|
所有功能已完成并通过测试，目前贴出的测试为早期实现，后续会补充更详尽的测试报告 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Modifications / 改动点
一、概述
增强版SubAgent对AstrBot的Subagent进行了功能扩充，支持两种使用方式：
subagent_orchestrator.agents在配置文件中预定义子代理所有子代理（无论静态还是动态）都由统一的
SubAgentManager管理，支持以下高级功能：独立上下文记忆、独立工具隔离、工作目录隔离、技能隔离、公共上下文共享、超时限制、后台任务模式等。核心特性
SubAgentManager统一管理create_subagent在运行时动态创建子代理wait_for_subagent阻塞式等待输出二、配置说明
2.1 配置项说明
在
cmd_config.json中通过subagent_orchestrator配置块进行设置：2.2 配置项详解
main_enablefalseremove_main_duplicate_toolsfalserouter_system_prompt""agents[]enabledfalsemax_dynamic_subagent_count3auto_cleanup_per_turntruerule_prompttools_blacklisttools_inherenttime_prompt_enabledtruehistory_enabledtruetrue可以保留跨多轮对话的记忆shared_context_enabledtrueshared_context_maxlen300subagent_history_maxlen300execution_timeout1200.0三、核心文件
3.1 subagent_manager.py
文件路径：
astrbot/core/subagent_manager.py核心数据结构：
核心类：
SubAgentManagerconfigure()get_execution_timeout()is_auto_cleanup_per_turn()is_shared_context_enabled()is_history_enabled()register_blacklisted_tool(tool_name)register_inherent_tool(tool_name)create_subagent(session_id, config, protected=False)register_static_subagent(session_id, handoff_tool, skills, workdir)remove_subagent(session_id, agent_name)cleanup_session_turn_end(session_id)protect_subagent(session_id, agent_name)is_protected(session_id, agent_name)update_subagent_history(session_id, agent_name, current_messages)get_subagent_history(session_id, agent_name)clear_subagent_history(session_id, agent_name)build_static_subagent_prompts(session_id, agent_name)build_dynamic_subagent_prompts(session_id, agent_name, runtime)build_task_router_prompt(session_id)get_subagent_tools(session_id, agent_name)add_shared_context(session_id, sender, context_type, content, target)get_shared_context(session_id, filter_by_agent)_build_shared_context_prompt(session_id, agent_name)create_pending_subagent_task(session_id, agent_name)get_pending_subagent_tasks(session_id, agent_name)get_latest_task_id(session_id, agent_name)store_subagent_result(session_id, agent_name, success, result, task_id, error, execution_time, metadata)get_subagent_result(session_id, agent_name, task_id)has_subagent_result(session_id, agent_name, task_id)clear_subagent_result(session_id, agent_name, task_id)get_subagent_status(session_id, agent_name)get_all_subagent_status(session_id)set_subagent_status(session_id, agent_name, status)cleanup_shared_context_by_agent(session_id, agent_name)clear_shared_context(session_id)set_shared_context_enabled(session_id, enabled)set_history_enabled(session_id, enabled)get_handoff_tools_for_session(session_id)cleanup_session(session_id)3.2 subagent_tools.py
文件路径：
astrbot/core/subagent_tools.py该文件包含所有子代理管理工具的类定义，这些工具供主Agent在运行时调用以创建、管理和与子代理交互。
核心工具类：
CreateSubAgentToolcreate_subagentRemoveSubagentToolremove_subagentListSubagentsToollist_subagentsResetSubAgentToolreset_subagentProtectSubagentToolprotect_subagentUnprotectSubagentToolunprotect_subagentSendSharedContextToolForMainAgentsend_shared_context_for_main_agentSendSharedContextToolsend_shared_contextViewSharedContextToolview_shared_contextWaitForSubagentToolwait_for_subagent四、功能实现
4.1 静态子代理
功能描述
静态子代理是原版的子代理实现，由配置文件定义人格。
为了实现统一管理，静态子代理在
build_main_agent()时自动注册到SubAgentManager，可以享受历史记忆、公共上下文、Skills调用等增强功能（如果开启）。实现方式
astrbot/core/subagent_orchestrator.pySubAgentOrchestrator.register_static_subagents_to_manager()astr_main_agent.py的_apply_subagent_manager_tools()中注册流程
build_main_agent()构建主Agent时调用_apply_subagent_manager_tools()_apply_subagent_manager_tools()调用so.register_static_subagents_to_manager(session_id)SubAgentOrchestrator.handoffs中的所有静态 handoff 工具SubAgentManager.register_static_subagent()register_static_subagent()内部：HandoffTool提取 agent 信息SubAgentConfigcreate_subagent(session_id, config, protected=True)4.2 动态子代理
功能描述
当动态子代理功能开启时，主Agent可以通过
create_subagent工具动态创建子代理，每个子代理拥有独立的人设、工具、技能和工作目录配置。随后系统会创建对应的transfer_to_xxx工具实现方式
astrbot/core/subagent_tools.pyCreateSubAgentToolname: 子代理名称system_prompt: 子代理的人设和系统提示tools: 可用工具列表（字符串名称）skills: 可用技能列表（字符串名称）workdir: 子代理工作目录（绝对路径，可选）provider_id: 子代理使用的LLM提供商ID（可选，默认跟随全局设置）使用示例
名称验证规则
^[a-zA-Z][a-zA-Z0-9_]{0,31}$）__开头和结尾创建后流程
astrbot_execute_shell,astrbot_execute_python。若公共上下文启用，还会添加send_shared_context）SubAgentConfig配置对象Agent和HandoffTool对象SubAgentManager__DYNAMIC_TOOL_CREATED__标记的消息，触发工具schema刷新工作目录安全检查
CreateSubAgentTool._check_path_safety()对传入的workdir进行安全验证：..windows,system32,syswow64,boot等/etc,/bin,/sbin,/root等/System,/Library,/private/var,/usr等验证失败时，工作目录回退到
get_astrbot_temp_path()。工具黑名单与固有工具
4.3 动态子代理委派 (transfer_to_xxx)
功能描述
创建子代理后，主Agent使用
transfer_to_xxx工具将任务委派给对应子代理。实现方式
astrbot/core/astr_agent_tool_exec.pyFunctionToolExecutor._execute_handoff()/_execute_handoff_background()子代理Prompt构建
委派执行流程
transfer_to_xxx工具ToolLoopAgentRunner._handle_function_tools()在SubAgentManager中查找handoff工具HandoffTool对象FunctionToolExecutor._execute_handoff()执行委派send_shared_context工具到子代理工具集provider_id使用不同的LLM提供商_get_subagent_execution_timeout()获取超时时间asyncio.wait_for()添加执行超时控制_handle_subagent_timeout()将状态设为 "FAILED"历史上下文注入流程
subagent_histories获取历史消息Message对象begin_dialogs之前runner_messages（Agent运行期间的所有消息）追加到历史subagent_history_maxlen（默认300条），超出时保留最新的。（此处仅用于约束数组长度，实际上下文长度仍由astrbot的truncate管理）执行超时控制
4.4 后台任务等待
功能描述
当主Agent认为某个任务耗时很长时，可以让SubAgent以后台模式运行，主Agent不会被阻塞，可继续执行其他可并行的任务
原版Subagent结束任务时，会通过
_wake_main_agent_for_background_result唤醒主Agent如果Subagent耗时很长，一些聪明的Agent可能会执行
time.sleep()的python代码进行等待，但我们要假设Agent是愚蠢的。事实上，大多数情况下，主Agent会觉得迟迟拿不到结果而试图自己执行，导致同一任务完成两遍。因此引入了一个
wait_for_subagent工具，主Agent需要拿到Subagent结果时，可进行主动的阻塞式等待，避免此类情况发生实现方式
astrbot/core/astr_agent_tool_exec.py_execute_handoff_background()- 执行后台委派_do_handoff_background()- 后台任务执行逻辑_register_subagent_task()- 注册SubAgent任务_handle_subagent_background_result()- 处理SubAgent结果_maybe_wake_main_agent_after_background()- 智能唤醒主Agent。Subagent完成任务时，如果主Agent已经结束运行，才执行_wake_main_agent_for_background_result，否则把结果存到subagent_background_results里执行流程
主Agent状态检测
后台任务完成后，需要决定是否通知用户。关键逻辑：
这依赖
AstrAgentContext.extra字段，在build_main_agent()中注入：主动等待工具
WaitForSubagentTool 轮询逻辑
task_id，获取最新的 pending 任务IDLE→ 返回错误：子代理未在运行任务COMPLETED→ 返回执行结果FAILED→ 返回失败信息RUNNING→ 继续等待后台Subagent任务管理
每个subagent都会储存结果到
subagent_background_results中。同一个subagent可以有多个任务，通过task_id来区分（每次创建任务递增）创建任务
获取结果
4.5 子代理历史记忆
功能描述
history_enabled=true时，子代理可以保留跨轮对话的历史上下文，实现连续对话能力。实现方式
SubAgentSession.subagent_historiessubagent_history_maxlen（默认300条）SubAgentManager.is_history_enabled()/set_history_enabled()历史管理机制
runner_messages（Agent运行期间的消息）追加到历史role=system的消息_MAX_TOOL_RESULT_LEN（2000字符）的tool结果截断，附加...[truncated]subagent_history_maxlen时，保留最新的消息Message对象，合并到contexts历史清理
reset_subagent工具主动清除指定子代理的历史4.6 动态子代理工作目录隔离（软约束）
功能描述
每个子代理可配置独立的工作目录，未指定时，或工作目录非法时，默认使用
get_astrbot_temp_path()。实现方式
SubAgentConfig.workdirCreateSubAgentTool._check_path_safety()_build_workdir_prompt()注入的工作目录提示
4.7 动态子代理行为规范注入
功能描述
子代理自动注入安全模式、输出规范和时间信息等行为约束。
注入内容
角色定义
_build_subagent_system_prompt()：# Role Your name is {agent_name}(used for tool calling) {base_instructions}静态行为规范
_build_rule_prompt()，可由用户在配置中决定。默认配置：动态时间信息
_build_time_prompt()：# Current Time 2026-04-14 23:09 (CST)4.8 Skills隔离
功能描述
修复了Subagent无法使用skills的问题，每个子代理可分配不同的Skills，相互隔离。
注入逻辑
SubAgentConfig.skills）SkillManager.list_skills()获取所有可用技能build_skills_prompt()生成提示词build_dynamic_subagent_prompts()注入到子代理的system_prompt4.9 公共上下文
功能描述
当
shared_context_enabled=true时，维护一个所有子代理共享的、实时更新的群聊区域。在SubAgent每次调用LLM之前，公共上下文的内容会被注入到子代理的追加信息中。公共上下文中每条信息的格式
上下文类型
statusmessagesystem公共上下文注入方式
按类型和优先级分组注入到子代理的追加信息中，让Agent可以更清晰地获取共享上下文的信息，并知道哪些需要优先处理。
实现方式
astrbot/core/subagent_manager.pySubAgentManager._build_shared_context_prompt()Prompt结构
容量管理
shared_context_maxlen时，保留最近90%的消息公共上下文特点
4.10 主Agent路由提示
功能描述
当
dynamic_agents.enabled=true时，在主Agent的System Prompt中注入动态SubAgent能力说明，包括创建指南和委派流程。如果dynamic_agents.enabled=false，则仍然使用router_system_prompt动态子代理路由的注入内容
核心方法：
SubAgentManager.build_task_router_prompt()4.11 动态子代理的自动清理
功能描述
当
auto_cleanup_per_turn=true时，每轮对话结束后，自动清理产生的动态子代理。auto_cleanup_per_turn=true，但又想要保留某个子代理留作后续使用，可以由主agent调用protect_subagent管理工具为子代理添加保护，防止被自动清理。主Agent也可以根据需要，手动通过
unprotect_subagent移除保护，或直接通过remove_subagent工具清理某个子代理。清理规则
remove_subagent()进行清理（移除配置、handoff工具、历史、后台结果）触发时机
在
internal.py的 agent 结束流程中：4.12 动态子代理数量限制
功能描述
限制单个会话中子代理的最大数量，防止资源耗尽。
注意：替换已存在的同名子代理不增加计数。
4.13 执行超时控制
功能描述
除了原有的工具调用超时外，为每个子代理设置总的执行超时时间，避免无限等待。这是一项全局设置，对所有子代理有效
实现方式
SubAgentConfig.execution_timeout（默认1200秒）subagent_orchestrator.execution_timeoutSubAgentManager.get_execution_timeout()超时处理流程
五、工具列表
5.1 主Agent新增可用工具
create_subagentremove_subagentlist_subagentsreset_subagentprotect_subagentunprotect_subagentview_shared_contextsend_shared_context_for_main_agentwait_for_subagent5.2 子代理可用工具
send_shared_contexttools_inherent给出六、修改文件清单
astrbot/core/subagent_manager.pySubAgentManager，包含会话管理、后台任务管理、历史信息、公共上下文、Prompt构建等；以及配套的SubAgentConfig，SubAgentSession数据结构astrbot/core/subagent_tools.pyastrbot/core/astr_agent_tool_exec.py_execute_handoff()（静态/动态Prompt构建、历史注入、超时控制）；_execute_handoff_background()（pending任务创建）；新增_do_handoff_background()（结果存储与智能唤醒逻辑）；新增_register_subagent_task()；新增_load_subagent_history、_build_subagent_system_prompt、_save_subagent_history、_handle_subagent_timeout、_is_managed_subagent、_handle_subagent_background_result、_maybe_wake_main_agent_after_background等方法封装核心逻辑astrbot/core/astr_main_agent.py_apply_subagent_manager_tools()，在enable_dynamic=True时注册动态管理工具。在build_main_agent()中注入main_agent_runner到AstrAgentContext.extraastrbot/core/subagent_orchestrator.pyregister_static_subagents_to_manager()方法；静态subagent自动注册到SubAgentManager享受增强功能astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.pyastrbot/core/agent/runners/tool_loop_agent_runner.py_handle_function_tools()中集成动态SubAgent工具查找；新增_resolve_dynamic_subagent_tool()和_maybe_register_dynamic_tool_from_result()方法；astrbot/core/config/default.pydynamic_agents、time_prompt_enabled、history_enabled、shared_context_enabled、shared_context_maxlen、subagent_history_maxlen、execution_timeout等配置项astrbot/core/astr_agent_context.pyextra字段的类型改成dict[str, Any]，使得可以记录任意类型的extra信息。例如在本PR中，它存储了一个Agent对象astrbot/core/star/context.pytool_loop_agent()支持runner_messages参数（通过kwargs传递），用于记录SubAgent历史astrbot/dashboard/routes/subagent.pydashboard/src/i18n/locales/en-US/features/subagent.jsondashboard/src/i18n/locales/ru-RU/features/subagent.jsondashboard/src/i18n/locales/zh-CN/features/subagent.jsondashboard/src/views/SubAgentPage.vue七、使用流程
7.1 静态子代理使用流程
与原版一致，但子代理获得增强
7.2 动态子代理使用流程
7.3 带保护的多轮对话流程
7.4 带公共上下文的协作流程
7.5 后台任务并行处理流程
7.6 后台任务与共享上下文结合
八、API 参考
8.1 SubAgentManager 核心方法
配置与生命周期
configure(max_subagent_count, auto_cleanup_per_turn, shared_context_enabled, shared_context_maxlen, subagent_history_maxlen, tools_blacklist, tools_inherent, execution_timeout, history_enabled, rule_prompt, time_prompt_enabled, timezone)get_execution_timeout() -> floatis_auto_cleanup_per_turn() -> boolis_shared_context_enabled() -> boolis_history_enabled() -> boolset_history_enabled(session_id, enabled)register_blacklisted_tool(tool_name)register_inherent_tool(tool_name)子代理管理
create_subagent(session_id, config, protected=False) -> tupleregister_static_subagent(session_id, handoff_tool, skills, workdir) -> tupleremove_subagent(session_id, agent_name) -> strprotect_subagent(session_id, agent_name)is_protected(session_id, agent_name) -> boolset_subagent_status(session_id, agent_name, status)cleanup_session_turn_end(session_id) -> dictcleanup_session(session_id) -> dictget_handoff_tools_for_session(session_id) -> listget_subagent_tools(session_id, agent_name) -> list | None历史管理
update_subagent_history(session_id, agent_name, current_messages)get_subagent_history(session_id, agent_name) -> listclear_subagent_history(session_id, agent_name) -> strPrompt构建
build_task_router_prompt(session_id) -> strbuild_static_subagent_prompts(session_id, agent_name) -> strbuild_dynamic_subagent_prompts(session_id, agent_name, runtime) -> str后台任务管理
create_pending_subagent_task(session_id, agent_name) -> strget_pending_subagent_tasks(session_id, agent_name) -> list[str]get_latest_task_id(session_id, agent_name) -> str | Nonestore_subagent_result(session_id, agent_name, success, result, task_id, error, execution_time, metadata)get_subagent_result(session_id, agent_name, task_id) -> SubAgentExecutionResult | Nonehas_subagent_result(session_id, agent_name, task_id) -> boolclear_subagent_result(session_id, agent_name, task_id)get_subagent_status(session_id, agent_name) -> strget_all_subagent_status(session_id) -> dict公共上下文管理
add_shared_context(session_id, sender, context_type, content, target)get_shared_context(session_id, filter_by_agent) -> listset_shared_context_enabled(session_id, enabled)cleanup_shared_context_by_agent(session_id, agent_name)clear_shared_context(session_id)8.2 SubAgentExecutionResult 数据结构
8.3 SubAgentConfig 数据结构
8.4 SubAgentSession 数据结构
九、Dashboard
为子代理页面添加更详细的用户配置
Screenshots or Test Results / 运行截图或测试结果
测试1：动态Agent创建和保护机制
测试流程：动态建立若干个SubAgent，给其中一些加入保护，另一些不加保护，观察清理情况，以及跨轮对话的记忆
测试1——动态创建子Agent和保护机制.md
测试2：子Agent历史上下文
测试流程：建立一个SubAgent，并使其跨多轮对话，查看其是否有多轮对话的记忆。
测试2——子Agent历史上下文.md
测试3：子Agent共享上下文
测试流程
send_shared_context发送工具向共享上下文中添加内容测试3——子Agent共享上下文.md
测试4：异步与等待
测试流程
设计一个同时包含并行与串行、前台与后台模式的任务，观察工作流程
测试4——异步与等待.md
测试5：超长上下文工作实例
测试流程
让agent阅读一个大型项目的全部代码，并生成细致的介绍文档
测试5——动态处理超长上下文示例.md
Checklist / 检查清单
😊 If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
/ 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。
👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
/ 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。
🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in
requirements.txtandpyproject.toml./ 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到
requirements.txt和pyproject.toml文件相应位置。😮 My changes do not introduce malicious code.
/ 我的更改没有引入恶意代码。
Summary by Sourcery
Introduce a unified SubAgentManager that manages both static and dynamically created sub-agents, adds enhanced subagent runtime features and background task controls, and exposes corresponding configuration, tooling, and dashboard UI to manage them.
New Features:
Enhancements:
Tests:
Summary by Sourcery
Introduce a unified SubAgentManager that manages both static and dynamically created sub-agents, adds enhanced subagent runtime features and background task controls, and exposes corresponding configuration and dashboard UI to manage them.
New Features:
Enhancements:
Tests: