Benefits Agent MCP Server

from fastmcp import FastMCP from typing import Dict, List, Optional from datetime import datetime, timedelta import json import os # Initialize FastMCP server for Benefit Agent mcp = FastMCP("Benefit Agent - Healthcare Financial Tracking") # Load mock data from JSON file def load_mock_data(): """Load mock data from JSON file""" json_file = "ben_agent_mcp_server_data.json" # Check if file exists if not os.path.exists(json_file): raise FileNotFoundError(f"Data file not found: {json_file}") with open(json_file, 'r') as f: data = json.load(f) return data # Load data at startup try: MOCK_DATA = load_mock_data() MOCK_MEMBERS = MOCK_DATA["members"] MOCK_DEDUCTIBLES = MOCK_DATA["deductibles"] MOCK_OUT_OF_POCKET = MOCK_DATA["out_of_pocket"] MOCK_BENEFIT_USAGE = MOCK_DATA["benefit_usage"] MOCK_SPENDING = MOCK_DATA["spending_summary"] print("✓ Mock data loaded successfully from ben_agent_mcp_server_data.json") except FileNotFoundError as e: print(f"ERROR: {e}") print("Please ensure ben_agent_mcp_server_data.json is in the same directory as this script") exit(1) except json.JSONDecodeError as e: print(f"ERROR: Invalid JSON format in data file: {e}") exit(1) @mcp.tool() def get_deductible_status(member_id: str, deductible_type: Optional[str] = None) -> Dict: """Get current deductible status showing amounts used and remaining. This tool shows how much of the annual deductible has been met and how much remains for medical and prescription coverage. Args: member_id: The unique identifier for the member (e.g., 'MEM001') deductible_type: Optional filter for specific deductible type. Valid values: 'medical', 'prescription', 'all' Default: 'all' (returns all deductible types) Returns: Dictionary containing deductible information including amounts met and remaining for individual and family deductibles. """ if member_id not in MOCK_MEMBERS: return { "error": "Member not found", "message": f"No member found with ID: {member_id}", "available_members": list(MOCK_MEMBERS.keys()) } member_info = MOCK_MEMBERS[member_id] deductible_data = MOCK_DEDUCTIBLES.get(member_id, {}) if deductible_type and deductible_type.lower() != "all": deductible_type_lower = deductible_type.lower() if deductible_type_lower in deductible_data: filtered_deductible = {deductible_type_lower: deductible_data[deductible_type_lower]} else: return { "error": "Invalid deductible type", "message": f"Deductible type '{deductible_type}' not recognized", "valid_types": list(deductible_data.keys()) } else: filtered_deductible = deductible_data result = { "member_id": member_id, "member_name": member_info["name"], "plan_type": member_info["plan_type"], "plan_year": f"{member_info['plan_year_start']} to {member_info['plan_year_end']}", "deductibles": {}, "query_date": datetime.now().isoformat() } for ded_type, ded_info in filtered_deductible.items(): if ded_type == "medical": result["deductibles"]["medical"] = { "individual": { "total": f"${ded_info['individual_total']}", "met": f"${ded_info['individual_met']}", "remaining": f"${ded_info['individual_total'] - ded_info['individual_met']}", "percent_met": f"{(ded_info['individual_met'] / ded_info['individual_total'] * 100):.1f}%" }, "family": { "total": f"${ded_info['family_total']}", "met": f"${ded_info['family_met']}", "remaining": f"${ded_info['family_total'] - ded_info['family_met']}", "percent_met": f"{(ded_info['family_met'] / ded_info['family_total'] * 100):.1f}%" } } elif ded_type == "prescription": result["deductibles"]["prescription"] = { "total": f"${ded_info['total']}", "met": f"${ded_info['met']}", "remaining": f"${ded_info['total'] - ded_info['met']}", "percent_met": f"{(ded_info['met'] / ded_info['total'] * 100):.1f}%" } return result @mcp.tool() def get_out_of_pocket_status(member_id: str) -> Dict: """Get out-of-pocket maximum status and progress toward annual limit. This tool shows current spending toward the out-of-pocket maximum, which is the most a member will pay in a plan year. Args: member_id: The unique identifier for the member (e.g., 'MEM001') Returns: Dictionary containing out-of-pocket spending details including current amounts, maximums, and remaining amounts before hitting the cap. """ if member_id not in MOCK_MEMBERS: return { "error": "Member not found", "message": f"No member found with ID: {member_id}" } member_info = MOCK_MEMBERS[member_id] oop_data = MOCK_OUT_OF_POCKET.get(member_id, {}) return { "member_id": member_id, "member_name": member_info["name"], "plan_type": member_info["plan_type"], "plan_year": f"{member_info['plan_year_start']} to {member_info['plan_year_end']}", "out_of_pocket": { "individual": { "maximum": f"${oop_data['individual_maximum']}", "current": f"${oop_data['individual_current']}", "remaining": f"${oop_data['individual_maximum'] - oop_data['individual_current']}", "percent_reached": f"{(oop_data['individual_current'] / oop_data['individual_maximum'] * 100):.1f}%" }, "family": { "maximum": f"${oop_data['family_maximum']}", "current": f"${oop_data['family_current']}", "remaining": f"${oop_data['family_maximum'] - oop_data['family_current']}", "percent_reached": f"{(oop_data['family_current'] / oop_data['family_maximum'] * 100):.1f}%" } }, "query_date": datetime.now().isoformat(), "note": "Once out-of-pocket maximum is reached, the plan pays 100% of covered services for the rest of the plan year." } @mcp.tool() def get_benefit_utilization(member_id: str, benefit_category: Optional[str] = None) -> Dict: """Get detailed benefit utilization across medical, dental, vision, and pharmacy. This tool shows how many services or benefits have been used compared to annual limits or allowances. Args: member_id: The unique identifier for the member (e.g., 'MEM001') benefit_category: Optional filter for specific benefit category. Valid values: 'medical', 'dental', 'vision', 'pharmacy', 'all' Default: 'all' (returns all benefit categories) Returns: Dictionary containing utilization details for each benefit category including services used, limits, and remaining allowances. """ if member_id not in MOCK_MEMBERS: return { "error": "Member not found", "message": f"No member found with ID: {member_id}" } member_info = MOCK_MEMBERS[member_id] usage_data = MOCK_BENEFIT_USAGE.get(member_id, {}) if benefit_category and benefit_category.lower() != "all": category_map = { "medical": "medical_visits", "dental": "dental", "vision": "vision", "pharmacy": "pharmacy" } category_key = category_map.get(benefit_category.lower()) if category_key and category_key in usage_data: filtered_usage = {category_key: usage_data[category_key]} else: return { "error": "Invalid benefit category", "message": f"Benefit category '{benefit_category}' not recognized", "valid_categories": ["medical", "dental", "vision", "pharmacy", "all"] } else: filtered_usage = usage_data return { "member_id": member_id, "member_name": member_info["name"], "plan_type": member_info["plan_type"], "plan_year": f"{member_info['plan_year_start']} to {member_info['plan_year_end']}", "benefit_utilization": filtered_usage, "query_date": datetime.now().isoformat() } @mcp.tool() def get_remaining_benefits(member_id: str) -> Dict: """Get summary of remaining benefits with annual limits. This tool highlights benefits that have annual limits (like dental maximums, vision allowances) and shows what remains available for the year. Args: member_id: The unique identifier for the member (e.g., 'MEM001') Returns: Dictionary containing remaining benefits summary with focus on limited benefits like dental annual maximums and vision allowances. """ if member_id not in MOCK_MEMBERS: return { "error": "Member not found", "message": f"No member found with ID: {member_id}" } member_info = MOCK_MEMBERS[member_id] usage_data = MOCK_BENEFIT_USAGE.get(member_id, {}) remaining = { "member_id": member_id, "member_name": member_info["name"], "plan_type": member_info["plan_type"], "remaining_benefits": {} } # Dental remaining if "dental" in usage_data: dental = usage_data["dental"] remaining["remaining_benefits"]["dental"] = { "annual_maximum_remaining": f"${dental['annual_max_total'] - dental['annual_max_used']}", "annual_maximum_total": f"${dental['annual_max_total']}", "preventive_visits_remaining": dental["preventive"]["limit"] - dental["preventive"]["used"] if dental["preventive"]["limit"] else "Unlimited" } # Vision remaining if "vision" in usage_data: vision = usage_data["vision"] remaining["remaining_benefits"]["vision"] = { "eye_exams_remaining": vision["eye_exams"]["limit"] - vision["eye_exams"]["used"], "frame_allowance_remaining": f"${vision['frames']['allowance_total'] - vision['frames']['allowance_used']}", "contact_allowance_remaining": f"${vision['contacts']['allowance_total'] - vision['contacts']['allowance_used']}" } days_remaining = ( datetime.strptime(member_info['plan_year_end'], '%Y-%m-%d') - datetime.now() ).days remaining["days_remaining_in_plan_year"] = days_remaining remaining["query_date"] = datetime.now().isoformat() remaining["reminder"] = f"You have {days_remaining} days left to use these benefits before they reset." return remaining @mcp.tool() def get_spending_summary(member_id: str) -> Dict: """Get comprehensive spending summary across all benefit categories. This tool provides a financial overview of year-to-date spending, deductible progress, and out-of-pocket progress. Args: member_id: The unique identifier for the member (e.g., 'MEM001') Returns: Dictionary containing complete spending summary including total spending, spending by category, deductible status, and out-of-pocket status. """ if member_id not in MOCK_MEMBERS: return { "error": "Member not found", "message": f"No member found with ID: {member_id}" } member_info = MOCK_MEMBERS[member_id] spending_by_category = MOCK_SPENDING.get(member_id, {}) deductible_data = MOCK_DEDUCTIBLES.get(member_id, {}) oop_data = MOCK_OUT_OF_POCKET.get(member_id, {}) total_spending = sum(spending_by_category.values()) # Deductible summary deductible_status = {} if "medical" in deductible_data: med_ded = deductible_data["medical"] deductible_status["medical"] = { "individual_met": f"${med_ded['individual_met']}", "individual_remaining": f"${med_ded['individual_total'] - med_ded['individual_met']}" } if "prescription" in deductible_data: rx_ded = deductible_data["prescription"] deductible_status["prescription"] = { "met": f"${rx_ded['met']}", "remaining": f"${rx_ded['total'] - rx_ded['met']}" } # Out-of-pocket summary oop_status = { "individual": { "current": f"${oop_data['individual_current']}", "remaining": f"${oop_data['individual_maximum'] - oop_data['individual_current']}" }, "family": { "current": f"${oop_data['family_current']}", "remaining": f"${oop_data['family_maximum'] - oop_data['family_current']}" } } return { "member_id": member_id, "member_name": member_info["name"], "plan_type": member_info["plan_type"], "plan_year": f"{member_info['plan_year_start']} to {member_info['plan_year_end']}", "total_spending_ytd": f"${total_spending}", "spending_by_category": { "breakdown": { "medical": f"${spending_by_category.get('medical', 0)}", "dental": f"${spending_by_category.get('dental', 0)}", "vision": f"${spending_by_category.get('vision', 0)}", "pharmacy": f"${spending_by_category.get('pharmacy', 0)}" } }, "deductible_progress": deductible_status, "out_of_pocket_progress": oop_status, "days_remaining_in_plan_year": ( datetime.strptime(member_info['plan_year_end'], '%Y-%m-%d') - datetime.now() ).days, "query_date": datetime.now().isoformat(), "financial_tip": "Consider scheduling needed services before year-end if you've met your deductible or are close to your out-of-pocket maximum." } if __name__ == "__main__": # Run the Benefit Agent MCP server print("Starting Benefit Agent MCP Server...") print("Available tools:") print(" - get_deductible_status: Check deductible amounts used and remaining") print(" - get_out_of_pocket_status: View out-of-pocket maximum progress") print(" - get_benefit_utilization: Detailed usage across all benefit categories") print(" - get_remaining_benefits: Check remaining benefits with annual limits") print(" - get_spending_summary: Comprehensive year-to-date spending overview") print("\nServer running on http://0.0.0.0:8000") mcp.run(transport="sse", port=8000, host="0.0.0.0")