File size: 4,378 Bytes
71cadf5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 |
import os
import uuid
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP, Context
# Load environment variables
load_dotenv()
# Create MCP server
mcp = FastMCP(
name="healthcare-mcp",
version="1.0.0",
description="Healthcare MCP server for medical information access"
)
# Import tools and 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
# Initialize tool instances and services
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")
# Generate a unique session ID for this connection
session_id = str(uuid.uuid4())
@mcp.tool()
async def fda_drug_lookup(ctx: Context, drug_name: str, search_type: str = "general"):
"""
Look up drug information from the FDA database
Args:
drug_name: Name of the drug to search for
search_type: Type of information to retrieve: 'label', 'adverse_events', or 'general'
"""
# Record usage
usage_service.record_usage(session_id, "fda_drug_lookup")
# Call the tool
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 PubMed database
Args:
query: Search query for medical literature
max_results: Maximum number of results to return
date_range: Limit to articles published within years (e.g. '5' for last 5 years)
"""
# Record usage
usage_service.record_usage(session_id, "pubmed_search")
# Call the tool
return await pubmed_tool.search_literature(query, max_results, date_range)
@mcp.tool()
async def health_topics(ctx: Context, topic: str, language: str = "en"):
"""
Get evidence-based health information on various topics
Args:
topic: Health topic to search for information
language: Language for content (en or es)
"""
# Record usage
usage_service.record_usage(session_id, "health_topics")
# Call the tool
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):
"""
Search for clinical trials by condition, status, and other parameters
Args:
condition: Medical condition or disease to search for
status: Trial status (recruiting, completed, active, not_recruiting, or all)
max_results: Maximum number of results to return
"""
# Record usage
usage_service.record_usage(session_id, "clinical_trials_search")
# Call the tool
return await clinical_trials_tool.search_trials(condition, status, max_results)
@mcp.tool()
async def lookup_icd_code(ctx: Context, code: str = None, description: str = None, max_results: int = 10):
"""
Look up ICD-10 codes by code or description
Args:
code: ICD-10 code to look up (optional if description is provided)
description: Medical condition description to search for (optional if code is provided)
max_results: Maximum number of results to return
"""
# Record usage
usage_service.record_usage(session_id, "lookup_icd_code")
# Call the tool
return await medical_terminology_tool.lookup_icd_code(code, description, max_results)
@mcp.tool()
async def get_usage_stats(ctx: Context):
"""
Get usage statistics for the current session
Returns:
A summary of API usage for the current session
"""
return usage_service.get_monthly_usage(session_id)
@mcp.tool()
async def get_all_usage_stats(ctx: Context):
"""
Get overall usage statistics for all sessions
Returns:
A summary of API usage across all sessions
"""
return usage_service.get_usage_stats()
if __name__ == "__main__":
# Using FastMCP's CLI
import sys
sys.exit(mcp.run())
|