fix: MCP per-call timeout, wire workflow context into prompt, fix parallel loop idx

src/praisonai-agents/praisonaiagents/agents/agents.py

@@ -388,6 +388,18 @@ def _build_execution_context(agents_instance, task_id, skip_memory_init=False):
 
     # Build context first to include in task prompt
     context_text = ""
+    # Inter-task context / validation feedback assembled by the workflow process
+    # engine (Process._build_task_context) is stored on the task; fold it in so
+    # downstream/retried tasks actually see upstream output and rejection reasons.
+    extra_context = getattr(task, '_execution_context', None)
+    if extra_context:
+        context_text = extra_context
+        # NOTE: We intentionally do NOT clear _execution_context here. Task
+        # retries (guardrail/completion failures) re-enter this helper via the
+        # run_task/arun_task retry loops, and clearing would strip the upstream
+        # output + validation feedback the retry needs. The Process engine owns
+        # this field's lifecycle: it re-sets it before each task yield and resets
+        # every task's _execution_context to None before selecting the next task.
     if task.context:
         context_results = []  # Collect contexts then de-duplicate
         for context_item in task.context:
@@ -404,7 +416,11 @@ def _build_execution_context(agents_instance, task_id, skip_memory_init=False):
             for i, ctx in enumerate(unique_contexts):
                 logger.debug(f"Context {i+1}: {ctx[:100]}...")
         context_separator = '\n\n'
-        context_text = context_separator.join(unique_contexts)
+        joined_contexts = context_separator.join(unique_contexts)
+        if context_text and joined_contexts:
+            context_text = context_text + context_separator + joined_contexts
+        elif joined_contexts:
+            context_text = joined_contexts
 
     # Build task prompt using DRY helper
     task_prompt = _prepare_task_prompt(task, task_description, context_text)

src/praisonai-agents/praisonaiagents/mcp/mcp.py

@@ -78,12 +78,14 @@ async def _run_async(self):
                                 response_queue, kind, name, arguments = item
                                 try:
                                     if kind == "resource":
-                                        result = await session.read_resource(name)
+                                        result = await asyncio.wait_for(session.read_resource(name), timeout=self.timeout)
                                     elif kind == "prompt":
-                                        result = await session.get_prompt(name, arguments or None)
+                                        result = await asyncio.wait_for(session.get_prompt(name, arguments or None), timeout=self.timeout)
                                     else:
-                                        result = await session.call_tool(name, arguments)
+                                        result = await asyncio.wait_for(session.call_tool(name, arguments), timeout=self.timeout)
                                     response_queue.put((True, result))
+                                except asyncio.TimeoutError:
+                                    response_queue.put((False, f"MCP {kind} call timed out after {self.timeout} seconds (server side)"))
                                 except Exception as e:
                                     response_queue.put((False, str(e)))
                             except queue.Empty:

src/praisonai-agents/praisonaiagents/workflows/workflows.py

@@ -2821,21 +2821,24 @@ def execute_item(idx_item_tuple, opt_prev=optimized_previous):
                     emitter.clear_branch()
 
             with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
-                futures = [
-                    executor.submit(copy_context_to_callable(lambda pair=(idx, item): execute_item(pair)))
-                    for idx, item in enumerate(items)
-                ]
+                # Map each future back to its original index so error attribution
+                # survives even if execute_item raises before returning its index.
+                future_to_idx = {}
+                for idx, item in enumerate(items):
+                    fut = executor.submit(copy_context_to_callable(lambda pair=(idx, item): execute_item(pair)))
+                    future_to_idx[fut] = idx
 
                 # Collect results in order
                 indexed_results = []
-                for future in concurrent.futures.as_completed(futures):
+                for future in concurrent.futures.as_completed(future_to_idx):
+                    idx = future_to_idx[future]
                     try:
-                        idx, step_result = future.result()
+                        _, step_result = future.result()
                         indexed_results.append((idx, step_result))
                         if verbose:
                             print(f"  ✓ Item {idx + 1}/{num_items} complete")
                     except Exception as e:
-                        logger.error(f"Parallel loop iteration failed: {e}")
+                        logger.error(f"Parallel loop iteration {idx} failed: {e}")
                         indexed_results.append((idx, {"step": f"loop_{idx}", "output": f"Error: {e}"}))
 
                 # Sort by index to maintain order