"thread" support for parallel workloads

Author: evalstate

examples/mcp_basic_agent/mcp_agent.config.yaml

@@ -4,7 +4,7 @@ execution_engine: asyncio
 logger:
   type: console
   level: info
-  path: "./mcp-agent.jsonl"
+  path: "./mcp-agent-4o-mini.jsonl"
   batch_size: 100
   flush_interval: 2
   max_queue_size: 2048
@@ -23,4 +23,5 @@ mcp:
 
 openai:
   # Secrets (API keys, etc.) are stored in an mcp_agent.secrets.yaml file which can be gitignored
-  default_model: o3-mini
+  #  default_model: "o3-mini"
+  default_model: "gpt-4o-mini"

examples/workflow_parallel/main.py

@@ -1,5 +1,4 @@
 import asyncio
-import os
 
 from mcp_agent.app import MCPApp
 from mcp_agent.agents.agent import Agent
@@ -83,7 +82,6 @@ async def example_usage():
 
         result = await parallel.generate_str(
             message=f"Student short story submission: {SHORT_STORY}",
-            request_params=RequestParams(model="gpt-4o"),
         )
 
         logger.info(f"{result}")

examples/workflow_parallel/mcp_agent.config.yaml

@@ -5,6 +5,7 @@ execution_engine: asyncio
 logger:
   type: console
   level: info
+  path: "./mcp-agent-o3-mini.jsonl"
   batch_size: 100
   flush_interval: 2
   max_queue_size: 2048
@@ -14,4 +15,4 @@ logger:
 
 openai:
   # Secrets (API keys, etc.) are stored in an mcp_agent.secrets.yaml file which can be gitignored
-  default_model: gpt-4o-mini
+  default_model: "gpt-4o-mini"

examples/workflow_router/main.py

@@ -82,7 +82,7 @@ async def example_usage():
             logger.info("Tools available:", data=result.model_dump())
 
             result = await agent.call_tool(
-                name="read_file",
+                tool_name="read_file",
                 arguments={
                     "path": str(os.path.join(os.getcwd(), "mcp_agent.config.yaml"))
                 },

examples/workflow_router/mcp_agent.config.yaml

@@ -28,4 +28,4 @@ mcp:
 
 openai:
   # Secrets (API keys, etc.) are stored in an mcp_agent.secrets.yaml file which can be gitignored
-  default_model: gpt-4o
+  default_model: "gpt-4o-mini"

scripts/event_replay.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+"""Event Replay Script
+
+Replays events from a JSONL log file using rich_progress display.
+"""
+
+import json
+import time
+from pathlib import Path
+
+import typer
+from mcp_agent.event_progress import convert_log_event
+from mcp_agent.logging.rich_progress import RichProgressDisplay
+
+
+def load_events(path: Path) -> list:
+    """Load events from JSONL file."""
+    events = []
+    with open(path) as f:
+        for line in f:
+            if line.strip():
+                events.append(json.loads(line))
+    return events
+
+
+def main(log_file: str):
+    """Replay MCP Agent events from a log file with progress display."""
+    # Load events from file
+    events = load_events(Path(log_file))
+
+    # Initialize progress display
+    progress = RichProgressDisplay()
+    progress.start()
+
+    try:
+        # Process each event in sequence
+        for event in events:
+            progress_event = convert_log_event(event)
+            if progress_event:
+                progress.update(progress_event)
+                # Add a small delay to make the replay visible
+                time.sleep(0.3)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        progress.stop()
+
+
+if __name__ == "__main__":
+    typer.run(main)

scripts/sample_logs/basic-agent-4o-mini.jsonl

@@ -39,7 +39,7 @@ class Agent(MCPAggregator):
 
     def __init__(
         self,
-        name: str,
+        name: str,  # agent name
         instruction: str | Callable[[Dict], str] = "You are a helpful agent.",
         server_names: List[str] = None,
         functions: List[Callable] = None,
@@ -52,6 +52,7 @@ def __init__(
             context=context,
             server_names=server_names or [],
             connection_persistence=connection_persistence,
+            name=name,
             **kwargs,
         )
 
@@ -188,6 +189,7 @@ async def list_tools(self) -> ListToolsResult:
 
         return result
 
+    # todo would prefer to use tool_name to disambiguate agent name
     async def call_tool(
         self, name: str, arguments: dict | None = None
     ) -> CallToolResult:

scripts/sample_logs/basic-agent-o3-mini.jsonl

@@ -2,7 +2,7 @@
 
 from dataclasses import dataclass
 from enum import Enum
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Union
 from mcp_agent.workflows.llm.llm_constants import FINAL_RESPONSE_LOG_MESSAGE
 
 
@@ -51,18 +51,30 @@ def convert_log_event(event: Dict[str, Any]) -> Optional[ProgressEvent]:
     """Convert a log event to a progress event if applicable."""
     namespace = event.get("namespace", "")
     message = event.get("message", "")
+
     # Extract agent name from namespace if present
     agent_name = None
-    parts = namespace.split(".")
-    if len(parts) > 3:
-        if parts[0:3] == ["mcp_agent", "agents", "agent"]:
-            agent_name = parts[3]
-        elif (
-            len(parts) > 4
-            and parts[0:3] == ["mcp_agent", "workflows", "llm"]
-            and parts[3].startswith("augmented_llm")
-        ):
-            agent_name = parts[4]
+
+    # Known class namespaces that may have agent names
+    CLASS_NAMESPACES = [
+        "mcp_agent.agents.agent",  # TODO: Use Agent.__module__
+        ("mcp_agent.workflows.llm", "augmented_llm"),  # Matches augmented_llm_* classes
+        "mcp_agent.mcp.mcp_aggregator",  # TODO: Use Finder.__module__
+    ]
+
+    # Check if namespace starts with any of our class prefixes and has an additional part
+    for class_ns in CLASS_NAMESPACES:
+        if isinstance(class_ns, tuple):
+            # Special case for augmented_llm_* classes
+            base_ns, class_prefix = class_ns
+            parts = namespace[len(base_ns) + 1 :].split(".")  # +1 for the dot
+            if len(parts) >= 2 and parts[0].startswith(class_prefix):
+                agent_name = parts[1]
+                break
+        elif namespace.startswith(class_ns + "."):
+            # Regular case - agent name is after the class namespace
+            agent_name = namespace.split(".")[-1]
+            break
 
     # Handle MCP connection events
     if "mcp_connection_manager" in namespace:
@@ -90,9 +102,10 @@ def convert_log_event(event: Dict[str, Any]) -> Optional[ProgressEvent]:
     # Handle MCPServerAggregator tool calls
     if "mcp_aggregator" in namespace:
         tool_name = extract_from_aggregator(message)
+        target = f"{agent_name} ({tool_name})" if agent_name else tool_name
         if tool_name:
             return ProgressEvent(
-                ProgressAction.CALLING_TOOL, tool_name, agent_name=agent_name
+                ProgressAction.CALLING_TOOL, target, agent_name=agent_name
             )
 
     # Handle LLM events

scripts/sample_logs/workflow_parallel-o3-mini.jsonl

@@ -13,7 +13,7 @@ class RichProgressDisplay:
     def __init__(self, console: Optional[Console] = None):
         """Initialize the progress display."""
         self.console = console or Console()
-        self._taskmap = {"default": None}
+        self._taskmap = {}
         self._progress = None
 
     def start(self):
@@ -27,9 +27,10 @@ def start(self):
             console=self.console,
             transient=False,
         ) as self._progress:
-            self._taskmap["default"] = self._progress.add_task(
+            task_id = self._progress.add_task(
                 "mcp-agent......", total=None, target="", details=""
             )
+            self._taskmap["default"] = task_id
 
         self._progress.start()
 

src/mcp_agent/agents/agent.py

@@ -22,7 +22,7 @@
     from mcp_agent.context import Context
 
 
-logger = get_logger(__name__)
+logger = get_logger(__name__)  # This will be replaced per-instance when agent_name is available
 
 SEP = "-"
 
@@ -77,6 +77,7 @@ def __init__(
         server_names: List[str],
         connection_persistence: bool = False,
         context: Optional["Context"] = None,
+        name: str = None,
         **kwargs,
     ):
         """
@@ -90,9 +91,14 @@ def __init__(
 
         self.server_names = server_names
         self.connection_persistence = connection_persistence
-
+        self.agent_name = name
         self._persistent_connection_manager: MCPConnectionManager = None
 
+        # Set up logger with agent name in namespace if available
+        global logger
+        logger_name = f"{__name__}.{name}" if name else __name__
+        logger = get_logger(logger_name)
+
         # Maps namespaced_tool_name -> namespaced tool info
         self._namespaced_tool_map: Dict[str, NamespacedTool] = {}
         # Maps server_name -> list of tools

src/mcp_agent/event_progress.py

@@ -61,26 +61,24 @@ def __init__(self, *args, **kwargs):
             intelligencePriority=0.3,
         )
         # Get default model from config if available
-        default_model = "gpt-4o"  # Fallback default
+        chosen_model = "gpt-4o"  # Fallback default
+
         self._reasoning_effort = "medium"
         if self.context and self.context.config and self.context.config.openai:
             if hasattr(self.context.config.openai, "default_model"):
-                default_model = self.context.config.openai.default_model
+                chosen_model = self.context.config.openai.default_model
             if hasattr(self.context.config.openai, "reasoning_effort"):
                 self._reasoning_effort = self.context.config.openai.reasoning_effort
 
         # o1 does not have tool support
-        self._reasoning = default_model.startswith("o3")
+        self._reasoning = chosen_model.startswith("o3")
         if self._reasoning:
             self.logger.info(
-                f"Using reasoning model '{default_model}' with '{self._reasoning_effort}' reasoning effort"
-            )
-            print(
-                f"\nUsing reasoning model [white on dark_blue]{default_model}[/white on dark_blue] with [white on dark_green]{self._reasoning_effort}[/white on dark_green] reasoning effort"
+                f"Using reasoning model '{chosen_model}' with '{self._reasoning_effort}' reasoning effort"
             )
 
         self.default_request_params = self.default_request_params or RequestParams(
-            model=default_model,
+            model=chosen_model,
             modelPreferences=self.model_preferences,
             maxTokens=4096,
             systemPrompt=self.instruction,