from typing import List
from fastmcp import FastMCP, Client
from data_processing_workflow import build_graph_workflow
proxy_mcp_server = FastMCP(name="Proxy Server")
config = {
"mcpServers": {
"server-name": {
"url": "http://localhost:8000/mcp/",
"headers": {"Authorization": "Bearer my_best_api_key"},
}
}
}
file_work_methods = [
"markup_process_file",
"transcribe_audio",
"document_convert_to_markdown",
]
@proxy_mcp_server.tool
async def get_server_and_tools() -> dict:
"""
Get tools from all servers
:return: List of servers and tools
"""
async with Client(config) as client:
response = await client.call_tool("get_server_and_tools")
return response
@proxy_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
"""
params = {"server_name": server_name, "tool_name": tool_name, "params": params}
async with Client(config) as client:
result = await client.call_tool("router", arguments=params, timeout=10)
return result
@proxy_mcp_server.tool
async def preprocessing_data_for_rag(file_paths: List[str]) -> str:
"""
This method is used to pre-prepare pdf files for RAG
:param file_paths: List with file paths
:return: The name of the vector collection that contains the processed search data for RAG.
"""
graph = build_graph_workflow()
state_result = await graph.ainvoke({"file_paths": file_paths, "config": config})
return state_result["qdrant_collection_name"]
@proxy_mcp_server.tool
async def health_check_servers():
async with Client(config) as client:
result = await client.call_tool("health_check_servers")
return result.content[0].text
