refactor: extract event_loop_utils module and improve agent response handling (eigent-ai#999)

Co-authored-by: mkdev11 <MkDev11@users.noreply.github.com>
Co-authored-by: Tao Sun <168447269+fengju0213@users.noreply.github.com>
Co-authored-by: a7m-1st <Ahmed.jimi.awelkeir500@gmail.com>
Co-authored-by: Wendong-Fan <w3ndong.fan@gmail.com>
Co-authored-by: Claude <noreply@anthropic.com>

Author: 5 people

backend/app/agent/__init__.py

@@ -12,7 +12,7 @@
 # limitations under the License.
 # ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
 
-from app.agent.agent_model import agent_model, set_main_event_loop
+from app.agent.agent_model import agent_model
 from app.agent.factory import (
     browser_agent,
     developer_agent,
@@ -31,7 +31,6 @@
     "agent_model",
     "get_mcp_tools",
     "get_toolkits",
-    "set_main_event_loop",
     "browser_agent",
     "developer_agent",
     "document_agent",

backend/app/agent/agent_model.py

@@ -12,80 +12,20 @@
 # limitations under the License.
 # ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
 
-import asyncio
-import contextvars
 import logging
 import uuid
-from threading import Lock
 from typing import Any, Callable
 
 from camel.messages import BaseMessage
 from camel.models import ModelFactory
 from camel.toolkits import FunctionTool, RegisteredAgentToolkit
 from camel.types import ModelPlatformType
 
+from app.utils.event_loop_utils import _schedule_async_task
 from app.agent.listen_chat_agent import ListenChatAgent, logger
 from app.model.chat import AgentModelConfig, Chat
 from app.service.task import ActionCreateAgentData, Agents, get_task_lock
 
-# Thread-safe reference to main event loop using contextvars
-# This ensures each request has its own event loop reference,
-# avoiding race conditions
-_main_event_loop_var: contextvars.ContextVar[asyncio.AbstractEventLoop
-                                             | None] = contextvars.ContextVar(
-                                                 "_main_event_loop",
-                                                 default=None
-                                             )
-
-# Global fallback for main event loop reference
-# Used when contextvars don't propagate to worker threads
-# (e.g., asyncio.to_thread)
-_GLOBAL_MAIN_LOOP: asyncio.AbstractEventLoop | None = None
-_GLOBAL_MAIN_LOOP_LOCK = Lock()
-
-
-def set_main_event_loop(loop: asyncio.AbstractEventLoop | None):
-    """Set the main event loop reference for thread-safe task scheduling.
-
-    This should be called from the main async context before spawning threads
-    that need to schedule async tasks. Uses both contextvars (for request
-    isolation) and a global fallback (for thread pool workers where
-    contextvars may not propagate).
-    """
-    global _GLOBAL_MAIN_LOOP
-    _main_event_loop_var.set(loop)
-    with _GLOBAL_MAIN_LOOP_LOCK:
-        _GLOBAL_MAIN_LOOP = loop
-
-
-def _schedule_async_task(coro):
-    """Schedule an async coroutine as a task, thread-safe.
-
-    This function handles scheduling from both the main event loop thread
-    and from worker threads (e.g., when using asyncio.to_thread).
-    """
-    try:
-        # Try to get the running loop (works in main event loop thread)
-        loop = asyncio.get_running_loop()
-        loop.create_task(coro)
-    except RuntimeError:
-        # No running loop in this thread (we're in a worker thread)
-        # First try contextvars, then fallback to global reference
-        main_loop = _main_event_loop_var.get()
-        if main_loop is None:
-            with _GLOBAL_MAIN_LOOP_LOCK:
-                main_loop = _GLOBAL_MAIN_LOOP
-        if main_loop is not None and main_loop.is_running():
-            asyncio.run_coroutine_threadsafe(coro, main_loop)
-        else:
-            # This should not happen in normal operation - log error and skip
-            logging.error(
-                "No event loop available for async task "
-                "scheduling, task skipped. Ensure "
-                "set_main_event_loop() is called "
-                "before parallel agent creation."
-            )
-
 
 def agent_model(
     agent_name: str,