feat: set up interactive multi-agent MCP test environment with multi-server coordination

.replit

@@ -4,7 +4,7 @@ modules = ["python-3.10", "bash"]
 channel = "stable-24_05"
 
 [workflows]
-runButton = "Run MCP Server"
+runButton = "Run Multiple MCP"
 
 [[workflows.workflow]]
 name = "Run MCP Tests"

examples/mcp_example/calc_server.py

@@ -1,27 +1,36 @@
 
 from fastmcp import FastMCP
 from typing import Dict, Any
-import math
 
-# Initialize MCP server
+# Initialize MCP server for business calculations
 mcp = FastMCP("Calc-Server")
 
 @mcp.tool()
-def square_root(x: float) -> float:
+def profit_margin(revenue: float, cost: float) -> Dict[str, Any]:
-    """Calculate square root of a number"""
+    """Calculate profit margin from revenue and cost"""
     try:
-        return math.sqrt(x)
+        profit = revenue - cost
+        margin = (profit / revenue) * 100
+        return {
+            "profit": profit,
+            "margin_percentage": margin,
+            "summary": f"On revenue of ${revenue:.2f} and costs of ${cost:.2f}, profit is ${profit:.2f} with a margin of {margin:.1f}%"
+        }
     except Exception as e:
         return {"error": str(e)}
 
 @mcp.tool()
-def power(base: float, exponent: float) -> float:
+def break_even_point(fixed_costs: float, price_per_unit: float, cost_per_unit: float) -> Dict[str, Any]:
-    """Raise a number to a power"""
+    """Calculate break-even point"""
     try:
-        return math.pow(base, exponent)
+        bep = fixed_costs / (price_per_unit - cost_per_unit)
+        return {
+            "break_even_units": bep,
+            "summary": f"You need to sell {bep:.0f} units to break even"
+        }
     except Exception as e:
         return {"error": str(e)}
 
 if __name__ == "__main__":
-    print("Starting Calc Server on port 6275...")
+    print("Starting Business Calculator Server on port 6275...")
-    mcp.run(transport="sse", host="0.0.0.0", port=6275)
+    mcp.run(transport="sse")

examples/mcp_example/math_server.py

@@ -2,33 +2,38 @@
 from fastmcp import FastMCP
 from typing import Dict, Any
 
-# Initialize MCP server
+# Initialize MCP server for math operations
 mcp = FastMCP("Math-Server")
 
 @mcp.tool()
-def add(a: int, b: int) -> int:
+def add(a: float, b: float) -> float:
-    """Add two numbers together"""
+    """Add two numbers"""
-    try:
+    return a + b
-        return a + b
-    except Exception as e:
-        return {"error": str(e)}
 
 @mcp.tool()
-def multiply(a: int, b: int) -> int:
+def subtract(a: float, b: float) -> float:
-    """Multiply two numbers together"""
+    """Subtract b from a"""
-    try:
+    return a - b
-        return a * b 
-    except Exception as e:
-        return {"error": str(e)}
 
 @mcp.tool()
-def subtract(a: int, b: int) -> int:
+def multiply(a: float, b: float) -> float:
-    """Subtract two numbers"""
+    """Multiply two numbers"""
-    try:
+    return a * b
-        return a - b
+
-    except Exception as e:
+@mcp.tool()
-        return {"error": str(e)}
+def divide(a: float, b: float) -> float:
+    """Divide a by b"""
+    if b == 0:
+        return {"error": "Cannot divide by zero"}
+    return a / b
+
+@mcp.tool()
+def calculate_percentage(part: float, whole: float) -> float:
+    """Calculate percentage"""
+    if whole == 0:
+        return {"error": "Cannot calculate percentage with zero total"}
+    return (part / whole) * 100
 
 if __name__ == "__main__":
     print("Starting Math Server on port 6274...")
-    mcp.run(transport="sse", host="0.0.0.0", port=6274)
+    mcp.run(transport="sse")

examples/mcp_example/test_integration.py

@@ -3,67 +3,82 @@ from swarms import Agent
 from swarms.prompts.finance_agent_sys_prompt import FINANCIAL_AGENT_SYS_PROMPT 
 from swarms.tools.mcp_integration import MCPServerSseParams
 import logging
+import time
+
+def setup_agent(name: str, description: str, servers: list) -> Agent:
+    """Setup an agent with MCP server connections"""
+    return Agent(
+        agent_name=name,
+        agent_description=description,
+        system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
+        max_loops=1,
+        mcp_servers=servers,
+        streaming_on=True
+    )
 
 def main():
-    # Configure multiple MCP server connections
+    # Configure MCP server connections
     math_server = MCPServerSseParams(
         url="http://0.0.0.0:6274",
         headers={"Content-Type": "application/json"},
-        timeout=10.0,
+        timeout=10.0
-        sse_read_timeout=300.0
     )
     
     calc_server = MCPServerSseParams(
         url="http://0.0.0.0:6275",
         headers={"Content-Type": "application/json"},
-        timeout=10.0,
+        timeout=10.0
-        sse_read_timeout=300.0
     )
 
-    # Initialize multiple agents with different MCP capabilities
+    # Initialize specialized agents
-    math_agent = Agent(
+    coordinator = setup_agent(
-        agent_name="Math-Agent",
+        "Coordinator",
-        agent_description="Agent that performs math operations",
+        "Analyzes tasks and coordinates between specialized agents",
-        system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
+        [math_server, calc_server]
-        max_loops=1,
-        mcp_servers=[math_server],
-        streaming_on=True
     )
     
-    calc_agent = Agent(
+    math_agent = setup_agent(
-        agent_name="Calc-Agent", 
+        "Math-Agent",
-        agent_description="Agent that performs calculations",
+        "Handles mathematical calculations",
-        system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
+        [math_server]
-        max_loops=1,
-        mcp_servers=[calc_server],
-        streaming_on=True
     )
     
-    agents = [math_agent, calc_agent]
+    business_agent = setup_agent(
+        "Business-Agent",
+        "Handles business calculations",
+        [calc_server]
+    )
+
+    print("\nMulti-Agent MCP Test Environment")
+    print("Type 'exit' to quit\n")
+    
+    while True:
+        try:
+            user_input = input("\nEnter your request: ")
             
-    try:
+            if user_input.lower() == 'exit':
-        # Test each agent
+                break
-        for agent in agents:
-            print(f"\nTesting {agent.agent_name}...")
-            print("\nDiscovering available tools from MCP server...")
-            tools = agent.mcp_tool_handling()
-            print(f"\nAvailable tools for {agent.agent_name}:", tools)
                 
-            while True:
+            # Coordinator analyzes task
-                user_input = input(f"\nEnter a math operation for {agent.agent_name} (or 'exit' to switch agent): ")
+            print("\nCoordinator analyzing task...")
+            coordinator_response = coordinator.run(
+                f"Analyze this task and determine required calculations: {user_input}"
+            )
+            print(f"\nCoordinator's plan: {coordinator_response}")
             
-                if user_input.lower() == 'exit':
+            # Route to appropriate agent(s)
-                    break
+            if "profit" in user_input.lower() or "margin" in user_input.lower():
+                print("\nRouting to Business Agent...")
+                result = business_agent.run(user_input)
+                print(f"\nBusiness calculation result: {result}")
             
-                try:
+            if any(op in user_input.lower() for op in ['add', 'subtract', 'multiply', 'divide']):
-                    result = agent.run(user_input)
+                print("\nRouting to Math Agent...")
-                    print(f"\nResult from {agent.agent_name}:", result)
+                result = math_agent.run(user_input)
-                except Exception as e:
+                print(f"\nMath calculation result: {result}")
-                    print(f"Error processing request: {e}")
 
-    except Exception as e:
+        except Exception as e:
-        logging.error(f"Test failed: {e}")
+            print(f"Error processing request: {e}")
-        raise
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.INFO)