test: rewrite MCP integration tests from user perspective

tests/tools/test_mcp_integration.py

@@ -1,75 +1,136 @@
 
 import pytest
-from swarms.tools.mcp_integration import MCPServerSseParams
+import asyncio
-from swarms import Agent
+from swarms.tools.mcp_integration import (
-from swarms.prompts.finance_agent_sys_prompt import FINANCIAL_AGENT_SYS_PROMPT
+    MCPServer,
-
+    MCPServerStdio,
-def test_interactive_multi_agent_mcp():
+    MCPServerSse,
-    # Configure two MCP servers
+    mcp_flow,
-    server_one = MCPServerSseParams(
+    mcp_flow_get_tool_schema,
-        url="http://0.0.0.0:6274",
+    batch_mcp_flow
-        headers={"Content-Type": "application/json"}
+)
-    )
+
-    
+# Test basic server connectivity
-    server_two = MCPServerSseParams(
+def test_server_connection():
-        url="http://0.0.0.0:6275", 
+    """
-        headers={"Content-Type": "application/json"}
+    Test that a user can connect to the MCP server successfully
-    )
+    """
-
+    params = {"url": "http://localhost:8000"}
-    # Create two agents with different roles
+    server = MCPServerSse(params, cache_tools_list=True)
-    finance_agent = Agent(
+    
-        agent_name="Finance-Agent",
+    # Connect should work
-        agent_description="Financial analysis expert",
+    asyncio.run(server.connect())
-        system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
+    assert server.session is not None
-        max_loops=1,
+    
-        mcp_servers=[server_one],
+    # Cleanup should work
-        interactive=True,
+    asyncio.run(server.cleanup())
-        streaming_on=True
+    assert server.session is None
-    )
+
-
+# Test tool listing functionality
-    research_agent = Agent(
+def test_list_tools():
-        agent_name="Research-Agent", 
+    """
-        agent_description="Market research specialist",
+    Test that a user can retrieve available tools from the server
-        system_prompt="You are a market research specialist. Analyze market trends and provide insights.",
+    """
-        max_loops=1,
+    params = {"url": "http://localhost:8000"}
-        mcp_servers=[server_two],
+    server = MCPServerSse(params)
-        interactive=True,
+    
-        streaming_on=True
+    asyncio.run(server.connect())
-    )
+    tools = asyncio.run(server.list_tools())
-
+    
-    try:
+    assert isinstance(tools, list)
-        # Interactive loop
+    assert len(tools) > 0
-        while True:
+    
-            # Get user input for which agent to use
+    asyncio.run(server.cleanup())
-            print("\nWhich agent would you like to interact with?")
+
-            print("1. Finance Agent")
+# Test tool execution
-            print("2. Research Agent")
+def test_tool_execution():
-            print("3. Exit")
+    """
-            
+    Test that a user can execute a tool successfully
-            choice = input("Enter your choice (1-3): ")
+    """
-            
+    params = {"url": "http://localhost:8000"}
-            if choice == "3":
+    function_call = {
-                break
+        "tool_name": "add",
-                
+        "arguments": {"a": 5, "b": 3}
-            # Get the task from user
+    }
-            task = input("\nEnter your task for the agent: ")
+    
-            
+    result = mcp_flow(params, function_call)
-            # Route to appropriate agent
+    assert result is not None
-            if choice == "1":
+
-                response = finance_agent.run(task)
+# Test batch operations
-                print(f"\nFinance Agent Response:\n{response}")
+def test_batch_execution():
-            elif choice == "2":
+    """
-                response = research_agent.run(task)
+    Test that a user can execute multiple tools in batch
-                print(f"\nResearch Agent Response:\n{response}")
+    """
-            else:
+    params_list = [
-                print("Invalid choice, please try again")
+        {"url": "http://localhost:8000"},
-
+        {"url": "http://localhost:8000"}
-    except Exception as e:
+    ]
-        pytest.fail(f"Interactive multi-agent test failed: {e}")
+    function_calls = [
-
+        {"tool_name": "add", "arguments": {"a": 1, "b": 2}},
-def test_mcp_invalid_params():
+        {"tool_name": "subtract", "arguments": {"a": 5, "b": 3}}
+    ]
+    
+    results = batch_mcp_flow(params_list, function_calls)
+    assert len(results) == 2
+    assert all(result is not None for result in results)
+
+# Test error handling
+def test_error_handling():
+    """
+    Test that users receive proper error messages for invalid operations
+    """
+    params = {"url": "http://localhost:8000"}
+    invalid_function = {
+        "tool_name": "nonexistent_tool",
+        "arguments": {}
+    }
+    
     with pytest.raises(Exception):
-        mcp_flow(None, {})
+        mcp_flow(params, invalid_function)
+
+# Test tool schema retrieval
+def test_get_tool_schema():
+    """
+    Test that users can retrieve tool schemas correctly
+    """
+    params = {"url": "http://localhost:8000"}
+    schema = mcp_flow_get_tool_schema(params)
+    
+    assert isinstance(schema, dict)
+    assert "tools" in schema or "functions" in schema
+
+# Test server reconnection
+def test_server_reconnection():
+    """
+    Test that users can reconnect to the server after disconnection
+    """
+    params = {"url": "http://localhost:8000"}
+    server = MCPServerSse(params)
+    
+    # First connection
+    asyncio.run(server.connect())
+    asyncio.run(server.cleanup())
+    
+    # Second connection should work
+    asyncio.run(server.connect())
+    assert server.session is not None
+    asyncio.run(server.cleanup())
+
+# Test cache functionality
+def test_cache_behavior():
+    """
+    Test that tool caching works as expected for users
+    """
+    params = {"url": "http://localhost:8000"}
+    server = MCPServerSse(params, cache_tools_list=True)
+    
+    asyncio.run(server.connect())
+    
+    # First call should cache
+    tools1 = asyncio.run(server.list_tools())
+    # Second call should use cache
+    tools2 = asyncio.run(server.list_tools())
+    
+    assert tools1 == tools2
     
-if __name__ == "__main__":
+    asyncio.run(server.cleanup())
-    test_interactive_multi_agent_mcp()