ai-mirow-projects
/
swarms
mirror of https://github.com/kyegomez/swarms
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor(math): simplify math operations and improve logging in mcp_client.py and mock_math_server.py

Browse Source
pull/819/head
DP37 3 months ago committed by ascender1729
parent 8fd41d24aa
commit 812115a2a6
3 changed files with 174 additions and 332 deletions
Show Stats Download Patch File Download Diff File

66
attached_assets/Pasted--MINIMAL-MCP-AGENT-INTEGRATION-TEST-Testing-only-the-core-MCP-integration-with-Agent-2025-0-1745173265725.txt
Unescape View File

@ -0,0 +1,66 @@
=== MINIMAL MCP AGENT INTEGRATION TEST ===
Testing only the core MCP integration with Agent
2025-04-20T18:19:47.080439+0000 | INFO | mcp_client:main:33 - Creating MCP server parameters...
2025-04-20T18:19:47.080601+0000 | INFO | mcp_client:main:45 - MCP Server URL: http://0.0.0.0:8000
2025-04-20T18:19:47.080686+0000 | INFO | mcp_client:main:46 - MCP Headers: {'Content-Type': 'application/json', 'Accept': 'text/event-stream'}
2025-04-20T18:19:47.080763+0000 | INFO | mcp_client:main:49 - Creating Agent with MCP integration...
2025-04-20T18:19:47.088927+0000 | WARNING | agent:llm_handling:613 - Model name is not provided, using gpt-4o-mini. You can configure any model from litellm if desired.
Agent created successfully!
Enter a math query or 'exit' to quit
Math query: what tools you have?
2025-04-20T18:20:39.747943+0000 | INFO | mcp_client:main:66 - Processing query: what tools you have?
╭────────────────── Agent Name MCP Test Agent [Max Loops: 1 ] ───────────────────╮
│ MCP Test Agent: {"tool_name": "none", "description": "I can perform addition, │
│ multiplication, and division using specific tools."} │
╰────────────────────────────────────────────────────────────────────────────────╯
2025-04-20T18:20:41.073474+0000 | ERROR | mcp_integration:connect:89 - Error initializing MCP server: unhandled errors in a TaskGroup (1 sub-exception)
2025-04-20T18:20:41.073628+0000 | ERROR | mcp_integration:abatch_mcp_flow:268 - Error in abatch_mcp_flow: unhandled errors in a TaskGroup (1 sub-exception)
╭────────── Agent Name MCP Test Agent - Tool Executor [Max Loops: 1 ] ───────────╮
│ MCP Test Agent - Tool Executor: Error in batch operation: unhandled errors in │
│ a TaskGroup (1 sub-exception) │
╰────────────────────────────────────────────────────────────────────────────────╯
╭────────── Agent Name MCP Test Agent - Agent Analysis [Max Loops: 1 ] ──────────╮
│ MCP Test Agent - Agent Analysis: It seems like you're encountering an error │
│ related to a batch operation in a programming or data processing context. │
│ However, I can assist you with mathematical calculations. If you have any │
│ specific calculations you'd like to perform, please provide the numbers and │
│ the operation (addition, multiplication, or division), and I'll format it as │
│ JSON for you. │
╰────────────────────────────────────────────────────────────────────────────────╯
Result: System: : Your Name: MCP Test Agent
Your Description: None
You are a math calculator assistant that uses tools to perform calculations.
When asked for calculations, determine the operation and numbers, then use one of these tools:
- add: Add two numbers
- multiply: Multiply two numbers
- divide: Divide first number by second
FORMAT as JSON:
{"tool_name": "add", "a": 5, "b": 10}
Human:: what tools you have?
MCP Test Agent: {"tool_name": "none", "description": "I can perform addition, multiplication, and division using specific tools."}
assistant: Tool execution result: ['Error in batch operation: unhandled errors in a TaskGroup (1 sub-exception)']
Tool Executor: Error in batch operation: unhandled errors in a TaskGroup (1 sub-exception)
MCP Test Agent: It seems like you're encountering an error related to a batch operation in a programming or data processing context. However, I can assist you with mathematical calculations. If you have any specific calculations you'd like to perform, please provide the numbers and the operation (addition, multiplication, or division), and I'll format it as JSON for you.
Math query:

231
examples/mcp_example/mcp_client.py
Unescape View File

@ -1,210 +1,77 @@
from swarms import Agent
from swarms.tools.mcp_integration import MCPServerSseParams, MCPServerSse, mcp_flow_get_tool_schema
from swarms.tools.mcp_integration import MCPServerSseParams
from loguru import logger
import sys
import asyncio
import json
import httpx
import time
# Configure logging for more detailed output
# Configure logging
logger.remove()
logger.add(sys.stdout,
level="DEBUG",
level="INFO",
format="{time} | {level} | {module}:{function}:{line} - {message}")
# Relaxed prompt that doesn't enforce strict JSON formatting
# Math prompt for testing MCP integration
MATH_PROMPT = """
You are a math calculator assistant that uses tools to perform calculations.
When asked for calculations, determine the operation and numbers, then use one of these tools:
- add: Add two numbers
- multiply: Multiply two numbers
- divide: Divide first number by second
# Create server parameters
def get_server_params():
"""Get the MCP server connection parameters."""
return MCPServerSseParams(
url="http://0.0.0.0:8000", # Use 0.0.0.0 to be accessible
headers={
"Content-Type": "application/json",
"Accept": "text/event-stream"
},
timeout=15.0, # Longer timeout
sse_read_timeout=60.0 # Longer read timeout
)
def initialize_math_system():
"""Initialize the math agent with MCP server configuration."""
# Create the agent with the MCP server configuration
math_agent = Agent(agent_name="Math Assistant",
agent_description="Friendly math calculator",
system_prompt=MATH_AGENT_PROMPT,
max_loops=1,
mcp_servers=[get_server_params()],
model_name="gpt-3.5-turbo",
verbose=True)
return math_agent
# Function to get list of available tools from the server
async def get_tools_list():
"""Fetch and format the list of available tools from the server."""
try:
server_params = get_server_params()
tools = await mcp_flow_get_tool_schema(server_params)
if not tools:
return "No tools are currently available on the server."
# Format the tools information
tools_info = "Available tools:\n"
for tool in tools:
tools_info += f"\n- {tool.name}: {tool.description or 'No description'}\n"
if tool.parameters and hasattr(tool.parameters, 'properties'):
tools_info += " Parameters:\n"
for param_name, param_info in tool.parameters.properties.items(
):
param_type = param_info.get('type', 'unknown')
param_desc = param_info.get('description',
'No description')
tools_info += f" - {param_name} ({param_type}): {param_desc}\n"
return tools_info
except Exception as e:
logger.error(f"Failed to get tools list: {e}")
return f"Error retrieving tools list: {str(e)}"
# Function to test server connection
def test_server_connection():
"""Test if the server is reachable and responsive."""
try:
params = get_server_params()
health_url = f"{params.url}{params.sse_path}"
response = httpx.get(
health_url,
headers={"Accept": "text/event-stream"},
timeout=5.0
)
if response.status_code == 200:
logger.info("✅ SSE endpoint is up")
return True
else:
logger.error(f"❌ Unexpected status {response.status_code}")
return False
except Exception as e:
logger.error(f"❌ Connection to SSE endpoint failed: {e}")
return False
# Manual math operation handler as ultimate fallback
def manual_math(query):
"""Parse and solve a math problem without using the server."""
query = query.lower()
# Check if user is asking for available tools/functions
if "list" in query and ("tools" in query or "functions" in query
or "operations" in query):
return """
Available tools:
1. add - Add two numbers together (e.g., "add 3 and 4")
2. multiply - Multiply two numbers together (e.g., "multiply 5 and 6")
3. divide - Divide the first number by the second (e.g., "divide 10 by 2")
FORMAT as JSON:
{"tool_name": "add", "a": 5, "b": 10}
"""
try:
if "add" in query or "plus" in query or "sum" in query:
# Extract numbers using a simple approach
numbers = [int(s) for s in query.split() if s.isdigit()]
if len(numbers) >= 2:
result = numbers[0] + numbers[1]
return f"The sum of {numbers[0]} and {numbers[1]} is {result}"
elif "multiply" in query or "times" in query or "product" in query:
numbers = [int(s) for s in query.split() if s.isdigit()]
if len(numbers) >= 2:
result = numbers[0] * numbers[1]
return f"The product of {numbers[0]} and {numbers[1]} is {result}"
elif "divide" in query or "quotient" in query:
numbers = [int(s) for s in query.split() if s.isdigit()]
if len(numbers) >= 2:
if numbers[1] == 0:
return "Cannot divide by zero"
result = numbers[0] / numbers[1]
return f"{numbers[0]} divided by {numbers[1]} is {result}"
return "I couldn't parse your math request. Try something like 'add 3 and 4'."
except Exception as e:
logger.error(f"Manual math error: {e}")
return f"Error performing calculation: {str(e)}"
def main():
try:
logger.info("Initializing math system...")
"""Main function to test MCP integration with Agent."""
print("=== MINIMAL MCP AGENT INTEGRATION TEST ===")
print("Testing only the core MCP integration with Agent")
# Test server connection first
server_available = test_server_connection()
if server_available:
math_agent = initialize_math_system()
print("\nMath Calculator Ready! (Server connection successful)")
else:
print(
"\nServer connection failed - using fallback calculator mode")
math_agent = None
try:
# Create the server parameters correctly
logger.info("Creating MCP server parameters...")
mcp_server = {
"url": "http://0.0.0.0:8000",
"headers": {
"Content-Type": "application/json",
"Accept": "text/event-stream"
},
"timeout": 10.0,
"sse_read_timeout": 30.0
}
# Log the server params to verify they're correct
logger.info(f"MCP Server URL: {mcp_server['url']}")
logger.info(f"MCP Headers: {mcp_server['headers']}")
# Create agent with minimal configuration
logger.info("Creating Agent with MCP integration...")
agent = Agent(
agent_name="MCP Test Agent",
system_prompt=MATH_PROMPT,
mcp_servers=[mcp_server], # Pass server config as a list of dicts
verbose=True)
print("Ask me any math question!")
print("Examples: 'what is 5 plus 3?' or 'can you multiply 4 and 6?'")
print("Type 'list tools' to see available operations")
print("Type 'exit' to quit\n")
print("\nAgent created successfully!")
print("Enter a math query or 'exit' to quit")
# Simple interaction loop
while True:
try:
query = input("What would you like to calculate? ").strip()
if not query:
continue
query = input("\nMath query: ").strip()
if query.lower() == 'exit':
break
# Handle special commands
if query.lower() in ('list tools', 'show tools',
'available tools', 'what tools'):
if server_available:
# Get tools list from server
tools_info = asyncio.run(get_tools_list())
print(f"\n{tools_info}\n")
else:
# Use manual fallback
print(manual_math("list tools"))
continue
# Run the agent, which should use the MCP server
logger.info(f"Processing query: {query}")
result = agent.run(query)
# First try the agent if available
if math_agent and server_available:
try:
result = math_agent.run(query)
print(f"\nResult: {result}\n")
continue
except Exception as e:
logger.error(f"Agent error: {e}")
print("Agent encountered an error, trying fallback...")
# If agent fails or isn't available, use manual calculator
result = manual_math(query)
print(f"\nCalculation result: {result}\n")
except KeyboardInterrupt:
print("\nGoodbye!")
break
except Exception as e:
logger.error(f"Error processing query: {e}")
print(f"Sorry, there was an error: {str(e)}")
# Display result
print(f"\nResult: {result}")
except Exception as e:
logger.error(f"System initialization error: {e}")
print(f"Failed to start the math system: {str(e)}")
logger.error(f"Error during MCP integration test: {e}", exc_info=True)
print(f"\nERROR: {type(e).__name__}: {str(e)}")
if __name__ == "__main__":

153
examples/mcp_example/mock_math_server.py
Unescape View File

@ -1,170 +1,79 @@
from fastmcp import FastMCP
from loguru import logger
import time
import json
# Create the MCP server with detailed debugging
# Create the MCP server with all interfaces binding
mcp = FastMCP(
host="0.0.0.0", # Bind to all interfaces
host=
"0.0.0.0", # Bind to all interfaces to be accessible from other contexts
port=8000,
transport="sse",
require_session_id=False,
cors_allowed_origins=["*"], # Allow connections from any origin
debug=True # Enable debug mode for more verbose output
cors_allowed_origins=["*"], # Allow all origins for testing
debug=True # Enable debug mode
)
# Add a more flexible parsing approach
def parse_input(input_str):
"""Parse input that could be JSON or natural language."""
try:
# First try to parse as JSON
return json.loads(input_str)
except json.JSONDecodeError:
# If not JSON, try to parse natural language
input_lower = input_str.lower()
# Parse for addition
if "add" in input_lower or "plus" in input_lower or "sum" in input_lower:
# Extract numbers - very simple approach
numbers = [int(s) for s in input_lower.split() if s.isdigit()]
if len(numbers) >= 2:
return {"a": numbers[0], "b": numbers[1]}
# Parse for multiplication
if "multiply" in input_lower or "times" in input_lower or "product" in input_lower:
numbers = [int(s) for s in input_lower.split() if s.isdigit()]
if len(numbers) >= 2:
return {"a": numbers[0], "b": numbers[1]}
# Parse for division
if "divide" in input_lower or "quotient" in input_lower:
numbers = [int(s) for s in input_lower.split() if s.isdigit()]
if len(numbers) >= 2:
return {"a": numbers[0], "b": numbers[1]}
# Could not parse successfully
return None
# Define tools with more flexible input handling
# Define tools
@mcp.tool()
def add(input_str=None, a=None, b=None):
"""Add two numbers. Can accept JSON parameters or natural language.
def add(a: int, b: int) -> str:
"""Add two numbers.
Args:
input_str (str, optional): Natural language input to parse
a (int, optional): First number if provided directly
b (int, optional): Second number if provided directly
a (int): First number
b (int): Second number
Returns:
str: A message containing the sum
"""
logger.info(f"Add tool called with input_str={input_str}, a={a}, b={b}")
# If we got a natural language string instead of parameters
if input_str and not (a is not None and b is not None):
parsed = parse_input(input_str)
if parsed:
a = parsed.get("a")
b = parsed.get("b")
# Validate we have what we need
if a is None or b is None:
return "Sorry, I couldn't understand the numbers to add"
try:
a = int(a)
b = int(b)
logger.info(f"Adding {a} and {b}")
result = a + b
return f"The sum of {a} and {b} is {result}"
except ValueError:
return "Please provide valid numbers for addition"
@mcp.tool()
def multiply(input_str=None, a=None, b=None):
"""Multiply two numbers. Can accept JSON parameters or natural language.
def multiply(a: int, b: int) -> str:
"""Multiply two numbers.
Args:
input_str (str, optional): Natural language input to parse
a (int, optional): First number if provided directly
b (int, optional): Second number if provided directly
a (int): First number
b (int): Second number
Returns:
str: A message containing the product
"""
logger.info(
f"Multiply tool called with input_str={input_str}, a={a}, b={b}")
# If we got a natural language string instead of parameters
if input_str and not (a is not None and b is not None):
parsed = parse_input(input_str)
if parsed:
a = parsed.get("a")
b = parsed.get("b")
# Validate we have what we need
if a is None or b is None:
return "Sorry, I couldn't understand the numbers to multiply"
try:
a = int(a)
b = int(b)
logger.info(f"Multiplying {a} and {b}")
result = a * b
return f"The product of {a} and {b} is {result}"
except ValueError:
return "Please provide valid numbers for multiplication"
@mcp.tool()
def divide(input_str=None, a=None, b=None):
"""Divide two numbers. Can accept JSON parameters or natural language.
def divide(a: int, b: int) -> str:
"""Divide two numbers.
Args:
input_str (str, optional): Natural language input to parse
a (int, optional): Numerator if provided directly
b (int, optional): Denominator if provided directly
a (int): Numerator
b (int): Denominator
Returns:
str: A message containing the division result or an error message
"""
logger.info(f"Divide tool called with input_str={input_str}, a={a}, b={b}")
# If we got a natural language string instead of parameters
if input_str and not (a is not None and b is not None):
parsed = parse_input(input_str)
if parsed:
a = parsed.get("a")
b = parsed.get("b")
# Validate we have what we need
if a is None or b is None:
return "Sorry, I couldn't understand the numbers to divide"
try:
a = int(a)
b = int(b)
logger.info(f"Dividing {a} by {b}")
if b == 0:
logger.warning("Division by zero attempted")
return "Cannot divide by zero"
result = a / b
return f"{a} divided by {b} is {result}"
except ValueError:
return "Please provide valid numbers for division"
if __name__ == "__main__":
try:
# Log server details
logger.info("Starting math server on http://0.0.0.0:8000")
print("Math MCP Server is running. Press Ctrl+C to stop.")
print(
"Server is configured to accept both JSON and natural language input"
)
print("Math MCP Server is running on http://0.0.0.0:8000")
print("Press Ctrl+C to stop.")
# List available tools
print("\nAvailable tools:")
print("- add: Add two numbers")
print("- multiply: Multiply two numbers")
print("- divide: Divide first number by second number")
# Add a small delay to ensure logging is complete before the server starts
# Add a small delay to ensure logging is complete
time.sleep(0.5)
# Run the MCP server

Write Preview
Loading…
Cancel
Save