Add initial implementation of ADE Explorer with requirements, plotting, and API client

README.md

@@ -1 +1,49 @@
-# ADE-Explorer-MCP
+# Medication Adverse-Event Explorer (ADE Explorer)
+
+A Gradio app that exposes a drug-event query tool as an MCP server. It has two core workflows: 
+(A) finding top adverse events for a drug, and 
+(B) checking the frequency of a specific drug-event pair. 
+
+It uses public data from OpenFDA.
+
+**Hackathon Track:** `mcp-server-track`
+
+---
+
+## Usage
+
+This application provides two tools, accessible via a tabbed interface or as MCP endpoints:
+
+1.  **Top Events**: For a given drug name, it returns a list and a bar chart of the top 10 most frequently reported adverse events.
+2.  **Event Frequency**: For a given drug and adverse event pair, it returns the total number of reports found.
+
+## MCP Server Configuration
+
+To use this application as a tool with an MCP client (like Cursor or Claude Desktop), you first need to run it and get the server URL.
+
+1.  **Run the App**:
+    ```bash
+    python src/app.py
+    ```
+2.  **Find the MCP URL**: The server will start and print the MCP URL in the console. It will look something like this:
+    `http://127.0.0.1:7860/gradio_api/mcp/sse`
+3.  **Configure Your Client**: Add the following configuration to your MCP client's settings, replacing the URL with the one from the previous step.
+
+    ```json
+    {
+      "mcpServers": {
+        "ade_explorer": {
+          "url": "http://127.0.0.1:7860/gradio_api/mcp/sse"
+        }
+      }
+    }
+    ```
+
+The server exposes two tools: `top_adverse_events_tool(drug_name)` and `drug_event_stats_tool(drug_name, event_name)`.
+
+---
+
+## Data Source and Disclaimer
+
+*   **Source**: All data is sourced from the public [FDA Adverse Event Reporting System (FAERS)](https://open.fda.gov/data/faers/) via the OpenFDA API.
+*   **Disclaimer**: The information provided by this tool is for informational purposes only and is based on spontaneous reports. It does not represent verified medical data or clinical proof of a causal relationship. **Always consult a qualified healthcare professional for medical advice.**

requirements.txt

@@ -0,0 +1,6 @@
+gradio[mcp]>=5.0
+pandas
+requests
+plotly
+cachetools
+pytest 

src/app.py

@@ -0,0 +1,113 @@
+import gradio as gr
+from src.openfda_client import get_top_adverse_events, get_drug_event_pair_frequency
+from src.plotting import create_bar_chart
+import pandas as pd
+
+# --- Formatting Functions ---
+
+def format_top_events_results(data: dict, drug_name: str) -> str:
+    """Formats the results for the top adverse events tool."""
+    if "error" in data:
+        return f"An error occurred: {data['error']}"
+    
+    if "results" not in data or not data["results"]:
+        return f"No adverse event data found for '{drug_name}'. The drug may not be in the database or it might be misspelled."
+
+    header = f"Top Adverse Events for '{drug_name.title()}'\n"
+    header += "Source: FDA FAERS via OpenFDA\n"
+    header += "Disclaimer: Spontaneous reports do not prove causation. Consult a healthcare professional.\n"
+    header += "---------------------------------------------------\n"
+
+    try:
+        df = pd.DataFrame(data["results"])
+        df = df.rename(columns={"term": "Adverse Event", "count": "Report Count"})
+        result_string = df.to_string(index=False)
+        return header + result_string
+    except Exception as e:
+        return f"An error occurred while formatting the data: {e}"
+
+def format_pair_frequency_results(data: dict, drug_name: str, event_name: str) -> str:
+    """Formats the results for the drug-event pair frequency tool."""
+    if "error" in data:
+        return f"An error occurred: {data['error']}"
+    
+    total_reports = data.get("meta", {}).get("results", {}).get("total", 0)
+
+    result = (
+        f"Found {total_reports:,} reports for the combination of "
+        f"'{drug_name.title()}' and '{event_name.title()}'.\n\n"
+        "Source: FDA FAERS via OpenFDA\n"
+        "Disclaimer: Spontaneous reports do not prove causation. Consult a healthcare professional."
+    )
+    return result
+
+# --- Tool Functions ---
+
+def top_adverse_events_tool(drug_name: str):
+    """
+    MCP Tool: Finds the top reported adverse events for a given drug.
+
+    Args:
+        drug_name (str): The name of the drug to search for.
+    
+    Returns:
+        tuple: A Plotly figure and a formatted string with the top adverse events.
+    """
+    data = get_top_adverse_events(drug_name)
+    chart = create_bar_chart(data, drug_name)
+    text_summary = format_top_events_results(data, drug_name)
+    return chart, text_summary
+
+def drug_event_stats_tool(drug_name: str, event_name: str):
+    """
+    MCP Tool: Gets the total number of reports for a specific drug and adverse event pair.
+
+    Args:
+        drug_name (str): The name of the drug to search for.
+        event_name (str): The name of the adverse event to search for.
+    
+    Returns:
+        str: A formatted string with the total count of reports.
+    """
+    data = get_drug_event_pair_frequency(drug_name, event_name)
+    return format_pair_frequency_results(data, drug_name, event_name)
+
+# --- Gradio Interface ---
+
+interface1 = gr.Interface(
+    fn=top_adverse_events_tool,
+    inputs=[
+        gr.Textbox(
+            label="Drug Name", 
+            info="Enter a brand or generic drug name (e.g., 'Aspirin', 'Lisinopril')."
+        )
+    ],
+    outputs=[
+        gr.Plot(label="Top Adverse Events Chart"),
+        gr.Textbox(label="Top Adverse Events (Raw Data)", lines=15)
+    ],
+    title="Top Adverse Events by Drug",
+    description="Find the most frequently reported adverse events for a specific medication.",
+    examples=[["Lisinopril"], ["Ozempic"], ["Metformin"]],
+)
+
+interface2 = gr.Interface(
+    fn=drug_event_stats_tool,
+    inputs=[
+        gr.Textbox(label="Drug Name", info="e.g., 'Ibuprofen'"),
+        gr.Textbox(label="Adverse Event", info="e.g., 'Headache'")
+    ],
+    outputs=[gr.Textbox(label="Report Count", lines=5)],
+    title="Drug/Event Pair Frequency",
+    description="Get the total number of reports for a specific drug and adverse event combination.",
+    examples=[["Lisinopril", "Cough"], ["Ozempic", "Nausea"]],
+)
+
+demo = gr.TabbedInterface(
+    [interface1, interface2], 
+    ["Top Events", "Event Frequency"],
+    title="Medication Adverse-Event Explorer"
+)
+
+if __name__ == "__main__":
+    demo.launch(mcp_server=True, server_name="0.0.0.0") 

src/openfda_client.py

@@ -0,0 +1,103 @@
+import requests
+from cachetools import TTLCache, cached
+import time
+
+API_BASE_URL = "https://api.fda.gov/drug/event.json"
+# Cache with a TTL of 10 minutes (600 seconds)
+cache = TTLCache(maxsize=256, ttl=600)
+
+# 240 requests per minute per IP. A 0.25s delay is a simple way to stay under.
+REQUEST_DELAY_SECONDS = 0.25
+
+def get_top_adverse_events(drug_name: str, limit: int = 10) -> dict:
+    """
+    Query OpenFDA to get the top adverse events for a given drug.
+
+    Args:
+        drug_name (str): The name of the drug to search for (brand or generic).
+        limit (int): The maximum number of adverse events to return.
+
+    Returns:
+        dict: The JSON response from the API, or an error dictionary.
+    """
+    if not drug_name:
+        return {"error": "Drug name cannot be empty."}
+
+    drug_name_processed = drug_name.lower().strip()
+
+    # Using a simple cache key
+    cache_key = f"top_events_{drug_name_processed}_{limit}"
+
+    if cache_key in cache:
+        return cache[cache_key]
+
+    query = (
+        f'search=patient.drug.medicinalproduct:"{drug_name_processed}"'
+        f'&count=patient.reaction.reactionmeddrapt.exact&limit={limit}'
+    )
+    
+    try:
+        # Respect rate limits
+        time.sleep(REQUEST_DELAY_SECONDS)
+        
+        response = requests.get(f"{API_BASE_URL}?{query}")
+        response.raise_for_status()  # Raise an exception for bad status codes (4xx or 5xx)
+        
+        data = response.json()
+        cache[cache_key] = data
+        return data
+
+    except requests.exceptions.HTTPError as http_err:
+        if response.status_code == 404:
+            return {"error": f"No data found for drug: '{drug_name}'. It might be misspelled or not in the database."}
+        return {"error": f"HTTP error occurred: {http_err}"}
+    except requests.exceptions.RequestException as req_err:
+        return {"error": f"A network request error occurred: {req_err}"}
+    except Exception as e:
+        return {"error": f"An unexpected error occurred: {e}"}
+
+def get_drug_event_pair_frequency(drug_name: str, event_name: str) -> dict:
+    """
+    Query OpenFDA to get the total number of reports for a specific
+    drug-adverse event pair.
+
+    Args:
+        drug_name (str): The name of the drug.
+        event_name (str): The name of the adverse event.
+
+    Returns:
+        dict: The JSON response from the API, or an error dictionary.
+    """
+    if not drug_name or not event_name:
+        return {"error": "Drug name and event name cannot be empty."}
+
+    drug_name_processed = drug_name.lower().strip()
+    event_name_processed = event_name.lower().strip()
+
+    cache_key = f"pair_freq_{drug_name_processed}_{event_name_processed}"
+    if cache_key in cache:
+        return cache[cache_key]
+
+    query = (
+        f'search=patient.drug.medicinalproduct:"{drug_name_processed}"'
+        f'+AND+patient.reaction.reactionmeddrapt:"{event_name_processed}"'
+    )
+
+    try:
+        time.sleep(REQUEST_DELAY_SECONDS)
+        
+        response = requests.get(f"{API_BASE_URL}?{query}")
+        response.raise_for_status()
+        
+        data = response.json()
+        cache[cache_key] = data
+        return data
+
+    except requests.exceptions.HTTPError as http_err:
+        if response.status_code == 404:
+            return {"error": f"No data found for drug '{drug_name}' and event '{event_name}'. They may be misspelled or not in the database."}
+        return {"error": f"HTTP error occurred: {http_err}"}
+    except requests.exceptions.RequestException as req_err:
+        return {"error": f"A network request error occurred: {req_err}"}
+    except Exception as e:
+        return {"error": f"An unexpected error occurred: {e}"}

src/plotting.py

@@ -0,0 +1,47 @@
+import plotly.graph_objects as go
+import pandas as pd
+
+def create_bar_chart(data: dict, drug_name: str):
+    """
+    Creates a Plotly bar chart from the OpenFDA data.
+
+    Args:
+        data (dict): The data from the OpenFDA client.
+        drug_name (str): The name of the drug.
+
+    Returns:
+        A Plotly Figure object if data is valid, otherwise None.
+    """
+    if "error" in data or "results" not in data or not data["results"]:
+        return None
+
+    try:
+        df = pd.DataFrame(data["results"])
+        df = df.rename(columns={"term": "Adverse Event", "count": "Report Count"})
+        
+        # Ensure 'Report Count' is numeric
+        df['Report Count'] = pd.to_numeric(df['Report Count'])
+
+        # Sort for better visualization
+        df = df.sort_values(by="Report Count", ascending=True)
+
+        fig = go.Figure(
+            go.Bar(
+                x=df["Report Count"],
+                y=df["Adverse Event"],
+                orientation='h',
+                marker=dict(color='skyblue')
+            )
+        )
+
+        fig.update_layout(
+            title_text=f"Top Reported Adverse Events for {drug_name.title()}",
+            xaxis_title="Number of Reports",
+            yaxis_title="Adverse Event",
+            yaxis=dict(automargin=True),
+            height=max(400, len(df) * 30) # Dynamically adjust height
+        )
+
+        return fig
+    except Exception:
+        return None 

tests/test_openfda_client.py

@@ -0,0 +1,77 @@
+import pytest
+from unittest.mock import patch, MagicMock
+import requests
+from src.openfda_client import get_top_adverse_events, get_drug_event_pair_frequency, cache
+
+@pytest.fixture(autouse=True)
+def clear_cache():
+    """Fixture to clear the cache before each test."""
+    cache.clear()
+
+def mock_response(status_code=200, json_data=None, raise_for_status=None):
+    """Helper function to create a mock response object."""
+    mock_resp = MagicMock()
+    mock_resp.status_code = status_code
+    mock_resp.json.return_value = json_data
+    if raise_for_status:
+        mock_resp.raise_for_status.side_effect = raise_for_status
+    return mock_resp
+
+@patch('requests.get')
+def test_get_top_adverse_events_success(mock_get):
+    """Test successful API call for top adverse events."""
+    mock_json = {"results": [{"term": "Nausea", "count": 100}]}
+    mock_get.return_value = mock_response(json_data=mock_json)
+    
+    result = get_top_adverse_events("testdrug")
+    
+    assert result == mock_json
+    mock_get.assert_called_once()
+
+@patch('requests.get')
+def test_get_top_adverse_events_404(mock_get):
+    """Test 404 Not Found error for top adverse events."""
+    http_error = requests.exceptions.HTTPError("404 Client Error")
+    mock_get.return_value = mock_response(status_code=404, raise_for_status=http_error)
+    
+    result = get_top_adverse_events("nonexistentdrug")
+    
+    assert "error" in result
+    assert "No data found" in result["error"]
+
+@patch('requests.get')
+def test_get_drug_event_pair_frequency_success(mock_get):
+    """Test successful API call for drug-event pair frequency."""
+    mock_json = {"meta": {"results": {"total": 50}}}
+    mock_get.return_value = mock_response(json_data=mock_json)
+    
+    result = get_drug_event_pair_frequency("testdrug", "testevent")
+    
+    assert result == mock_json
+    mock_get.assert_called_once()
+
+def test_empty_drug_name_returns_error():
+    """Test that empty inputs are handled correctly without calling the API."""
+    result = get_top_adverse_events("")
+    assert "error" in result
+    
+    result2 = get_drug_event_pair_frequency("", "testevent")
+    assert "error" in result2
+
+@patch('requests.get')
+def test_caching_works(mock_get):
+    """Test that results are cached to avoid repeated API calls."""
+    mock_json = {"results": [{"term": "Headache", "count": 200}]}
+    mock_get.return_value = mock_response(json_data=mock_json)
+
+    # First call - should call the API
+    get_top_adverse_events("cacheddrug")
+    assert mock_get.call_count == 1
+
+    # Second call - should hit the cache
+    get_top_adverse_events("cacheddrug")
+    assert mock_get.call_count == 1 # Still 1, not 2
+
+    # Call with different params - should trigger a new API call
+    get_top_adverse_events("cacheddrug", limit=20)
+    assert mock_get.call_count == 2