import re
import base64
import asyncio
from fastmcp import Client
def file_to_base64(file_path: str) -> str:
"""
Converts a file to a Base64 encoded string.
Args:
file_path (str): The path to the file to be encoded.
Returns:
str: The Base64 encoded string of the file content, or None if an error occurs.
"""
with open(file_path, "rb") as file:
binary_data = file.read()
encoded_data = base64.b64encode(binary_data)
return encoded_data.decode("utf-8")
async def use_router(config, server_name, tool_name, params):
async with Client(config) as client:
arguments = {
"server_name": server_name,
"tool_name": tool_name,
"params": params,
}
result = await client.call_tool("router", arguments=arguments)
return result.content[0].text
async def get_texts_gather(tasks):
return await asyncio.gather(*tasks)
def upsert_vectors(
config, chunks_list, collection_name, vector_lists, server_name, tool_upsert_name
):
data = []
for i, text in enumerate(chunks_list):
data.append({"id": i, "payload": {"text": text}})
upsert_result = asyncio.run(
use_router(
config,
server_name,
tool_upsert_name,
params={
"collection_name": collection_name,
"vectors": vector_lists,
"data": data,
},
)
)
return upsert_result
def create_collection(
config, server_name, tool_get_collection_info_name, tool_create_name, vector_size
):
collection_name = "collection_for_rag_"
for i in range(1, 11):
curr_name = collection_name + str(i)
exist_result = asyncio.run(
use_router(
config,
server_name,
tool_get_collection_info_name,
params={"name": curr_name},
)
)
print(exist_result)
if f"Collection `{curr_name}` doesn't exist!" in exist_result:
collection_name = curr_name
break
create_result = asyncio.run(
use_router(
config,
server_name,
tool_create_name,
params={"name": collection_name, "vector_size": vector_size},
)
)
return collection_name, create_result
