swarms/attached_assets/Pasted-from-swarms-import-Agent-from-swarms-tools-mcp-integration-import-MCPServerSseParams-MCPServerSse--1745170779273.txt

from swarms import Agent
from swarms.tools.mcp_integration import MCPServerSseParams, MCPServerSse, mcp_flow_get_tool_schema
from loguru import logger
import sys
import asyncio
import json
import httpx
import time

# Configure logging for more detailed output
logger.remove()
logger.add(sys.stdout,
           level="DEBUG",
           format="{time} | {level} | {module}:{function}:{line} - {message}")

# Relaxed prompt that doesn't enforce strict JSON formatting



# Create server parameters
def get_server_params():
    """Get the MCP server connection parameters."""
    return MCPServerSseParams(
        url=
        "http://127.0.0.1:8000",  # Use 127.0.0.1 instead of localhost/0.0.0.0
        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:
        # Create a short-lived connection to check server
        server = MCPServerSse(get_server_params())

        # Try connecting (this is synchronous)
        asyncio.run(server.connect())
        asyncio.run(server.cleanup())
        logger.info("✅ Server connection test successful")
        return True
    except Exception as e:
        logger.error(f"❌ Server connection test 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")
"""

    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...")

        # 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

        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")

        while True:
            try:
                query = input("What would you like to calculate? ").strip()
                if not query:
                    continue
                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

                logger.info(f"Processing query: {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)}")

    except Exception as e:
        logger.error(f"System initialization error: {e}")
        print(f"Failed to start the math system: {str(e)}")


if __name__ == "__main__":
    main() "from fastmcp import FastMCP
from loguru import logger
import time
import json

# Create the MCP server with detailed debugging
mcp = FastMCP(
    host="0.0.0.0",  # Bind to all interfaces
    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
)


# 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
@mcp.tool()
def add(input_str=None, a=None, b=None):
    """Add two numbers. Can accept JSON parameters or natural language.

    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

    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)
        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.

    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

    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)
        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.

    Args:
        input_str (str, optional): Natural language input to parse
        a (int, optional): Numerator if provided directly
        b (int, optional): Denominator if provided directly

    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)

        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:
        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"
        )

        # Add a small delay to ensure logging is complete before the server starts
        time.sleep(0.5)

        # Run the MCP server
        mcp.run()
    except KeyboardInterrupt:
        logger.info("Server shutdown requested")
        print("\nShutting down server...")
    except Exception as e:
        logger.error(f"Server error: {e}")
        raise
" server is runnig poeroperly "2025-04-20 17:35:01.251 | INFO     | __main__:<module>:161 - Starting math server on http://0.0.0.0:8000
Math MCP Server is running. Press Ctrl+C to stop.
Server is configured to accept both JSON and natural language input
[04/20/25 17:35:01] INFO     Starting server "FastMCP"...            " butwhy im getting these errore "2025-04-20T17:35:04.174629+0000 | INFO | mcp_client:main:159 - Initializing math system...
2025-04-20T17:35:04.203591+0000 | ERROR | mcp_integration:connect:89 - Error initializing MCP server: unhandled errors in a TaskGroup (1 sub-exception)
2025-04-20T17:35:04.204437+0000 | ERROR | mcp_client:test_server_connection:110 - ❌ Server connection test failed: unhandled errors in a TaskGroup (1 sub-exception)

Server connection failed - using fallback calculator mode
Ask me any math question!
Examples: 'what is 5 plus 3?' or 'can you multiply 4 and 6?'
Type 'list tools' to see available operations
Type 'exit' to quit

What would you like to calculate? "