FHIR MCP Server

Instructions

Implementation Reference

• src/fhir_mcp_server/server.py:697-764 (handler)

  The 'get_user' tool implementation is defined as an MCP tool in 'server.py'. It authenticates the user, retrieves their FHIR resource using the ID token, and formats the result as a user profile.

  @mcp.tool(
      description=(
          "Retrieves the authenticated user's FHIR profile. "
          "Use this tool when you need to access the current user's demographic and contact details."
      )
  )
  async def get_user() -> Annotated[
      list[Dict[str, Any]] | Dict[str, Any],
      Field(
          description="A dictionary containing the authenticated user's demographic information such as 'id', 'name', and 'birthDate'."
      ),
  ]:
      try:
          logger.debug("Retrieving authenticated user's profile.")
  
          # Validate user authentication
          user_token = await get_user_access_token()
          if not user_token:
              logger.debug("Unauthorized access attempt to get_me endpoint.")
              return {}
  
          # Retrieve token metadata
          token_metadata = server_provider.token_metadata_mapping.get(
              user_token.access_token
          )
          if not token_metadata:
              logger.debug("Token metadata not found for authenticated user.")
              return {}
  
          # Extract ID token information
          id_token = token_metadata.get_id_token()
          if not id_token:
              logger.debug("ID token not found in token metadata.")
              return {}
  
          # Validate resource identifiers
          resource_id = id_token.resource_id
          resource_type = id_token.resource_type
  
          if not resource_id or not resource_type:
              logger.debug("Resource ID or type missing from ID token.")
              return {}
  
          logger.debug(f"Fetching FHIR resource: {resource_type}/{resource_id}")
  
          # Fetch user's FHIR resource
          client: AsyncFHIRClient = await get_async_fhir_client()
          resource: Dict[str, Any] = await client.get(
              resource_type_or_resource_or_ref=resource_type, id_or_ref=resource_id
          )
  
          # Build response with only available fields
          profile: Dict[str, Any] = build_user_profile(resource)
  
          logger.debug(
              f"Successfully retrieved profile for user: {resource_type}/{resource_id}"
          )
          return profile
  
      except ValueError as ex:
          logger.exception(
              "Authorization error occurred while reading user resource. Caused by, ",
              exc_info=ex,
          )
          return await get_operation_outcome(
              code="forbidden",
              diagnostics="The user does not have the rights to perform read operations.",
          )