Function run_agent_loop

src/mcp_cli/agents/loop.py:17–113

Run a headless chat loop for a spawned agent. Similar to ``_run_enhanced_chat_loop`` in ``chat_handler.py`` but: * No terminal I/O — reads prompts from *input_queue*. * Uses ``HeadlessUIManager`` (no rich/curses). * Reports completion via *done_event*. * Returns a summary of th

(
    context: Any,
    ui_manager: Any,
    input_queue: asyncio.Queue,
    done_event: asyncio.Event,
    max_turns: int = 50,
)

async def run_agent_loop(
    context: Any,
    ui_manager: Any,
    input_queue: asyncio.Queue,
    done_event: asyncio.Event,
    max_turns: int = 50,
) -> str:
    """Run a headless chat loop for a spawned agent.

    Similar to ``_run_enhanced_chat_loop`` in ``chat_handler.py`` but:

    * No terminal I/O — reads prompts from *input_queue*.
    * Uses ``HeadlessUIManager`` (no rich/curses).
    * Reports completion via *done_event*.
    * Returns a summary of the final assistant response.

    Parameters
    ----------
    context:
        ChatContext for this agent.
    ui_manager:
        HeadlessUIManager instance.
    input_queue:
        ``asyncio.Queue`` that receives user/supervisor prompts.
    done_event:
        Set when the loop finishes.
    max_turns:
        Maximum conversation turns per prompt.

    Returns
    -------
    str
        Summary of the agent's last response, or error description.
    """
    from mcp_cli.chat.conversation import ConversationProcessor

    convo = ConversationProcessor(context, ui_manager)
    last_response = ""

    # Wire dashboard bridge if present
    if bridge := getattr(context, "dashboard_bridge", None):
        bridge.set_input_queue(input_queue)

    try:
        while not context.exit_requested:
            try:
                # Wait for a prompt (with timeout so we can check exit)
                try:
                    user_msg = await asyncio.wait_for(input_queue.get(), timeout=1.0)
                except asyncio.TimeoutError:
                    continue

                if user_msg is None:
                    continue

                # Exit signals
                msg_lower = str(user_msg).lower().strip()
                if msg_lower in ("exit", "quit", "__stop__"):