# server.py: MCP server hosting two RAG-based healthcare tools using Groq API and Chroma
# Located in C:\Users\sniki\OneDrive\Desktop\health_rag_mcp\mcprag\, runs in Cursor IDE on Windows with uv
from fastmcp import FastMCP
from langchain.chains import RetrievalQA
from langchain.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_groq import ChatGroq
import json
import os
from dotenv import load_dotenv
# Load environment variables from .env
load_dotenv()
GROQ_API_KEY = os.getenv("GROQ_API_KEY")
if not GROQ_API_KEY:
raise ValueError("GROQ_API_KEY not found in .env file")
# Initialize MCP server
mcp = FastMCP("HealthcareRAGTools")
# Initialize Groq API and RAG components
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
llm = ChatGroq(groq_api_key=GROQ_API_KEY, model_name="llama-3.3-70b-versatile")
persist_directory = "./chroma_db"
# Load Chroma database (assumes setup_db.py has run)
docsearch = Chroma(persist_directory=persist_directory, embedding_function=embeddings)
qa = RetrievalQA.from_chain_type(llm=llm, retriever=docsearch.as_retriever(search_kwargs={"k": 2}))
# File for patient symptom records
PATIENT_RECORDS_FILE = "patient_records.json"
@mcp.tool()
async def search_documents(query: str) -> str:
"""
Search medical documents using RAG and Groq API.
Args:
query: The search query (e.g., "symptoms of flu").
Returns:
Relevant document content or error message.
"""
try:
if not query.strip():
return "Error: Query cannot be empty."
result = qa.invoke(query)
return result["result"] if result["result"] else "No relevant documents found."
except Exception as e:
return f"Error searching documents: {str(e)}"
@mcp.tool()
async def log_patient_symptoms(patient_id: str, symptoms: str, severity: str, get_similar: bool = False) -> str:
"""
Log patient symptoms and optionally retrieve similar cases using RAG.
Args:
patient_id: Unique patient identifier (string).
symptoms: Description of symptoms (string).
severity: Symptom severity (Mild, Moderate, Severe).
get_similar: If true, retrieve similar cases.
Returns:
Confirmation or similar cases.
"""
if not patient_id.strip() or not symptoms.strip():
return "Error: Patient ID and symptoms cannot be empty."
if severity not in ["Mild", "Moderate", "Severe"]:
return "Error: Severity must be Mild, Moderate, or Severe."
if not os.path.exists(PATIENT_RECORDS_FILE):
with open(PATIENT_RECORDS_FILE, "w") as f:
json.dump([], f)
with open(PATIENT_RECORDS_FILE, "r") as f:
records = json.load(f)
record = {
"patient_id": patient_id.strip(),
"symptoms": symptoms.strip(),
"severity": severity,
"timestamp": os.popen("powershell Get-Date -Format 'yyyy-MM-ddTHH:mm:ssZ'").read().strip()
}
records.append(record)
with open(PATIENT_RECORDS_FILE, "w") as f:
json.dump(records, f, indent=2)
docsearch.add_texts([f"Patient {patient_id}: {symptoms}"], metadatas=[{"type": "patient_record"}])
if get_similar:
similar_docs = docsearch.similarity_search(symptoms, k=2, filter={"type": "patient_record"})
similar_cases = "\n".join([f"- {doc.page_content}" for doc in similar_docs])
return f"Symptoms logged for Patient {patient_id}: '{symptoms}' ({severity}).\nSimilar cases:\n{similar_cases or 'None'}"
return f"Symptoms logged for Patient {patient_id}: '{symptoms}' ({severity})."
if __name__ == "__main__":
mcp.run(transport="stdio")
