diff --git a/examples/mcp_example/math_server.py b/examples/mcp_example/math_server.py
index 08be7a00..af533c90 100644
--- a/examples/mcp_example/math_server.py
+++ b/examples/mcp_example/math_server.py
@@ -1,77 +1,25 @@
-
from fastmcp import FastMCP
from typing import Dict, Any, Optional
# Initialize MCP server
mcp = FastMCP("Math-Server")
-# Add tool documentation and type hints
@mcp.tool()
-def add(a: int, b: int) -> Dict[str, Any]:
- """Add two numbers together
-
- Args:
- a (int): First number to add
- b (int): Second number to add
-
- Returns:
- Dict[str, Any]: Result dictionary containing sum and metadata
- """
+def add(a: int, b: int) -> int:
+ """Add two numbers together"""
try:
- result = a + b
- return {
- "status": "success",
- "result": result,
- "message": f"Successfully added {a} and {b}"
- }
+ return a + b
except Exception as e:
- return {
- "status": "error",
- "error": str(e),
- "message": "Failed to perform addition"
- }
+ return {"error": str(e)}
@mcp.tool()
-def multiply(a: int, b: int) -> Dict[str, Any]:
- """Multiply two numbers together
-
- Args:
- a (int): First number to multiply
- b (int): Second number to multiply
-
- Returns:
- Dict[str, Any]: Result dictionary containing product and metadata
- """
+def multiply(a: int, b: int) -> int:
+ """Multiply two numbers together"""
try:
- result = a * b
- return {
- "status": "success",
- "result": result,
- "message": f"Successfully multiplied {a} and {b}"
- }
+ return a * b
except Exception as e:
- return {
- "status": "error",
- "error": str(e),
- "message": "Failed to perform multiplication"
- }
-
-@mcp.tool()
-def get_available_operations() -> Dict[str, Any]:
- """Get list of available mathematical operations
-
- Returns:
- Dict[str, Any]: Dictionary containing available operations and their descriptions
- """
- return {
- "status": "success",
- "operations": {
- "add": "Add two numbers together",
- "multiply": "Multiply two numbers together"
- }
- }
+ return {"error": str(e)}
if __name__ == "__main__":
print("Starting Math Server...")
- print("Available operations:", get_available_operations())
- mcp.run(host="0.0.0.0", port=6274, transport="sse")
+ mcp.run(transport="sse")
\ No newline at end of file
diff --git a/examples/mcp_example/test_integration.py b/examples/mcp_example/test_integration.py
index b9dbd74e..192542ae 100644
--- a/examples/mcp_example/test_integration.py
+++ b/examples/mcp_example/test_integration.py
@@ -1,66 +1,43 @@
-
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
-from typing import Dict, Any, Optional
-def handle_mcp_response(response: Dict[str, Any]) -> str:
- """Handle MCP response and extract meaningful output"""
- if response.get("status") == "error":
- return f"Error: {response.get('message', 'Unknown error occurred')}"
- return str(response.get("result", response))
-
-def setup_mcp_agent(name: str, description: str) -> Agent:
- """Setup an MCP-enabled agent with proper configuration"""
- try:
- server = MCPServerSseParams(
- url="http://0.0.0.0:6274",
- headers={"Content-Type": "application/json"},
- timeout=10.0,
- sse_read_timeout=300.0
- )
-
- return Agent(
- agent_name=name,
- agent_description=description,
- system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
- max_loops=1,
- mcp_servers=[server],
- streaming_on=True
- )
- except Exception as e:
- logging.error(f"Failed to setup agent: {e}")
- raise
+# Configure MCP server connection
+server = MCPServerSseParams(
+ url="http://0.0.0.0:6274",
+ headers={"Content-Type": "application/json"},
+ timeout=10.0,
+ sse_read_timeout=300.0
+)
+
+# Initialize agent with MCP capabilities
+agent = Agent(
+ agent_name="Math-Agent",
+ agent_description="Agent that performs math operations",
+ system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
+ max_loops=1,
+ mcp_servers=[server],
+ streaming_on=True
+)
def test_mcp_operations():
"""Test basic MCP operations with error handling"""
try:
- # Initialize agent
- agent = setup_mcp_agent(
- "Math-Agent",
- "Agent that performs math operations"
- )
-
- # Get available operations
- print("\nQuerying available operations...")
- result = agent.run("What operations are available?")
- print("Available operations:", result)
-
# Test addition
print("\nTesting addition...")
add_result = agent.run("Use the add tool to add 5 and 3")
- print("Addition result:", handle_mcp_response(add_result))
+ print("Addition result:", add_result)
# Test multiplication
print("\nTesting multiplication...")
- mult_result = agent.run("Use the multiply tool to multiply 4 and 6")
- print("Multiplication result:", handle_mcp_response(mult_result))
+ mult_result = agent.run("Use the multiply tool to multiply 4 and 6")
+ print("Multiplication result:", mult_result)
# Test error case
print("\nTesting error handling...")
error_result = agent.run("Use the add tool with invalid inputs")
- print("Error handling result:", handle_mcp_response(error_result))
+ print("Error handling result:", error_result)
except Exception as e:
logging.error(f"Test failed: {e}")
@@ -68,4 +45,4 @@ def test_mcp_operations():
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
- test_mcp_operations()
+ test_mcp_operations()
\ No newline at end of file