MCP FHIR Server

""" Zus Health FHIR API Extensions This module contains Zus-specific functionality for working with Zus FHIR servers. These tools are optional and only relevant when working with Zus Health's FHIR API. For generic FHIR server operations, use the tools in server.py instead. """ from typing import Any import httpx import json import os # Import base FHIR configuration FHIR_BASE_URL = os.getenv("FHIR_BASE_URL", "http://localhost:8080/fhir") FHIR_AUTH_TOKEN = os.getenv("FHIR_AUTH_TOKEN", "") def get_zus_fhir_headers(builder_id: str | None = None) -> dict[str, str]: """ Get headers for Zus FHIR API requests. Args: builder_id: Zus customer/builder ID for multi-tenant access Returns: Dictionary of HTTP headers with Zus-specific additions """ headers = { "Content-Type": "application/fhir+json", "Accept": "application/fhir+json", } if FHIR_AUTH_TOKEN: headers["Authorization"] = f"Bearer {FHIR_AUTH_TOKEN}" if builder_id: headers["Zus-Account"] = builder_id return headers def calculate_name_similarity(patient_name: dict, search_first: str, search_last: str) -> float: """ Calculate similarity score between patient name and search terms. This helper function is used for intelligent name matching when multiple patients are returned from a search. Args: patient_name: FHIR name object from Patient resource search_first: First name being searched for search_last: Last name being searched for Returns: Similarity score between 0.0 and 1.0 (higher is better match) """ if not patient_name: return 0.0 # Extract name components given_names = patient_name.get("given", []) family_name = patient_name.get("family", "") # Convert to lowercase for case-insensitive comparison search_first_lower = search_first.lower().strip() search_last_lower = search_last.lower().strip() family_lower = family_name.lower().strip() # Calculate family name similarity (exact match gets highest score) family_score = 1.0 if family_lower == search_last_lower else 0.0 # Calculate given name similarity given_score = 0.0 if given_names: # Check for exact match in any given name for given in given_names: if given.lower().strip() == search_first_lower: given_score = 1.0 break # If no exact match, check for partial matches if given_score == 0.0: for given in given_names: given_lower = given.lower().strip() # Check for partial matches - allow shorter names to match longer ones # This allows "John" to match "Johnny" when there's no exact match if (search_first_lower in given_lower and len(search_first_lower) >= 3) or \ (given_lower in search_first_lower and len(given_lower) >= 3): given_score = 0.7 # Partial match score break # Weighted combination: family name is more important total_score = (family_score * 0.6) + (given_score * 0.4) return total_score async def get_patient_zus_upid( first_name: str, last_name: str, builder_id: str | None = None, is_http_method_allowed_fn=None, _allowed_methods_set=None ) -> str: """ Get the Zus UPID (Universal Patient ID) for a Patient resource from Zus FHIR server. This is a Zus-specific tool that searches for a Patient by first and last name, optionally filtered by builderID, then extracts the Zus UPID from the Patient's identifiers. **This tool is only relevant when working with Zus Health's FHIR API.** Args: first_name: Patient's first name last_name: Patient's last name builder_id: Optional builder ID (string) to filter the search is_http_method_allowed_fn: Function to check if HTTP method is allowed _allowed_methods_set: Set of allowed HTTP methods Returns: The Zus UPID value or an error message if not found """ # Check if GET method is allowed if is_http_method_allowed_fn and not is_http_method_allowed_fn("GET"): if _allowed_methods_set is not None: return f"Error: HTTP method GET is not allowed. Allowed methods: {', '.join(sorted(_allowed_methods_set))}" else: return "Error: Read operations are not allowed. Set FHIR_ALLOW_READ=true to enable." # Build search parameters search_params = { "name": f"{first_name} {last_name}" } if builder_id: search_params["builderID"] = builder_id url = f"{FHIR_BASE_URL}/Patient" try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get( url, params=search_params, headers=get_zus_fhir_headers(builder_id) ) if response.status_code == 200: bundle_data = response.json() entries = bundle_data.get("entry", []) if not entries: return f"Error: No Patient found with name '{first_name} {last_name}'" + (f" and builderID '{builder_id}'" if builder_id else "") # Look for Zus UPID in each patient zus_upid_system = "https://zusapi.com/fhir/identifier/universal-id" found_patients = [] for entry in entries: patient = entry.get("resource", {}) patient_id = patient.get("id", "unknown") identifiers = patient.get("identifier", []) patient_name = patient.get("name", [{}])[0] if patient.get("name") else {} # Look for Zus UPID identifier zus_upid = None for identifier in identifiers: if identifier.get("system") == zus_upid_system: zus_upid = identifier.get("value") break if zus_upid: # Calculate name similarity score similarity_score = calculate_name_similarity(patient_name, first_name, last_name) found_patients.append({ "patient_id": patient_id, "zus_upid": zus_upid, "name": patient_name, "similarity_score": similarity_score }) if not found_patients: return f"Error: No Zus UPID found for Patient(s) with name '{first_name} {last_name}'" + (f" and builderID '{builder_id}'" if builder_id else "") if len(found_patients) == 1: return f"Zus UPID: {found_patients[0]['zus_upid']}" else: # Multiple patients found, find the best match # Sort by similarity score (highest first) found_patients.sort(key=lambda x: x["similarity_score"], reverse=True) best_match = found_patients[0] best_score = best_match["similarity_score"] # If the best match has a good score (>= 0.5), return it as the primary result if best_score >= 0.5: name_parts = best_match["name"] display_name = "" if name_parts: given = name_parts.get("given", []) family = name_parts.get("family", "") display_name = f" {' '.join(given)} {family}".strip() result = f"Zus UPID: {best_match['zus_upid']} (Best match: {display_name})" # If there are other patients with decent scores, mention them other_matches = [p for p in found_patients[1:] if p["similarity_score"] >= 0.3] if other_matches: result += f"\n\nOther matches found:" for patient in other_matches[:3]: # Limit to top 3 other matches name_parts = patient["name"] display_name = "" if name_parts: given = name_parts.get("given", []) family = name_parts.get("family", "") display_name = f" {' '.join(given)} {family}".strip() result += f"\n- Patient ID: {patient['patient_id']} ({display_name}) -> Zus UPID: {patient['zus_upid']}" return result else: # No good matches, return all patients result = f"Found {len(found_patients)} Patient(s) with Zus UPID(s) (no clear name match):\n" for patient in found_patients: name_parts = patient["name"] display_name = "" if name_parts: given = name_parts.get("given", []) family = name_parts.get("family", "") display_name = f" {' '.join(given)} {family}".strip() result += f"- Patient ID: {patient['patient_id']} ({display_name}) -> Zus UPID: {patient['zus_upid']}\n" return result elif response.status_code == 400: try: error_data = response.json() return f"Invalid search parameters (400):\n{json.dumps(error_data, indent=2)}" except: return f"Invalid search parameters (400): {response.text}" elif response.status_code == 401: return ( "Error: Authentication failed. Check FHIR_AUTH_TOKEN configuration." ) elif response.status_code == 403: return "Error: Authorization failed. Insufficient permissions to search Patient resources." elif response.status_code == 404: return ( "Error: Patient resource type not found or not supported." ) else: return f"Server error ({response.status_code}):\n{response.text}" except httpx.TimeoutException: return f"Error: Request to FHIR server timed out after 30 seconds" except httpx.RequestError as e: return f"Error connecting to FHIR server: {str(e)}\nVerify FHIR_BASE_URL is correct: {FHIR_BASE_URL}" except Exception as e: return f"Unexpected error: {str(e)}"