server.py•2.92 kB
# server.py
from mcp.server.fastmcp import FastMCP
from rag_code import *
# Create an MCP server
mcp = FastMCP("MCP-RAG-app",
host="127.0.0.1",
port=8080,
timeout=30)
@mcp.tool()
def machine_learning_faq_retrieval_tool(query: str) -> str:
# Recupera os documentos mais relevantes da coleção de perguntas frequentes
# sobre aprendizado de máquina.
"""Recupere os documentos mais relevantes da coleção de perguntas frequentes sobre aprendizado de máquina.
Use esta ferramenta quando o usuário fizer perguntas sobre ML.
Entrada:
query: str -> A consulta do usuário para recuperar os documentos mais relevantes
Saída:
context: str -> Documentos mais relevantes recuperados de um banco de dados vetorial
"""
# check type of text
if not isinstance(query, str):
raise ValueError("query must be a string")
retriever = Retriever(QdrantVDB("ml_faq_collection"), EmbedData())
response = retriever.search(query)
return response
@mcp.tool()
def bright_data_web_search_tool(query: str) -> list[str]:
# Recupera os resultados de pesquisa orgânica do Google usando o Bright Data Proxy.
"""
Pesquise informações sobre um determinado tópico usando o Bright Data.
Use esta ferramenta quando o usuário fizer uma pergunta ou consulta sobre um tópico específico
que não esteja relacionado ao aprendizado de máquina em geral.
Entrada:
query: str -> A consulta do usuário para buscar informações
Saída:
context: list[str] -> Lista dos resultados de pesquisa na web mais relevantes
"""
# check type of text
if not isinstance(query, str):
raise ValueError("query must be a string")
import os
import ssl
import requests
from dotenv import load_dotenv
# Load environment variables and configure SSL
load_dotenv()
ssl._create_default_https_context = ssl._create_unverified_context
# Bright Data configuration
host = 'brd.superproxy.io'
port = 33335
from dotenv import load_dotenv
load_dotenv() # Isso precisa estar antes da leitura do username/password
# get username and password from brightdata.com
username = os.getenv("BRIGHT_DATA_USERNAME")
password = os.getenv("BRIGHT_DATA_PASSWORD")
proxy_url = f'http://{username}:{password}@{host}:{port}'
proxies = {
'http': proxy_url,
'https': proxy_url
}
# Format query and make request
formatted_query = "+".join(query.split(" "))
url = f"https://www.google.com/search?q={formatted_query}&brd_json=1&num=50"
response = requests.get(url, proxies=proxies, verify=False)
# Return organic search results
return response.json()['organic']
if __name__ == "__main__":
print("Starting MCP server at http://127.0.0.1:8080 on port 8080")
mcp.run()