feat: implement multi-agent MCP system with multiple server instances

- Created `multi_server_test.py` to test coordination across agents and servers
- Added `calc_server.py` for handling computation requests
- Referenced swarms-rs Rust architecture for Python-based design structure

.replit

@@ -4,7 +4,7 @@ modules = ["python-3.10", "bash"]
 channel = "stable-24_05"
 
 [workflows]
-runButton = "Run Multiple MCP"
+runButton = "Run Multi Server"
 
 [[workflows.workflow]]
 name = "Run MCP Tests"
@@ -75,3 +75,24 @@ args = "sleep 2"
 [[workflows.workflow.tasks]]
 task = "shell.exec"
 args = "python examples/mcp_example/test_integration.py"
+
+[[workflows.workflow]]
+name = "Run Multi Server"
+author = 13983571
+mode = "sequential"
+
+[[workflows.workflow.tasks]]
+task = "shell.exec"
+args = "python examples/mcp_example/math_server.py &"
+
+[[workflows.workflow.tasks]]
+task = "shell.exec"
+args = "python examples/mcp_example/calc_server.py &"
+
+[[workflows.workflow.tasks]]
+task = "shell.exec"
+args = "sleep 2"
+
+[[workflows.workflow.tasks]]
+task = "shell.exec"
+args = "python examples/mcp_example/multi_server_test.py"

examples/mcp_example/calc_server.py

@@ -1,36 +1,33 @@
 
 from fastmcp import FastMCP
-from typing import Dict, Any
+from litellm import LiteLLM
+import logging
 
-# Initialize MCP server for business calculations
+# Configure logging
+logging.basicConfig(level=logging.ERROR)
+logger = logging.getLogger(__name__)
+
+# Initialize MCP server for financial calculations
 mcp = FastMCP("Calc-Server")
 
-@mcp.tool()
+@mcp.tool(name="compound_interest", description="Calculate compound interest")
-def profit_margin(revenue: float, cost: float) -> Dict[str, Any]:
+def compound_interest(principal: float, rate: float, time: float) -> float:
-    """Calculate profit margin from revenue and cost"""
     try:
-        profit = revenue - cost
+        result = principal * (1 + rate/100) ** time
-        margin = (profit / revenue) * 100
+        return round(result, 2)
-        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)}
+        logger.error(f"Error calculating compound interest: {e}")
+        raise
 
-@mcp.tool()
+@mcp.tool(name="percentage", description="Calculate percentage")
-def break_even_point(fixed_costs: float, price_per_unit: float, cost_per_unit: float) -> Dict[str, Any]:
+def percentage(value: float, percent: float) -> float:
-    """Calculate break-even point"""
     try:
-        bep = fixed_costs / (price_per_unit - cost_per_unit)
+        return (value * percent) / 100
-        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)}
+        logger.error(f"Error calculating percentage: {e}")
+        raise
 
 if __name__ == "__main__":
-    print("Starting Business Calculator Server on port 6275...")
+    print("Starting Calculation Server on port 6275...")
-    mcp.run(transport="sse", transport_kwargs={"host": "0.0.0.0", "port": 6275})
+    llm = LiteLLM()
+    mcp.run(transport="sse", host="0.0.0.0", port=6275)

examples/mcp_example/multi_server_test.py

@@ -0,0 +1,82 @@
+
+from swarms import Agent
+from swarms.tools.mcp_integration import MCPServerSseParams
+from swarms.prompts.finance_agent_sys_prompt import FINANCIAL_AGENT_SYS_PROMPT
+
+# Configure multiple MCP servers
+math_server = MCPServerSseParams(
+    url="http://0.0.0.0:6274",
+    headers={"Content-Type": "application/json"},
+    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,
+    sse_read_timeout=300.0
+)
+
+# Create specialized agents with different server access
+math_agent = Agent(
+    agent_name="Math-Specialist",
+    agent_description="Advanced mathematics expert",
+    system_prompt="You are a mathematics expert. Use available math operations.",
+    max_loops=1,
+    mcp_servers=[math_server],
+    interactive=True,
+    streaming_on=True
+)
+
+finance_agent = Agent(
+    agent_name="Finance-Specialist",
+    agent_description="Financial calculations expert", 
+    system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
+    max_loops=1,
+    mcp_servers=[calc_server],
+    interactive=True,
+    streaming_on=True
+)
+
+# Multi-server agent with access to all operations
+super_agent = Agent(
+    agent_name="Super-Calculator",
+    agent_description="Multi-capable calculation expert",
+    system_prompt="You have access to multiple calculation servers. Use them appropriately.",
+    max_loops=1,
+    mcp_servers=[math_server, calc_server],
+    interactive=True,
+    streaming_on=True
+)
+
+def main():
+    print("\nMulti-Agent MCP Test Environment")
+    print("Type 'exit' to quit\n")
+
+    while True:
+        try:
+            user_input = input("\nEnter your calculation request: ")
+            
+            if user_input.lower() == 'exit':
+                break
+
+            # Route request to appropriate agent based on keywords
+            if 'finance' in user_input.lower():
+                response = finance_agent.run(user_input)
+                print(f"\nFinance Agent Response: {response}")
+            elif 'math' in user_input.lower():
+                response = math_agent.run(user_input)
+                print(f"\nMath Agent Response: {response}")
+            else:
+                response = super_agent.run(user_input)
+                print(f"\nSuper Agent Response: {response}")
+
+        except KeyboardInterrupt:
+            print("\nExiting gracefully...")
+            break
+        except Exception as e:
+            print(f"Error processing request: {e}")
+
+if __name__ == "__main__":
+    main()