chore(runner): refactored agent runner, preparing for parallel delegate tasks

Author: daltonnyx

AgentCrew/app.py

@@ -303,7 +303,8 @@ def run_job(
         memory_llm: Optional[str] = None,
         output_schema: Optional[str] = None,
     ) -> str:
-        from AgentCrew.modules.chat import MessageHandler
+        from AgentCrew.modules.agents.agent_runner import run_agent_loop
+        from AgentCrew.modules.agents.base import MessageType
         from AgentCrew.modules.mcpclient import MCPSessionManager
         from AgentCrew.modules.llm.model_registry import ModelRegistry
 
@@ -361,37 +362,63 @@ def run_job(
 
                 self.agent_manager.select_agent(current_agent.name)
 
-                message_handler = MessageHandler(
-                    services["memory"], services["context_persistent"]
-                )
-                message_handler.is_non_interactive = True
-                message_handler.agent = current_agent
+                history: List[Dict[str, Any]] = []
 
                 if files:
+                    from AgentCrew.modules.utils.file_handler import FileHandler
+
+                    file_handler = FileHandler()
+                    all_file_contents: List[Dict[str, Any]] = []
                     for file_path in files:
-                        asyncio.run(
-                            message_handler.process_user_input(f"/file {file_path}")
+                        file_path = os.path.expanduser(file_path.strip())
+                        file_content = file_handler.process_file(file_path)
+                        if not file_content:
+                            file_content = current_agent.format_message(
+                                MessageType.FileContent, {"file_uri": file_path}
+                            )
+                        if file_content:
+                            all_file_contents.append(file_content)
+                    if all_file_contents:
+                        history.append(
+                            {
+                                "role": "user",
+                                "content": all_file_contents,
+                            }
                         )
 
+                history.append(
+                    {
+                        "role": "user",
+                        "content": [{"type": "text", "text": task}],
+                    }
+                )
+
                 max_attempts = 4
                 attempt = 0
                 response = None
 
-                asyncio.run(message_handler.process_user_input(task))
-
                 while attempt < max_attempts:
                     attempt += 1
-                    response, _, _ = asyncio.run(
-                        message_handler.get_assistant_response()
+                    response = asyncio.run(
+                        run_agent_loop(
+                            agent=current_agent,
+                            history=history,
+                        )
                     )
                     if not output_schema or not schema_dict:
                         break
 
                     if response is None:
-                        asyncio.run(
-                            message_handler.process_user_input(
-                                "No response was generated. Please try again."
-                            )
+                        history.append(
+                            {
+                                "role": "user",
+                                "content": [
+                                    {
+                                        "type": "text",
+                                        "text": "No response was generated. Please try again.",
+                                    }
+                                ],
+                            }
                         )
                         continue
 
@@ -402,12 +429,17 @@ def run_job(
                         break
                     else:
                         if retry_message:
-                            asyncio.run(
-                                message_handler.process_user_input(retry_message)
+                            history.append(
+                                {
+                                    "role": "user",
+                                    "content": [
+                                        {"type": "text", "text": retry_message}
+                                    ],
+                                }
                             )
 
                 MCPSessionManager.get_instance().cleanup()
-                return response.strip() if response else ""
+                return self._clean_json_response(response).strip() if response else ""
             else:
                 raise ValueError(f"Agent '{agent}' not found")
 

AgentCrew/modules/a2a/task_manager.py

@@ -156,7 +156,7 @@ async def on_send_message(
             (m for m in message.get("content", []) if m.get("type", "text") == "file"),
             None,
         ):
-            from AgentCrew.modules.chat.file_handler import FileHandler
+            from AgentCrew.modules.utils.file_handler import FileHandler
 
             new_parts = []
             if self.file_handler is None:

AgentCrew/modules/agents/agent_runner.py

@@ -0,0 +1,87 @@
+from typing import Any, Callable, Dict, List, Optional
+
+from AgentCrew.modules.agents.local_agent import LocalAgent
+from AgentCrew.modules.agents.base import MessageType
+
+
+async def run_agent_loop(
+    agent: LocalAgent,
+    history: List[Dict[str, Any]],
+    *,
+    tool_filter: Optional[Callable[[Dict[str, Any]], bool]] = None,
+) -> str:
+    current_response = ""
+    thinking_content = ""
+    thinking_signature = ""
+    tool_uses: List[Dict[str, Any]] = []
+    input_tokens = 0
+    output_tokens = 0
+
+    def process_result(_tool_uses, _input_tokens, _output_tokens):
+        nonlocal tool_uses, input_tokens, output_tokens
+        tool_uses = _tool_uses
+        input_tokens += _input_tokens
+        output_tokens += _output_tokens
+
+    async for (
+        response_message,
+        chunk_text,
+        thinking_chunk,
+    ) in agent.process_messages(history, callback=process_result):
+        if response_message:
+            current_response = response_message
+        if thinking_chunk:
+            think_text_chunk, signature = thinking_chunk
+            if think_text_chunk:
+                thinking_content += think_text_chunk
+            if signature:
+                thinking_signature += signature
+
+    if not tool_uses:
+        return current_response
+
+    if tool_filter:
+        filtered = [t for t in tool_uses if tool_filter(t)]
+    else:
+        filtered = tool_uses
+
+    if not filtered:
+        return current_response
+
+    thinking_data = (thinking_content, thinking_signature) if thinking_content else None
+    thinking_message = agent.format_message(
+        MessageType.Thinking, {"thinking": thinking_data}
+    )
+    if thinking_message:
+        history.append(thinking_message)
+
+    assistant_message = agent.format_message(
+        MessageType.Assistant,
+        {"message": current_response, "tool_uses": filtered},
+    )
+    if assistant_message:
+        history.append(assistant_message)
+
+    for tool_use in filtered:
+        try:
+            tool_result = await agent.execute_tool_call(
+                tool_use["name"], tool_use["input"]
+            )
+        except Exception as e:
+            tool_result = str(e)
+            error_msg = agent.format_message(
+                MessageType.ToolResult,
+                {"tool_use": tool_use, "tool_result": tool_result, "is_error": True},
+            )
+            if error_msg:
+                history.append(error_msg)
+            continue
+
+        result_msg = agent.format_message(
+            MessageType.ToolResult,
+            {"tool_use": tool_use, "tool_result": tool_result},
+        )
+        if result_msg:
+            history.append(result_msg)
+
+    return await run_agent_loop(agent, history, tool_filter=tool_filter)

AgentCrew/modules/agents/tools/delegate.py

@@ -2,9 +2,7 @@
 import asyncio
 
 from AgentCrew.modules.agents import AgentManager
-
-# from AgentCrew.modules.llm.message import MessageTransformer
-from AgentCrew.modules.agents.base import MessageType
+from AgentCrew.modules.agents.agent_runner import run_agent_loop
 
 
 def get_delegate_tool_definition(provider="claude") -> Dict[str, Any]:
@@ -121,10 +119,10 @@ def handler(**params) -> str:
 
         try:
             # Get the target agent
-            target_agent = agent_manager.get_agent(target_agent_name)
+            target_agent = agent_manager.get_local_agent(target_agent_name)
             if not target_agent:
                 raise ValueError(
-                    f"Error: Could not retrieve agent '{target_agent_name}'"
+                    f"Error: Could not retrieve local agent '{target_agent_name}'"
                 )
 
             # Prepare context from current conversation
@@ -177,106 +175,14 @@ def handler(**params) -> str:
 
             try:
 
-                async def _process_delegation():
-                    """Process delegation similar to _process_agent_task - runs until no tool uses"""
-
-                    current_response = ""
-                    tool_uses = []
-                    input_tokens = 0
-                    output_tokens = 0
-                    thinking_content = ""
-
-                    def process_result(_tool_uses, _input_tokens, _output_tokens):
-                        nonlocal tool_uses, input_tokens, output_tokens
-                        tool_uses = _tool_uses
-                        input_tokens += _input_tokens
-                        output_tokens += _output_tokens
-
-                    if not target_agent:
-                        return current_response
-                    async for (
-                        response_message,
-                        chunk_text,
-                        thinking_chunk,
-                    ) in target_agent.process_messages(
-                        delegate_history, callback=process_result
-                    ):
-                        if response_message:
-                            current_response = response_message
-
-                        if thinking_chunk:
-                            think_text_chunk, _ = thinking_chunk
-                            if think_text_chunk:
-                                thinking_content += think_text_chunk
-
-                    if tool_uses and len(tool_uses) > 0:
-                        filtered_tool_uses = [
-                            t
-                            for t in tool_uses
-                            if t["name"] not in ["transfer", "delegate"]
-                        ]
-
-                        if not filtered_tool_uses:
-                            return current_response
-
-                        # Add thinking content as a separate message if available
-                        thinking_data = (
-                            (thinking_content, None) if thinking_content else None
-                        )
-                        if thinking_data:
-                            thinking_message = target_agent.format_message(
-                                MessageType.Thinking, {"thinking": thinking_data}
-                            )
-                            if thinking_message:
-                                delegate_history.append(thinking_message)
-
-                        # Format assistant message with the response and tool uses
-                        assistant_message = target_agent.format_message(
-                            MessageType.Assistant,
-                            {
-                                "message": current_response,
-                                "tool_uses": filtered_tool_uses,
-                            },
-                        )
-                        if assistant_message:
-                            delegate_history.append(assistant_message)
-
-                        # Process each tool use
-                        for tool_use in filtered_tool_uses:
-                            try:
-                                tool_result = await target_agent.execute_tool_call(
-                                    tool_use["name"],
-                                    tool_use["input"],
-                                )
-
-                                tool_result_message = target_agent.format_message(
-                                    MessageType.ToolResult,
-                                    {
-                                        "tool_use": tool_use,
-                                        "tool_result": tool_result,
-                                    },
-                                )
-                                if tool_result_message:
-                                    delegate_history.append(tool_result_message)
-
-                            except Exception as e:
-                                error_message = target_agent.format_message(
-                                    MessageType.ToolResult,
-                                    {
-                                        "tool_use": tool_use,
-                                        "tool_result": str(e),
-                                        "is_error": True,
-                                    },
-                                )
-                                if error_message:
-                                    delegate_history.append(error_message)
-
-                        # Continue the loop to process more messages
-                        return await _process_delegation()
-
-                    return current_response
-
-                response = asyncio.run(_process_delegation())
+                async def _do_delegation():
+                    return await run_agent_loop(
+                        agent=target_agent,
+                        history=delegate_history,
+                        tool_filter=lambda t: t["name"] not in ["transfer", "delegate"],
+                    )
+
+                response = asyncio.run(_do_delegation())
 
             except Exception as e:
                 raise ValueError(

AgentCrew/modules/chat/message/command_processor.py

@@ -3,7 +3,7 @@
 from typing import Optional, Tuple, List, Dict, Any
 import os
 
-from AgentCrew.modules.chat.file_handler import FileHandler
+from AgentCrew.modules.utils.file_handler import FileHandler
 from AgentCrew.modules.llm.model_registry import ModelRegistry
 from AgentCrew.modules.llm.service_manager import ServiceManager
 from AgentCrew.modules.chat.consolidation import ConversationConsolidator

AgentCrew/modules/chat/message/handler.py

@@ -9,7 +9,7 @@
 from AgentCrew.modules.agents.base import MessageType
 from AgentCrew.modules.chat.history import ChatHistoryManager
 from AgentCrew.modules.agents import AgentManager
-from AgentCrew.modules.chat.file_handler import FileHandler
+from AgentCrew.modules.utils.file_handler import FileHandler
 
 from AgentCrew.modules.memory import (
     BaseMemoryService,
@@ -74,7 +74,6 @@ def __init__(
         self._queued_attached_files = []
         self.stop_streaming = False
         self.streamline_messages = []
-        self.is_non_interactive = False
         self.current_conversation_id: Optional[str] = None  # ID for persistence
 
         # Initialize components
@@ -419,11 +418,6 @@ def process_result(_tool_uses, _input_tokens, _output_tokens):
             # Final assistant message
             self._notify("response_completed", assistant_response)
 
-            if self.is_non_interactive:
-                # No need to persist or store conversation turn since is single turn task
-                return assistant_response, input_tokens, output_tokens
-
-            # --- Start of Persistence Logic ---
             if self.current_conversation_id and self.last_assisstant_response_idx >= 0:
                 try:
                     messages_for_this_turn = self.get_recent_agent_responses()

AgentCrew/modules/chat/message/tool_manager.py

@@ -125,8 +125,7 @@ async def execute_tool(self, tool_use: Dict[str, Any]):
             return
 
         if (
-            not self.message_handler.is_non_interactive
-            and not self.get_effective_yolo_mode()
+            not self.get_effective_yolo_mode()
             and tool_name not in self._auto_approved_tools
         ):
             # Request confirmation from the user