search_fhir

Instructions

Implementation Reference

• src/azure_fhir_mcp_server/server.py:603-686 (handler)

  Primary implementation of the 'search_fhir' tool: decorated with @mcp.tool() for MCP registration, executes FHIR search API requests, extracts results from Bundle, attaches catalog metadata including inputSchema for validation.

  @mcp.tool()
  async def search_fhir(resource_type: str, search_params: Optional[Dict[str, Any]] = None, ctx: Optional[Context] = None) -> List[Dict[str, Any]]:
      """
      Search FHIR resources using comprehensive Azure FHIR search capabilities.
  
      Supports resource-specific and common search parameters, modifiers, prefixes, chained searches, 
      and result management. Returns paginated results in FHIR searchset bundles.
  
      Key Features:
      • Resource-specific and common search parameters (_id, _lastUpdated, _tag, etc.)
      • Search modifiers (:missing, :exact, :contains, :text, :not, etc.)
      • Prefixes for ordered parameters (gt, lt, ge, le, etc.)
      • Chained searches (e.g., Encounter?subject:Patient.name=Jane)
      • Reverse chained searches using _has parameter
      • Include and revinclude searches (_include, _revinclude)
      • Result parameters (_count, _sort, _elements, _summary, _total)
      • Composite search parameters for complex queries
      • Pagination support with configurable page sizes (max 1000)
  
      Args:
          resource_type: FHIR resource type to search (e.g., 'Patient', 'Observation', 'Condition')
          search_params: Dictionary of FHIR search parameters. Common examples:
              • {"name": "Smith", "_count": 50} - Search patients by name, limit 50 results
              • {"birthdate": "gt1990-01-01", "_sort": "birthdate"} - Patients born after 1990, sorted
              • {"identifier": "12345", "_include": "Patient:general-practitioner"} - Include GP
              • {"code": "77386006", "_include": "Observation:subject"} - Pregnancy observations with patients
              • {"_lastUpdated": "gt2024-01-01"} - Resources updated after date
          ctx: MCP Context for logging and progress reporting
  
      Returns:
          List of matching FHIR resources extracted from searchset Bundle
      """
      if search_params is None:
          search_params = {}
  
      if resource_type not in FHIR_RESOURCES:
          error_msg = f"Invalid resource type: {resource_type}"
          logger.error(error_msg)
          if ctx:
              await ctx.error(error_msg)
          return []
  
      try:
          token = await get_fhir_token(ctx)
          headers = {"Authorization": f"Bearer {token}"}
  
          # Build URL with search parameters
          url = f"{FHIR_URL}/{resource_type}"
  
          if ctx:
              await ctx.info(f"Searching {resource_type} with params: {search_params}")
  
          response = requests.get(url, headers=headers,
                                  params=search_params, timeout=30)
  
          if response.status_code != 200:
              error_msg = f"Search failed: {response.status_code} - {response.text}"
              logger.error(error_msg)
              if ctx:
                  await ctx.error(error_msg)
              return []
  
          bundle = cast(Dict[str, Any], response.json())
  
          # Extract resources from Bundle
          if "entry" in bundle:
              results = [entry["resource"] for entry in bundle["entry"]]
              if ctx:
                  await ctx.info(f"Found {len(results)} {resource_type} resources")
              return results
          else:
              if ctx:
                  await ctx.info(f"No {resource_type} resources found")
              return []
  
      except (requests.RequestException, ValueError, KeyError) as e:
          error_msg = f"Error during search: {str(e)}"
          logger.error(error_msg)
          if ctx:
              await ctx.error(error_msg)
          return []
  
  
  _attach_tool_metadata("search_fhir", search_fhir)

• src/azure_fhir_mcp_server/server.py:686-686 (registration)

  Attaches tool metadata (name, description, inputSchema) from catalog.yaml to the search_fhir function for FastMCP discovery.

  _attach_tool_metadata("search_fhir", search_fhir)

• tests/test_fastmcp_metadata.py:30-35 (schema)

  Test fixture loading catalog.yaml which defines inputSchema for 'search_fhir' tool (not the primary schema location, but confirms existence). Actual schema in src/azure_fhir_mcp_server/config/catalog.yaml (file not directly accessible).

  catalog_path = Path(__file__).parent.parent / "src" / \
      "azure_fhir_mcp_server" / "config" / "catalog.yaml"
  
  with open(catalog_path, 'r', encoding='utf-8') as f:
      catalog_data = yaml.safe_load(f)

• src/azure_fhir_mcp_server/server.py:227-227 (helper)

  Computes list of valid FHIR resource types from catalog for validating search_fhir resource_type parameter.

  FHIR_RESOURCES = _compute_resource_types(RESOURCE_METADATA)