import os
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier
from dotenv import load_dotenv
from fastmcp import FastMCP
from fastmcp import Client
import logging
import sys
sys.path.append(r"../mcp-server")
from mcp_services.pdf_extractor_server import pdf_extract_mcp
from mcp_services.embedding_server import embedding_mcp
from mcp_services.reranker_server import reranker_mcp
from mcp_services.postgres_server import postgres_mcp
from mcp_services.markup_server import markup_mcp
from mcp_services.qdrant_server import qdrant_mcp
from mcp_services.llm_server import llm_mcp
from mcp_services.transcribe_server import transcribe_mcp
load_dotenv()
public_key_pem = os.getenv("MAIN_SERVER_API_KEY")
verifier = StaticTokenVerifier(
tokens={
public_key_pem: {
"client_id": "user",
"scopes": ["read:data", "write:data", "admin:users"],
}
}
)
logger = logging.getLogger(__name__)
MCP_SERVERS = [
embedding_mcp,
pdf_extract_mcp,
reranker_mcp,
qdrant_mcp,
postgres_mcp,
llm_mcp,
markup_mcp,
transcribe_mcp,
]
MCP_SERVERS_DICT = {server.name: server for server in MCP_SERVERS}
main_mcp_server = FastMCP(name="Main MCP server", auth=verifier)
@main_mcp_server.tool
async def get_server_and_tools():
"""
Get tools from all servers
:return: List of servers and tools
"""
total_description = {}
# TODO: Add dict servers
for server in MCP_SERVERS:
tools = await server.get_tools()
total_description[server.name] = []
for tool_name, tool_obj in tools.items():
total_description[server.name].append(
{"tool_name": tool_name, "description": tool_obj.description}
)
return total_description
@main_mcp_server.tool
async def router(server_name: str, tool_name: str, params: dict):
"""
Router for all servers
:param server_name: Name of the server
:param tool_name: Name of the tool
:param params: Parameters for the tool
:return: Result of the tool
"""
curr_server = MCP_SERVERS_DICT[server_name]
async with Client(curr_server) as client:
result = await client.call_tool(tool_name, params)
logger.info(f"Result is {result}")
return str(result.data)
@main_mcp_server.tool
async def health_check_servers():
"""
health check for all server
:return: List of servers status
"""
health_check_result = {}
for server_name, server in MCP_SERVERS_DICT.items():
try:
async with Client(server) as client:
result = await client.call_tool("health_check")
health_check_result[server_name] = result.data["status"]
except Exception as ex:
health_check_result[server_name] = str(ex)
logger.info(health_check_result)
return health_check_result
if __name__ == "__main__":
main_mcp_server.run(transport="stdio")
