Update app.py

app.py

@@ -1,64 +1,255 @@
+import os
+import uuid
+import json
+import asyncio
+from dotenv import load_dotenv
+
 import gradio as gr
-from huggingface_hub import InferenceClient
+from mcp.server.fastmcp import FastMCP, Context
+
+# ---------------------------------------------------------------------------
+# Environment & MCP initialisation
+# ---------------------------------------------------------------------------
+load_dotenv()
+
+mcp = FastMCP(
+    name="healthcare-mcp",
+    version="1.0.0",
+    description="Healthcare MCP server for medical information access (Gradio edition)"
+)
+
+# ---------------------------------------------------------------------------
+# Internal tools & services
+# ---------------------------------------------------------------------------
+from src.tools.fda_tool import FDATool
+from src.tools.pubmed_tool import PubMedTool
+from src.tools.healthfinder_tool import HealthFinderTool
+from src.tools.clinical_trials_tool import ClinicalTrialsTool
+from src.tools.medical_terminology_tool import MedicalTerminologyTool
+from src.services.usage_service import UsageService
+
+fda_tool = FDATool()
+pubmed_tool = PubMedTool()
+healthfinder_tool = HealthFinderTool()
+clinical_trials_tool = ClinicalTrialsTool()
+medical_terminology_tool = MedicalTerminologyTool()
+
+usage_service = UsageService(db_path="healthcare_usage.db")
 
-"""
-For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-"""
-client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
+# Unique session – ensures all usage is grouped for the current interface user
+session_id = str(uuid.uuid4())
 
+# ---------------------------------------------------------------------------
+# FastMCP tool definitions – these can be reached programmatically by MCP
+# clients (and are also reused by the Gradio UI wrappers below).
+# ---------------------------------------------------------------------------
 
-def respond(
-    message,
-    history: list[tuple[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
+@mcp.tool()
+async def fda_drug_lookup(ctx: Context, drug_name: str, search_type: str = "general"):
+    """Look up drug information from the FDA database."""
+    usage_service.record_usage(session_id, "fda_drug_lookup")
+    return await fda_tool.lookup_drug(drug_name, search_type)
+
+
+@mcp.tool()
+async def pubmed_search(ctx: Context, query: str, max_results: int = 5, date_range: str = ""):
+    """Search for medical literature in the PubMed database."""
+    usage_service.record_usage(session_id, "pubmed_search")
+    return await pubmed_tool.search_literature(query, max_results, date_range)
+
+
+@mcp.tool()
+async def health_topics(ctx: Context, topic: str, language: str = "en"):
+    """Retrieve evidence‑based information for a specific health topic."""
+    usage_service.record_usage(session_id, "health_topics")
+    return await healthfinder_tool.get_health_topics(topic, language)
+
+
+@mcp.tool()
+async def clinical_trials_search(
+    ctx: Context,
+    condition: str,
+    status: str = "recruiting",
+    max_results: int = 10,
 ):
-    messages = [{"role": "system", "content": system_message}]
-
-    for val in history:
-        if val[0]:
-            messages.append({"role": "user", "content": val[0]})
-        if val[1]:
-            messages.append({"role": "assistant", "content": val[1]})
-
-    messages.append({"role": "user", "content": message})
-
-    response = ""
-
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        token = message.choices[0].delta.content
-
-        response += token
-        yield response
-
-
-"""
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-demo = gr.ChatInterface(
-    respond,
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
-)
+    """Search for clinical trials by condition and status."""
+    usage_service.record_usage(session_id, "clinical_trials_search")
+    return await clinical_trials_tool.search_trials(condition, status, max_results)
+
+
+@mcp.tool()
+async def lookup_icd_code(
+    ctx: Context,
+    code: str | None = None,
+    description: str | None = None,
+    max_results: int = 10,
+):
+    """Look up ICD‑10 codes by code or text description."""
+    usage_service.record_usage(session_id, "lookup_icd_code")
+    return await medical_terminology_tool.lookup_icd_code(code, description, max_results)
+
+
+@mcp.tool()
+async def get_usage_stats(ctx: Context):
+    """Return per‑session usage statistics."""
+    return usage_service.get_monthly_usage(session_id)
+
+
+@mcp.tool()
+async def get_all_usage_stats(ctx: Context):
+    """Return aggregate usage statistics across all sessions."""
+    return usage_service.get_usage_stats()
+
+
+# ---------------------------------------------------------------------------
+# Helper – convert coroutine results to nicely‑formatted JSON for UI display
+# ---------------------------------------------------------------------------
+
+def pretty_json(data):
+    try:
+        return json.dumps(data, indent=2, ensure_ascii=False)
+    except TypeError:
+        # Fallback if the object isn’t serialisable
+        return str(data)
+
+
+# ---------------------------------------------------------------------------
+# Gradio UI definitions
+# ---------------------------------------------------------------------------
+
+# Notes on async handling:
+#   – Blocks.launch supports async handlers directly; we expose async functions
+#     that call the underlying tool methods.
 
+async def fda_drug_lookup_gr(drug_name, search_type):
+    usage_service.record_usage(session_id, "fda_drug_lookup")
+    res = await fda_tool.lookup_drug(drug_name, search_type)
+    return pretty_json(res)
+
+
+async def pubmed_search_gr(query, max_results, date_range):
+    usage_service.record_usage(session_id, "pubmed_search")
+    res = await pubmed_tool.search_literature(query, int(max_results), str(date_range or ""))
+    return pretty_json(res)
+
+
+async def health_topics_gr(topic, language):
+    usage_service.record_usage(session_id, "health_topics")
+    res = await healthfinder_tool.get_health_topics(topic, language)
+    return pretty_json(res)
+
+
+async def clinical_trials_search_gr(condition, status, max_results):
+    usage_service.record_usage(session_id, "clinical_trials_search")
+    res = await clinical_trials_tool.search_trials(condition, status, int(max_results))
+    return pretty_json(res)
+
+
+async def lookup_icd_code_gr(code, description, max_results):
+    usage_service.record_usage(session_id, "lookup_icd_code")
+    code = code or None
+    description = description or None
+    res = await medical_terminology_tool.lookup_icd_code(code, description, int(max_results))
+    return pretty_json(res)
+
+
+# Usage statistics are synchronous; no external IO
+
+def get_usage_stats_gr():
+    return pretty_json(usage_service.get_monthly_usage(session_id))
+
+
+def get_all_usage_stats_gr():
+    return pretty_json(usage_service.get_usage_stats())
+
+
+# ---------------------------------------------------------------------------
+# Build the Gradio Blocks interface
+# ---------------------------------------------------------------------------
+
+with gr.Blocks(title="Healthcare MCP Explorer") as demo:
+    gr.Markdown("""# 🩺 Healthcare MCP Dashboard\nInteractively query FDA, PubMed, HealthFinder, Clinical‑Trials, and ICD‑10 data — all from one screen.""")
+
+    # ---------------- Drug lookup ----------------
+    with gr.Group():
+        gr.Markdown("## 💊 FDA Drug Lookup")
+        with gr.Row():
+            drug_name_tb = gr.Textbox(label="Drug name", placeholder="e.g. atorvastatin")
+            search_type_dd = gr.Dropdown(
+                label="Search type",
+                choices=["general", "label", "adverse_events"],
+                value="general",
+            )
+            fda_button = gr.Button("Search")
+        fda_output = gr.Textbox(label="Result", lines=12)
+
+    # ---------------- PubMed search -------------
+    with gr.Group():
+        gr.Markdown("## 🔍 PubMed Literature Search")
+        query_tb = gr.Textbox(label="Query", placeholder="e.g. type 2 diabetes glycemic control")
+        with gr.Row():
+            max_results_slider = gr.Slider(1, 20, value=5, step=1, label="Max articles")
+            date_range_tb = gr.Textbox(label="Published within last N years (optional)")
+            pubmed_button = gr.Button("Search PubMed")
+        pubmed_output = gr.Textbox(label="Result", lines=12)
+
+    # ---------------- Health topics -------------
+    with gr.Group():
+        gr.Markdown("## 📚 HealthFinder Topics")
+        topic_tb = gr.Textbox(label="Health topic", placeholder="e.g. hypertension")
+        language_dd = gr.Dropdown(label="Language", choices=["en", "es"], value="en")
+        hf_button = gr.Button("Get Topic Info")
+        hf_output = gr.Textbox(label="Result", lines=12)
+
+    # ---------------- Clinical trials -----------
+    with gr.Group():
+        gr.Markdown("## 🧪 Clinical‑Trials.gov Search")
+        condition_tb = gr.Textbox(label="Condition", placeholder="e.g. melanoma")
+        with gr.Row():
+            status_dd = gr.Dropdown(
+                label="Status",
+                choices=["recruiting", "completed", "active", "not_recruiting", "all"],
+                value="recruiting",
+            )
+            ct_max_slider = gr.Slider(1, 50, value=10, step=1, label="Max results")
+            ct_button = gr.Button("Search Trials")
+        ct_output = gr.Textbox(label="Result", lines=12)
+
+    # ---------------- ICD‑10 lookup -------------
+    with gr.Group():
+        gr.Markdown("## 🏷️ ICD‑10 Lookup")
+        code_tb = gr.Textbox(label="ICD‑10 Code (optional)")
+        description_tb = gr.Textbox(label="Description (optional)")
+        with gr.Row():
+            icd_max_slider = gr.Slider(1, 50, value=10, step=1, label="Max results")
+            icd_button = gr.Button("Lookup ICD‑10")
+        icd_output = gr.Textbox(label="Result", lines=12)
+
+    # ---------------- Usage stats ---------------
+    with gr.Group():
+        gr.Markdown("## 📈 Usage Statistics")
+        with gr.Row():
+            session_button = gr.Button("Current Session")
+            all_button = gr.Button("All‑time Usage")
+        session_output = gr.Textbox(label="Session Usage", lines=6)
+        all_output = gr.Textbox(label="All Usage", lines=6)
+
+    # -------------------------------------------------------------------
+    # Wire‑up callbacks
+    # -------------------------------------------------------------------
+    fda_button.click(fda_drug_lookup_gr, inputs=[drug_name_tb, search_type_dd], outputs=fda_output)
+    pubmed_button.click(pubmed_search_gr, inputs=[query_tb, max_results_slider, date_range_tb], outputs=pubmed_output)
+    hf_button.click(health_topics_gr, inputs=[topic_tb, language_dd], outputs=hf_output)
+    ct_button.click(clinical_trials_search_gr, inputs=[condition_tb, status_dd, ct_max_slider], outputs=ct_output)
+    icd_button.click(lookup_icd_code_gr, inputs=[code_tb, description_tb, icd_max_slider], outputs=icd_output)
+
+    session_button.click(get_usage_stats_gr, outputs=session_output)
+    all_button.click(get_all_usage_stats_gr, outputs=all_output)
+
+# ---------------------------------------------------------------------------
+# Entrypoint – run Gradio and MCP in one process
+# ---------------------------------------------------------------------------
 
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch(mcp_server=True, show_error=True)
+