AI¶

ai_dto ¶

ProcessDSDto ¶

Bases: BaseModel

Data Transfer Object for processing discharge summary requests.

Attributes:

Name	Type	Description
`case_type`	`str`	The type of case (e.g., `OPConsultation`, `DischargeSummary`). Serialized as `caseType`.
`files`	`Optional[list[str]]`	List of file paths or file identifiers to be processed.
`encrypted_data`	`Optional[str]`	Pre-encrypted data, if available.
`public_key`	`Optional[str]`	Public key for encryption, if required.
`encounter_id`	`Optional[str]`	Unique identifier for the encounter, serialized as `encounterId`, used for partial upload. Example: "enc12345"
`date`	`Optional[str]`	The date of the request, usually in ISO 8601 format. Example: "2025-09-23T12:30:00Z"
`Example`	`Optional[str]`

DischargeSummaryResponse ¶

Bases: BaseModel

Represents the response for a discharge summary generation request.

Attributes:

Name	Type	Description
`id`	`str`	Unique identifier for the discharge summary.
`discharge_summary`	`Optional[dict[str, Any]]`	The generated discharge summary content, if available.
`extracted_data`	`dict[str, Any]`	Extracted clinical data from the input.
`fhir_bundle`	`dict[str, Any]`	FHIR-compliant bundle generated from the case data.

GenerateFhirBundleDto ¶

Bases: BaseModel

Data Transfer Object for generating a FHIR bundle.

Attributes:

Name	Type	Description
`case_type`	`str`	The type of case (e.g., `OPConsultation`, `DischargeSummary`). Serialized as `caseType`.
`enable_extraction`	`bool`	Flag to enable data extraction from the provided documents.
`document_references`	`list[str]`	List of document references to be included in the bundle.
`record_id`	`Optional[str]`	Identifier for the record, if available. Serialized as `recordId`.
`extracted_data`	`Optional[dict[str, Any]]`	Extracted clinical data for the bundle.
`encrypted_data`	`Optional[str]`	Pre-encrypted data, if available.
`public_key`	`Optional[str]`	Public key for encryption, if required.

FhirBundleResponse ¶

Bases: RootModel[dict[str, Any]]

Represents the response for a FHIR bundle generation request.

The root model is a dictionary containing the FHIR bundle content.

AiService ¶

AiService(config: ClientConfig)

Bases: BaseService

AiService provides a high-level interface for AI-powered healthcare document generation.

This service enables SDK users to interact with CareStack AI endpoints for:¶

Generating discharge summaries (generate_discharge_summary)
Generating FHIR bundles (generate_fhir_bundle)

Key Features

Validates input data using Pydantic models (ProcessDSDto, GenerateFhirBundleDto)
Handles encryption of sensitive data before transmission
Provides robust error handling and logging for all operations

Methods:

Name	Description
`generate_discharge_summary`	Generates a discharge summary from provided case data.
`generate_fhir_bundle`	Generates a FHIR-compliant bundle from provided case data.

Parameters:

Name	Type	Description	Default
`config`	`ClientConfig`	API credentials and settings for service initialization.	required

Raises:

Type	Description
`EhrApiError`	For validation, API, or unexpected errors during operations.

Example Usage

config = ClientConfig(
    api_key="your_api_key",
)
service = AiService(config)
summary = await service.generate_discharge_summary({...})
bundle = await service.generate_fhir_bundle({...})

generate_case_summary `async` ¶

generate_case_summary(process_data: dict[str, Any], case_type: str) -> DischargeSummaryResponse

Generic case summary generator for all case types (DischargeSummary, OpConsultation, Radiology).

generate_fhir_bundle `async` ¶

generate_fhir_bundle(data: dict[str, Any]) -> dict[str, Any]

Generates a FHIR bundle based on the provided data.

This method validates and processes the input, encrypts extracted data if necessary, and sends it to the AI API to generate a FHIR-compliant bundle. Use this method to automate generation of interoperable FHIR bundles from structured clinical data.

Attributes:

Name	Type	Description
`generate_fhir_bundle_data`	`GenerateFhirBundleDto`	GenerateFhirBundleDto containing required inputs for generating the bundle.

Args:¶

generate_fhir_bundle_data (dict): Dictionary containing:
    - caseType (str): Type of the case (`inpatient`, `outpatient`, etc.)
    - enableExtraction (bool): Flag to enable data extraction from provided documents.
    - documentReferences (list[str]): List of document references to include in the bundle.
    - recordId (Optional[str]): Unique identifier for the record.
    - extractedData (Optional[dict]): Structured clinical data to generate the bundle.
    - encryptedData (Optional[str]): If provided, skips encryption and uses this encrypted payload.
    - publicKey (Optional[str]): Required if using `extractedData` without pre-encryption.

Returns:¶

dict[str, Any]: The generated FHIR-compliant bundle.
    Example:
    {
        "resourceType": "Bundle",
        "type": "document",
        "entry": [
            {
                "resource": {
                    "resourceType": "Patient",
                    "id": "123",
                    ...
                }
            },
            ...
        ]
    }

Raises:

Type	Description
`ValidationError`	If input fails Pydantic model validation.
`EhrApiError`	Raised on API failure (status 400/422/500).
`ValueError`	If both `extractedData` and `encryptedData` are missing.

Example (Success):¶

response = await service.generate_fhir_bundle({
    "caseType": "DischargeSummary",
    "enableExtraction": True,
    "documentReferences": ["doc123", "doc456"],
    "recordId": "rec-789",
    "extractedData": {
        "patientName": "John Doe",
        "diagnosis": "Hypertension",
        "treatment": "Medication and lifestyle changes"
    },
    "publicKey": "-----BEGIN PUBLIC KEY-----...",
})

print(response)

Output will look like:

{
    "resourceType": "Bundle",
    "entry": [
        {"resource": {"resourceType": "Patient", "id": "123", ...}},
        ...
    ]
}

Example (Validation Failure):¶

await service.generate_fhir_bundle({
    "caseType": "DischargeSummary"
})
# Raises EhrApiError: No extracted data or encrypted data provided (422)

generate_discharge_summary `async` ¶

generate_discharge_summary(process_data: dict[str, Any]) -> DischargeSummaryResponse

Generate a discharge summary using AI based on the provided case data.

This method validates and processes the input data, encrypts it if necessary, and calls the AI API to generate a discharge summary. Use this to automate the creation of discharge summaries from structured/unstructured clinical data.

Parameters:

Name	Type	Description	Default
`process_data`	`dict[str, Any]`	Dictionary containing required inputs such as case type, files, encryptedData, and publicKey.	required

Returns:

Name	Type	Description
`DischargeSummaryResponse`	`DischargeSummaryResponse`	Contains: - dischargeSummary (Optional[dict]): AI-generated discharge summary content - extractedData (dict): Extracted structured clinical content - fhirBundle (dict): FHIR-compliant bundle based on extracted data

Raises:

Type	Description
`ValueError`	If both 'files' and 'encryptedData' are missing.
`ValidationError`	If input fails schema validation.
`EhrApiError`	Raised when the API returns error status codes (400, 422, 500).

Example (Success):

response = await service.generate_discharge_summary({
    "files": ["file123.pdf"],
})

print(response.dischargeSummary["diagnosis"])

Example (Validation Failure):

await service.generate_discharge_summary({
    "caseType": "DISCHARGE_SUMMARY"
})
# Raises EhrApiError: No files or encrypted data provided (422)

generate_op_consultation_summary `async` ¶

generate_op_consultation_summary(process_data: dict[str, Any]) -> DischargeSummaryResponse

Generate an OP consultation summary using AI based on the provided case data.

This method validates and processes the input, encrypts it if necessary, and calls the AI API to generate an outpatient consultation summary.

Parameters:

Name	Type	Description	Default
`process_data`	`dict[str, Any]`	Dictionary containing required inputs such as case type, files, encryptedData, and publicKey.	required

Returns:

Name	Type	Description
`DischargeSummaryResponse`	`DischargeSummaryResponse`	Contains: - dischargeSummary (Optional[dict]): AI-generated OP consultation summary - extractedData (dict): Extracted structured consultation data - fhirBundle (dict): FHIR-compliant bundle based on extracted data

Raises:

Type	Description
`ValueError`	If both 'files' and 'encryptedData' are missing.
`ValidationError`	If input fails schema validation.
`EhrApiError`	Raised when the API returns error status codes (400, 422, 500).

Example

response = await service.generate_op_consultation_summary({
    "files": ["consultation_notes.pdf"],
})

generate_radiology_summary `async` ¶

generate_radiology_summary(process_data: dict[str, Any]) -> RadiologySummaryResponse

Generate a radiology/diagnostic report summary using AI.

This method validates and processes radiology/diagnostic data, encrypts it if necessary, and calls the AI API to generate a diagnostic summary report.

Parameters:

Name	Type	Description	Default
`process_data`	`dict[str, Any]`	Dictionary containing required inputs such as files, encryptedData, and publicKey.	required

Returns:

Name	Type	Description
`DischargeSummaryResponse`	`RadiologySummaryResponse`	Contains: - dischargeSummary (Optional[dict]): AI-generated radiology summary - extractedData (dict): Extracted structured diagnostic data - fhirBundle (dict): FHIR-compliant bundle based on extracted data

Raises:

Type	Description
`ValueError`	If both 'files' and 'encryptedData' are missing.
`EhrApiError`	Raised when the API returns error status codes (400, 422, 500).

partial_upload_for_discharge_summary `async` ¶

partial_upload_for_discharge_summary(process_ds_dto: ProcessDSDto) -> Dict[str, Any]

Perform a partial upload for discharge summary generation.

This method validates the provided data, performs encryption if necessary, constructs the appropriate payload, and makes an API call to trigger a partial discharge summary upload.

Parameters:

Name	Type	Description	Default
`process_ds_dto`	`ProcessDSDto`	DTO containing encounter details, files, encryption data, and metadata required for the upload.	required

Returns:

Type	Description
`Dict[str, Any]`	Dict[str, Any]: The JSON response from the AI service after
`Dict[str, Any]`	processing the partial upload.

Raises:

Type	Description
`ValueError`	If neither `encrypted_data` nor `files` are provided.
`EhrApiError`	If an error occurs during encryption or the API call.

Example

dto = ProcessDSDto( ... encounter_id="enc-12345"(not required for first time), ... files=[{"file1": "lab_report.pdf", "file2": "..."}], ... date=2025-10-01T10:00:00Z, ... ) result = await ai_service.partial_upload_for_discharge_summary(dto) print(result) { "id": "0c4cce09-b7bd-4255-8182-b2c3586ere001c9", "dischargeSummary": {}, "extractedData": { "Patient Details": { "Name": "Mr. Sri Ram", "Age": "16 Year(s) 6 Month(s)", "Sex": "Male", ..... }
}, }

trigger_discharge_summary `async` ¶

trigger_discharge_summary(encounter_id: str) -> Any

        Trigger discharge summary generation for a given encounter ID.

        This method sends a PUT request to the AI service endpoint to initiate
        discharge summary generation for the specified encounter. It ensures the
        `encounter_id` is valid before making the request.

        Args:
            encounter_id (str): The unique identifier of the encounter.
                Must be a non-empty string.

        Returns:
            Dict[str, Any]: The JSON response from the AI service containing
            the status of the discharge summary generation.

        Raises:
            ValueError: If `encounter_id` is missing or blank.
            EhrApiError: If an error occurs during the API request.

        Example:
            >>> result = await ai_service.trigger_discharge_summary("enc-12345")
            >>> print(result)
            {
                "data": "{

"patientDetails": { "Name": "Mr. Sri Ram", "Age": "16 years 6 months", "Gender": "Male", .........."}", "message": "Discharge for record enc-12345 generated successfully" }

generate_careplan `async` ¶

generate_careplan(process_ds_dto: ProcessDSDto) -> Dict[str, Any]

        Generate a patient care plan from provided files or encrypted data.

        This method validates the input data, encrypts it if required, and sends the
        payload to the AI service endpoint for care plan generation. It handles
        validation errors, encryption failures, and unexpected errors gracefully.

        Args:
            process_ds_dto (ProcessDSDto):
                The input DTO containing files, encrypted data, and optional encryption keys.

        Returns:
            Dict[str, Any]: 
                A dictionary containing the AI-generated care plan response.

        Raises:
            EhrApiError: If the API call fails (status code 500)
            ValueError: Raised internally for missing `files` or `encrypted_data` before being
                wrapped as an `EhrApiError`.

        Example Input if files are provided:
            >>> dto = ProcessDSDto(
            ...     files=["file1.pdf", "file2.pdf"],
            ... )
            >>> response = await ai_service.generate_careplan(dto)

        Example Input if encrypted data is provided:
            >>> dto = ProcessDSDto(
            ...     encrypted_data="encrypted_payload_string",
            ...     public_key="-----BEGIN PUBLIC KEY-----

... -----END PUBLIC KEY-----" ... ) response = await ai_service.generate_careplan(dto)

        Example Output:
            {
                "id": "a1956730-546534265f-4c2d-8902-a34fe3645886b24",
                "carePlan": {
                    "patientDetails": {
                        "name": "Ram sai",
                        "age": "75Y(s) 6M(s) 30D(s)",
                        "sex": "Male",
                        "uhid": "MRKO252643654325015739",
                        "visitId": "IPKO243534005789",
                        "address": "Mumbai",
                        "contactNumber": "9876543210"
                    },
                    "doctorDetails": [
                        {
                            "name": "DR. BALAKRISHNA N",
                            "designation": "MBBS, MD, DM (Cardiology), Sr. Consultant Cardiologist.",
                            "department": "CARDIOLOGY"
                        },
                        {
                            "name": "DR. SHYAM SINGA ROY. G",
                            "designation": "MBBS, MD (General Medicine), DM (Neurology), Consultant Neurologist",
                            "department": "NEUROLOGY"
                        }
                    ],
                    ....
                }
            }

AiUtilities ¶

AiUtilities(config: ClientConfig)

Bases: BaseService

Utility class for cryptographic operations required by AI services, such as encryption of payloads and loading public keys from X.509 certificates.

This class is intended to be used internally by AI-related service classes to ensure secure transmission of sensitive healthcare data to backend APIs.

Key Features

Loads RSA public keys from PEM-encoded X.509 certificates and converts them to JWK format.
Encrypts arbitrary payloads using JWE (JSON Web Encryption) with RSA-OAEP-256 and AES-GCM.
Handles environment variable management for encryption keys.
Provides robust error handling and logging for all cryptographic operations.

Methods:

Name	Description
`load_public_key_from_x509_certificate`	Loads an RSA public key from a PEM-encoded X.509 certificate and returns it as a JWK dictionary.
`encryption`	Encrypts a payload dictionary using JWE with an RSA public key loaded from the `ENCRYPTION_PUBLIC_KEY` environment variable.

Example usage

utils = AiUtilities()
encrypted = await utils.encryption({"foo": "bar"})

encryption `async` ¶

encryption(payload: dict[str, Any], public_key: Optional[str] = None) -> dict

Calls the remote NestJS service to encrypt a payload using JWE.

This method sends the plain payload dictionary to the /encrypt endpoint, along with an optional public key.

Args:¶

payload (Dict[str, Any]): The data to be encrypted.
public_key (Optional[str]): An optional PEM-encoded public key to use for encryption.

Returns:¶

str: The JWE compact-serialized encrypted string returned by the service.

Raises:¶

RuntimeError: If the request to the encryption service fails or the service returns an error status.

Example

utils = AiUtilities()
encrypted_payload = await utils.encryption({"foo": "bar"})
print(encrypted_payload)
# Output: eyJhbGciOiJSU0EtT0FFUC0yNTYiLCJlbmMiOiJBMTI4R0NN...

decryption `async` ¶

decryption(encrypted_data: str, private_key: Optional[str] = None) -> dict

Calls the remote NestJS service to decrypt a JWE payload.

This method sends an encrypted JWE string to the /decrypt endpoint. It mirrors the structure of the NestJS controller, accepting the encrypted payload and an optional private key.

Args:¶

encrypted_data (str): The JWE compact-serialized string to be decrypted.
private_key (Optional[str]): An optional PEM-encoded private key. If provided,
    it will be sent to the decryption service. Otherwise, the service
    will use its default key.

Returns:¶

Any: The decrypted payload returned from the service.

Raises:¶

RuntimeError: If the request to the decryption service fails or the
              service returns an error status.

Example

utils = AiUtilities()
encrypted_jwe = "eyJhbGciOiJSU0EtT0FFUC0yNTYiLCJlbmMiOiJBMTI4R0NN..."
decrypted_payload = await utils.decryption(encrypted_jwe)
print(decrypted_payload)
# Output from service: {'patientName': 'John Doe', 'diagnosis': 'Hypertension'}

AI¶

ai_dto ¶

ProcessDSDto ¶

DischargeSummaryResponse ¶

GenerateFhirBundleDto ¶

FhirBundleResponse ¶

AiService ¶

This service enables SDK users to interact with CareStack AI endpoints for:¶

generate_case_summary async ¶

generate_fhir_bundle async ¶

Args:¶

Returns:¶

Example (Success):¶

Example (Validation Failure):¶

generate_discharge_summary async ¶

generate_op_consultation_summary async ¶

generate_radiology_summary async ¶

partial_upload_for_discharge_summary async ¶

trigger_discharge_summary async ¶

generate_careplan async ¶

AiUtilities ¶

encryption async ¶

Args:¶

Returns:¶

Raises:¶

decryption async ¶

Args:¶

Returns:¶

Raises:¶

generate_case_summary `async` ¶

generate_fhir_bundle `async` ¶

generate_discharge_summary `async` ¶

generate_op_consultation_summary `async` ¶

generate_radiology_summary `async` ¶

partial_upload_for_discharge_summary `async` ¶

trigger_discharge_summary `async` ¶

generate_careplan `async` ¶

encryption `async` ¶

decryption `async` ¶