from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings
import logging
from httpclient import SimpleMCPHttpClient
import asyncio
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# List of sub-server URLs
sub_server_urls = [
]
# List of server names for unique identification
server_names = [
]
# Dictionary to map tool names to server clients
tool_to_server = {}
# Create server clients
servers = [SimpleMCPHttpClient(url) for url in sub_server_urls]
async def discover_tools():
global tool_to_server
for i, server in enumerate(servers):
server_name = server_names[i]
try:
tools = await server.list_tools()
for tool in tools:
tool_name = tool.get('name')
if tool_name:
tool_to_server[tool_name] = (server, server_name)
logger.info(f"Discovered tool '{tool_name}' on {server_name}")
except Exception as e:
logger.error(f"Failed to discover tools from {server_name}: {e}")
# Run discovery on import (assuming servers are running)
# asyncio.run(discover_tools()) # Removed to avoid event loop issues
# Create FastMCP instance - this will be imported by tools
master_MCP = FastMCP(
"master-mcp-server",
stateless_http=True,
transport_security=TransportSecuritySettings(
enable_dns_rebinding_protection=False,
)
)
logger.info("FastMCP instance created successfully")
logger.info("FastMCP instance created successfully")
@master_MCP.tool(description="Call any discovered tool by name with arguments")
async def call_tool(tool_name: str, arguments: dict):
if tool_name not in tool_to_server:
return f"Tool '{tool_name}' not found or not discovered."
server, _ = tool_to_server[tool_name]
return await server.call_tool(tool_name, arguments)
@master_MCP.tool(description="List all discovered tools")
async def list_discovered_tools():
return {
"discovered_tools": list(tool_to_server.keys()),
"tool_mappings": {name: {"url": str(server.base_url), "name": server_name} for name, (server, server_name) in tool_to_server.items()}
}
@master_MCP.tool(description="add the new mcp server url and name and get the url and name from users")
async def add_new_mcp_server(base_url: str, server_name: str):
sub_server_urls.append(base_url)
server_names.append(server_name)
new_server = SimpleMCPHttpClient(base_url)
servers.append(new_server)
try:
tools = await new_server.list_tools()
for tool in tools:
tool_name = tool.get('name')
if tool_name:
tool_to_server[tool_name] = (new_server, server_name)
logger.info(f"Discovered tool '{tool_name}' on {server_name}")
return {"message": f"New MCP server '{server_name}' added Successfully."}
except Exception as e:
logger.error(f"Failed to discover tools from {server_name}: {e}")
return {"error": str(e)}
