Examples

This section contains examples of fully executable flows. We recommend using one of these examples as a template for building a flow to fit your exact business process.

Most examples marked contain a release and/or sample documents necessary to execute the flow in your instance of the Hyperscience Platform. You can still build your own release and test with additional documents.

V44

Name	Description	Release	Sample Docs	Download
Document Processing Starter V44	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v44.zip
OnError flow V44	Example implementation of an OnError flow that is triggered once the original flow fails.			on_error_flow_v44.zip

Document Processing Starter V44

The basic document processing flow. From V38 on, the IDP Document processing is split into a wrapper flow (IDP) and a subflow (IDP Core). The package includes:

• IDP Core Flow - a read-only subflow containing the steps to process documents into machine-readable output

• Document Processing Flow - a wrapper around the IDP Core flow, used for managing input and output connector configuration and document processing workflow settings

• IDP Submission Notify Flow - A flow that can be configured for some blocks of the IDP Core Flow, used to send notifications to external systems when a submission has been created or is waiting for supervision

• Submission-aware On Error Flow - When the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered. The submission-aware on-error flow calls a helper subflow which generates the error message to be sent to the output blocks.

• Submission-aware On Error Subflow - The submission-aware on-error subflow gathers relevant submission data and combines it with data about the failed flow run. In case this OnError implementation is used on a non-IDP flow (i.e. a submissionless flow), submission data will be omitted

In V42.3+ a new Document Processing with ORCA Subflow is available for Advanced and Premium license.

• IDP ORCA Flow - a read-only subflow similar to IDP Core Flow but ORCA is used for field identification. _Note: Currently this flow does not support Structured forms._

• ORCA QA Flow - Quality assurance for ORCA flow. Meant to be used together with IDP ORCA Flow to spawn QA tasks for human review. Configurable through IDP ORCA Flow settings.

The package includes also release and sample documents for processing Form W-9s.

Download idp_flow_v44.zip

Document Processing Sample

from uuid import UUID

from flows_sdk.error_handling import ErrorHandling
from flows_sdk.flows import Flow
from flows_sdk.package_utils import export_flow

from flows_sdk.implementations.idp_v44 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v44.idp_blocks import IDPCoreBlock
from flows_sdk.implementations.idp_v44.idp_values import (
    DEFAULT_IDP_CORE_IDENTIFIER,
    DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER,
    IdpWorkflowConfig,
    IDPWrapperManifest,
    Settings,
    get_idp_wf_config,
)

IDP_UUID = 'feac0001-c17e-42e8-b367-0000feac0001'
IDP_IDENTIFIER = 'DOCUMENT_PROCESSING_EXAMPLE'
IDP_TITLE = 'Document Processing - flows-sdk example'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:
    # By default, all the notification flows are set to `None`.
    # I.e. the system will send no intermediate notifications for submission change events.
    # So let's specify the notification flows that we want to use explicitly here:
    idp_wf_config.notification_workflow_initialization = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_classification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_identification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_transcription = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_flexible_extraction = (
        DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    )

    idp_flow = IDPCoreBlock(idp_wf_config, identifier=DEFAULT_IDP_CORE_IDENTIFIER)

    outputs = idp_blocks.IDPOutputsBlock(inputs={'submission': idp_flow.output('submission')})

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=IDP_TITLE,
        description='Use this flow to manage your input and output connector configuration and '
        'document processing workflow settings.',
        manifest=IDPWrapperManifest(IDP_IDENTIFIER),
        output={'submission': idp_flow.output('submission')},
        triggers=idp_values.IDPTriggers(),
        input={
            Settings.LayoutReleaseUuid: None,
        },
        blocks=[
            idp_flow,
            outputs,
        ],
        error_handling=ErrorHandling(on_error_flow=None),
    )


def entry_point_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Core V44 Sample

from uuid import UUID

from flows_sdk.blocks import Routing
from flows_sdk.flows import Flow
from flows_sdk.package_utils import export_flow

from flows_sdk.implementations.idp_v44 import additional_blocks, idp_blocks
from flows_sdk.implementations.idp_v44.idp_blocks.defaults import IDP_CORE_FLOW_ROLE
from flows_sdk.implementations.idp_v44.idp_values import IDPCoreManifest
from flows_sdk.implementations.idp_v44.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED
from flows_sdk.utils import str_to_deterministic_uuid_4

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into IDPCoreBlock(identifier=...) within your flow.
IDP_CORE_IDENTIFIER = f'IDP_CORE_{FLOWS_VERSION}_EXAMPLE'
IDP_CORE_UUID = str(str_to_deterministic_uuid_4(IDP_CORE_IDENTIFIER))
IDP_CORE_TITLE = f'Document Processing Subflow {FLOWS_VERSION_DOTTED} - flows-sdk example'
NO_LABEL = 'No'
YES_LABEL = 'Yes'


def idp_workflow() -> Flow:
    file_filter = additional_blocks.FileFilterBlock(reference_name='file_filter')

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        trigger=file_filter.output('result.filtered_trigger'),
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        nlc_metadata=machine_classification.output('nlc_metadata'),
    )

    manual_classification_routing = Routing(
        reference_name='manual_classification_routing',
        decision=machine_classification.output('needs_manual_classification'),
        title='Is Manual Classification needed?',
        branches=[
            Routing.Branch(
                case='false', label=NO_LABEL, blocks=[], output='machine_classification'
            ),
        ],
        default_branch=Routing.DefaultBranch(label=YES_LABEL, blocks=[manual_classification]),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification_routing = Routing(
        reference_name='manual_identification_routing',
        decision=machine_identification.output('needs_manual_identification'),
        title='Is Manual Identification needed?',
        branches=[
            Routing.Branch('false', blocks=[], label=NO_LABEL, output='machine_identification'),
        ],
        default_branch=Routing.DefaultBranch(blocks=[manual_identification], label=YES_LABEL),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription_routing = Routing(
        reference_name='manual_transcription_routing',
        decision=machine_transcription.output('needs_manual_transcription'),
        title='Is Manual Transcription needed?',
        branches=[
            Routing.Branch('false', blocks=[], label=NO_LABEL, output='machine_transcription'),
        ],
        default_branch=Routing.DefaultBranch(blocks=[manual_transcription], label=YES_LABEL),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction_routing = Routing(
        reference_name='flexible_extraction_routing',
        decision=machine_transcription.output('needs_flexible_extraction'),
        title='Is Flexible Extraction needed?',
        branches=[
            Routing.Branch(
                'false', blocks=[], label=NO_LABEL, output='manual_transcription_routing'
            ),
        ],
        default_branch=Routing.DefaultBranch(blocks=[flexible_extraction], label=YES_LABEL),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction_routing.output('submission'),
    )

    document_rendering = idp_blocks.DocumentRendererBlock(
        reference_name='document_rendering',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=document_rendering.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    return Flow(
        uuid=UUID(IDP_CORE_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=IDP_CORE_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output.',
        manifest=IDPCoreManifest(
            flow_identifier=IDP_CORE_IDENTIFIER,
            roles=['read_only', 'supporting', IDP_CORE_FLOW_ROLE],
        ),
        output={'submission': submission_complete.output('submission')},
        blocks=[
            file_filter,
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification_routing,
            machine_identification,
            manual_identification_routing,
            machine_transcription,
            manual_transcription_routing,
            flexible_extraction_routing,
            reprocessing,
            document_rendering,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Submission Notify Sample

from pathlib import Path
from uuid import UUID

from flows_sdk.blocks import Outputs
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import icon_path_to_base64, str_to_deterministic_uuid_4, workflow_input

BASE_DIR = Path(__file__).parents[0]

IDP_SUBMISSION_NOTIFY_IDENTIFIER = 'IDP_SUBMISSION_NOTIFY_EXAMPLE'
IDP_SUBMISSION_NOTIFY_UUID = str(str_to_deterministic_uuid_4(IDP_SUBMISSION_NOTIFY_IDENTIFIER))
IDP_SUBMISSION_NOTIFY_TITLE = 'Submission State Notifications - flows-sdk example'
IDP_OUTPUT_ROLE = 'idp_output'
# The title of the Document Processing flow in document_processing_flow.py.
IDP_TITLE = 'Document Processing - flows-sdk example'


class SubmissionNotifyInputKeys:
    submission: str = 'submission'


class SubmissionNotifyManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['notifications', 'supporting'],
            input=[
                Parameter(
                    name=SubmissionNotifyInputKeys.submission,
                    type='Submission',
                    title='Submission Object',
                    ui={
                        'hidden': True,
                    },
                )
            ],
            ui={
                'icon': icon_path_to_base64(str(BASE_DIR / 'submission_notify_icon.png')),
            },
        )


def idp_submission_notify_workflow() -> Flow:
    return Flow(
        title=IDP_SUBMISSION_NOTIFY_TITLE,
        owner_email='flows-sdk@hyperscience.ai',
        description=(
            'Send notifications to external systems when a submission has been '
            'created or is waiting for supervision. \n'
            f'If the "{IDP_TITLE}" flow is live, this flow '
            'must also be live, but it can be empty.'
        ),
        manifest=SubmissionNotifyManifest(IDP_SUBMISSION_NOTIFY_IDENTIFIER),
        blocks=[
            Outputs(
                reference_name='outputs',
                title='Outputs',
                description=(
                    'Send submission data to external systems when a submission '
                    'has been created or is waiting for supervision'
                ),
                role_filter=[IDP_OUTPUT_ROLE],
                input_template={
                    'submission': {'id': workflow_input('submission.id')},
                    'enabled': True,
                },
                blocks=[],
            )
        ],
        uuid=UUID(IDP_SUBMISSION_NOTIFY_UUID),
        input={},
        output={},
    )


def entry_point_workflow() -> Flow:
    return idp_submission_notify_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

Submission-aware On Error Flow Sample

from uuid import UUID

from flows_sdk.blocks import Block
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v44.additional_blocks import JSONOutputsBlock
from flows_sdk.package_utils import export_flow
from flows_sdk.types import CompatibilitySpec
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER = 'SUBMISSION_AWARE_ON_ERROR_FLOW_3_EXAMPLE'
SUBMISSION_AWARE_ON_ERROR_FLOW_UUID = str(
    str_to_deterministic_uuid_4(SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER)
)
SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE = (
    'On-Error with included Submission data V3 - flows-sdk example'
)
SUBMISSION_AWARE_ON_ERROR_DESCRIPTION = (
    'Flow designed to be used as an "on-error" handler for submission-based flows.'
    '\n\n'
    'With the on-error flow configured, when the original flow fails for any reason, and all '
    'the configured auto-retries have been exhausted, the on-error flow will be triggered.'
    '\n\n'
    'The input it receives is the ID of the failed flow run, which can be used to query '
    'additional information about the failure.'
    '\n\n'
    'If you need more complex logic in this flow please consult the documentation of Flows SDK '
    '(https://flows-sdk.hyperscience.ai) and the available public APIs '
    '(https://docs.hyperscience.ai). Note that the custom logic should be implemented in the '
    'On-Error Subflow, so that it can be safely updated without losing the configured Output '
    'blocks. The Output blocks will receive the output of the On-Error Subflow as input.'
)

class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)
SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER = 'SUBMISSION_AWARE_ON_ERROR_CORE_EXAMPLE'

def submission_aware_on_error_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)

    on_error_core = Block(
        reference_name='on_error_core',
        identifier=SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER,
        input={'failed_run_uuid': flow_run_uuid},
        compatibility_spec=CompatibilitySpec(filter_roles=['on_error_core']),
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': on_error_core.output()})

    return Flow(
        uuid=UUID(SUBMISSION_AWARE_ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE,
        description=SUBMISSION_AWARE_ON_ERROR_DESCRIPTION,
        manifest=Manifest(
            identifier=SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER,
            roles=['on_error', 'supporting'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER],
        ),
        output={},
        input={},
        blocks=[
            on_error_core,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return submission_aware_on_error_flow()


if __name__ == '__main__':
    export_flow(submission_aware_on_error_flow())

Submission-aware On Error Subflow Sample

from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v44.additional_blocks import HyperscienceRestApiBlock
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsTask
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER = 'SUBMISSION_AWARE_ON_ERROR_CORE_EXAMPLE'
SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_UUID = str(
    str_to_deterministic_uuid_4(SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER)
)
SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_TITLE = (
    'Retrieve error and submission information - flows-sdk example'
)
SUBMISSION_AWARE_ON_ERROR_CORE_DESCRIPTION = (
    'Core logic of the on-error flow. Compiles the data, relevant to the error. This includes '
    'metadata for the failed flow run and the JSON serialized submission data, if the flow run '
    'is linked to a submission. The flow run inputs, outputs and logs are excluded for '
    'performance reasons - their size is unpredictable.'
    '\n\n'
    'In this On-Error Subflow implementation, we fetch the failed flow run, alongside with the '
    'full submission metadata (if the submission ID was found). The data format of the output '
    'conforms to the schema documented here: https://docs.hyperscience.ai/#flows-runs and here: '
    'https://docs.hyperscience.ai/#submissions, under the respective keys "flow_run" and '
    '"submission".'
    '\n\n'
    'Any custom error handling logic should be added here. The main on-error flow should only be '
    'used for configuring the outputs in the Outputs block, i.e. where the compiled on-error data '
    'should be sent to.'
)


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)


def submission_aware_on_error_core_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        title='Get Flow Run',
        description='Get the full metadata of the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{flow_run_uuid}/summary',
    )

    def _get_correlation_id(_hs_task: HsTask) -> dict:
        return {'correlation_id': _hs_task.correlation_id}

    get_correlation_id_block = PythonBlock(
        reference_name='get_correlation_id',
        title='Get Correlation ID',
        description='Get the correlation ID for the failed Submission / Flow Run',
        code=_get_correlation_id,
    )

    fetch_submissions_block = HyperscienceRestApiBlock(
        reference_name='fetch_submission',
        title='Fetch Submission',
        description='Fetch the submission data for the failed flow run',
        method='GET',
        app_endpoint='/api/v5/submissions?debug=true'
        f'&correlation_id={get_correlation_id_block.output('correlation_id')}',
    )

    return Flow(
        uuid=UUID(SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_TITLE,
        description=SUBMISSION_AWARE_ON_ERROR_CORE_DESCRIPTION,
        manifest=Manifest(
            identifier=SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER,
            roles=['on_error_core', 'supporting', 'read_only'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER]
        ),
        output={
            'flow_run': flow_run_rest_api_block.output('result.data'),
            'submission': fetch_submissions_block.output('result.data'),
        },
        input={},
        blocks=[
            flow_run_rest_api_block,
            get_correlation_id_block,
            fetch_submissions_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return submission_aware_on_error_core_flow()


if __name__ == '__main__':
    export_flow(submission_aware_on_error_core_flow())

IDP ORCA V44 Sample

from uuid import UUID

from flows_sdk.blocks import Routing
from flows_sdk.flows import Flow
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

from flows_sdk.implementations.idp_v44 import additional_blocks, idp_blocks
from flows_sdk.implementations.idp_v44.additional_blocks import QASchedulerBlock
from flows_sdk.implementations.idp_v44.idp_blocks.defaults import IDP_CORE_FLOW_ROLE
from flows_sdk.implementations.idp_v44.idp_values import IDPCoreManifest, Settings
from flows_sdk.implementations.idp_v44.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into ORCACoreBlock(identifier=...) within your flow.
IDP_ORCA_IDENTIFIER = f'IDP_ORCA_{FLOWS_VERSION}_EXAMPLE'
IDP_ORCA_UUID = str(str_to_deterministic_uuid_4(IDP_ORCA_IDENTIFIER))
IDP_ORCA_TITLE = f'Document Processing with ORCA Subflow {FLOWS_VERSION_DOTTED} - flows-sdk example'
NO_LABEL = 'No'
YES_LABEL = 'Yes'


def idp_orca_workflow() -> Flow:
    file_filter = additional_blocks.FileFilterBlock(reference_name='file_filter')

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        trigger=file_filter.output('result.filtered_trigger'),
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        nlc_metadata=machine_classification.output('nlc_metadata'),
    )

    manual_classification_routing = Routing(
        reference_name='manual_classification_routing',
        decision=machine_classification.output('needs_manual_classification'),
        title='Is Manual Classification needed?',
        branches=[
            Routing.Branch(
                case='false', label=NO_LABEL, blocks=[], output='machine_classification'
            ),
        ],
        default_branch=Routing.DefaultBranch(label=YES_LABEL, blocks=[manual_classification]),
    )

    orca_extraction = idp_blocks.OpticalReasoningCognitionAgentBlock(
        reference_name='orca_extraction',
        submission=manual_classification_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=orca_extraction.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction_routing = Routing(
        reference_name='flexible_extraction_routing',
        decision=orca_extraction.output('needs_flexible_extraction'),
        title='Is Flexible Extraction needed?',
        branches=[
            Routing.Branch('false', blocks=[], label=NO_LABEL, output='orca_extraction'),
        ],
        default_branch=Routing.DefaultBranch(blocks=[flexible_extraction], label=YES_LABEL),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction_routing.output('submission'),
    )

    document_rendering = idp_blocks.DocumentRendererBlock(
        reference_name='document_rendering',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    schedule_orca_qa = QASchedulerBlock(
        reference_name='schedule_orca_qa',
        flow_identifier=workflow_input(Settings.OrcaQaFlowIdentifier),
        submission=document_rendering.output('submission'),
        # The keys are the input names of the ORCA QA flow (FlowInputs in orca_qa_flow.py).
        qa_config={
            'task_restrictions': workflow_input(Settings.OrcaQaTaskRestrictions),
            'orca_qa_sample_rate': workflow_input(Settings.OrcaQaSampleRate),
        },
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )
    schedule_orca_qa_routing = Routing(
        reference_name='schedule_orca_qa_routing',
        decision=workflow_input(Settings.OrcaQaFlowIdentifier),
        branches=[
            Routing.Branch(case='null', label='Skip QA', blocks=[], output='document_rendering')
        ],
        default_branch=Routing.DefaultBranch(label='Schedule QA', blocks=[schedule_orca_qa]),
        title='Should schedule QA?',
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=document_rendering.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    manifest = IDPCoreManifest(
        flow_identifier=IDP_ORCA_IDENTIFIER,
        roles=['read_only', 'supporting', IDP_CORE_FLOW_ROLE, 'orca'],
    )
    # Enable ORCA and ORCA only extraction for this flow.
    # (Shows ORCA Extraction parameters and hides Identification blocks related parameters from
    # IDPCoreManifest)
    for p in manifest.input:
        if p.name in [Settings.OrcaEnabled, Settings.OrcaOnlyExtraction]:
            p.value = True

    return Flow(
        uuid=UUID(IDP_ORCA_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=IDP_ORCA_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output using ORCA (Optical Reasoning & Cognition Agent).',
        manifest=manifest,
        output={'submission': submission_complete.output('submission')},
        blocks=[
            file_filter,
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification_routing,
            orca_extraction,
            flexible_extraction_routing,
            reprocessing,
            document_rendering,
            schedule_orca_qa_routing,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_orca_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

ORCA QA Flow Sample

from uuid import UUID

from flows_sdk.blocks import Block
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

ORCA_QA_FLOW_IDENTIFIER = 'ORCA_QA_FLOW_EXAMPLE'
ORCA_QA_FLOW_UUID = str(str_to_deterministic_uuid_4(ORCA_QA_FLOW_IDENTIFIER))
ORCA_QA_FLOW_TITLE = 'ORCA Quality Assurance Flow - flows-sdk example'
ORCA_QA_FLOW_DESCRIPTION = 'Optical Reasoning Cognition Agent Quality Assurance flow.'


class FlowInputs:
    Submission = 'submission'
    TaskRestrictions = 'task_restrictions'
    OrcaQASampleRate = 'orca_qa_sample_rate'

    @classmethod
    def get_manifest_inputs(cls) -> list[Parameter]:
        return [
            Parameter(
                name=cls.Submission,
                type='Submission',
                title='Submission Object',
                ui={'hidden': True},
            ),
            Parameter(
                name=cls.TaskRestrictions,
                type='TaskRestrictions',
                title='Task Restrictions',
                description=(
                    'Defines what users can access Supervision tasks created by this flow.'
                ),
                optional=True,
                ui={'hidden': True},
            ),
            Parameter(
                name=cls.OrcaQASampleRate,
                type='Percentage',
                schema={'type': 'number', 'minimum': 0, 'maximum': 1},
                title='ORCA QA Sample Rate',
                description='Defines the percentage of documents sampled for QA.',
                value=1,
                ui={'hidden': True},
            ),
        ]


def entry_point_workflow() -> Flow:
    return orca_qa_flow()


def orca_qa_flow() -> Flow:
    submission_ref = workflow_input(FlowInputs.Submission)
    task_restrictions_ref = workflow_input(FlowInputs.TaskRestrictions)
    orca_qa_sample_rate_ref = workflow_input(FlowInputs.OrcaQASampleRate)

    orca_qa_block = Block(
        identifier='VISION_LANGUAGE_MODEL_QA_2',
        reference_name='orca_qa',
        input={
            'submission': submission_ref,
            'sample_rate': orca_qa_sample_rate_ref,
            'task_restrictions': task_restrictions_ref,
        },
        title='ORCA Quality Assurance',
        description='Quality assurance for ORCA document processing',
    )

    return Flow(
        uuid=UUID(ORCA_QA_FLOW_UUID),
        owner_email='flows.sdk@hyperscience.com',
        title=ORCA_QA_FLOW_TITLE,
        manifest=Manifest(
            identifier=ORCA_QA_FLOW_IDENTIFIER,
            input=FlowInputs.get_manifest_inputs(),
            roles=['supporting', 'quality_assurance', 'read_only'],
            enable_overrides=True,
        ),
        blocks=[orca_qa_block],
        description=ORCA_QA_FLOW_DESCRIPTION,
    )


if __name__ == '__main__':
    export_flow(flow=entry_point_workflow())

OnError flow V44

This example shows how to produce an error message when a flow fails. The package includes the source code for an On-Error flow and its helper subflow. The On-Error flow should be configured to the IDP flow. The helper subflow generates the error message to be sent to the output blocks.

With the on-error flow configured, when the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered.

on_error_flow_v44.zip

OnError flow V44

from uuid import UUID

from flows_sdk.blocks import Block
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v44.additional_blocks import (
    JSONOutputsBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.types import CompatibilitySpec
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

ON_ERROR_FLOW_IDENTIFIER = 'ON_ERROR_3_EXAMPLE'
ON_ERROR_FLOW_UUID = str(str_to_deterministic_uuid_4(ON_ERROR_FLOW_IDENTIFIER))
ON_ERROR_TITLE = 'On Error V3 - flows-sdk example'
ON_ERROR_DESCRIPTION = (
    'Flow designed to be used as an "on-error" handler for other flows.'
    '\n\n'
    'With the on-error flow configured, when the original flow fails for any reason, and all '
    'the configured auto-retries have been exhausted, the on-error flow will be triggered.'
    '\n\n'
    'The input it receives is the ID of the failed flow run, which can be used to query '
    'additional information about the failure.'
    '\n\n'
    'If you need more complex logic in this flow, '
    'please consult the documentation of Flows SDK (https://flows-sdk.hyperscience.ai) '
    'and the available public APIs (https://docs.hyperscience.ai). Note that the custom logic '
    'should be implemented in the On-Error Subflow, so that it can be safely updated without '
    'losing the configured Output blocks. The Output blocks will receive the output of the '
    'On-Error Subflow as input.'
)


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)
ON_ERROR_CORE_FLOW_IDENTIFIER = 'ON_ERROR_CORE_EXAMPLE'

def on_error_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)
    on_error_core = Block(
        reference_name='on_error_core',
        identifier=ON_ERROR_CORE_FLOW_IDENTIFIER,
        input={'failed_run_uuid': flow_run_uuid},
        compatibility_spec=CompatibilitySpec(filter_roles=['on_error_core']),
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': on_error_core.output()})

    return Flow(
        uuid=UUID(ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=ON_ERROR_TITLE,
        description=ON_ERROR_DESCRIPTION,
        manifest=Manifest(
            identifier=ON_ERROR_FLOW_IDENTIFIER,
            roles=['on_error', 'supporting'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER],
        ),
        output={},
        input={},
        blocks=[
            on_error_core,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return on_error_flow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

OnError Subflow V44

from uuid import UUID

from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v44.additional_blocks import (
    HyperscienceRestApiBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

ON_ERROR_CORE_FLOW_IDENTIFIER = 'ON_ERROR_CORE_EXAMPLE'
ON_ERROR_CORE_FLOW_UUID = str(str_to_deterministic_uuid_4(ON_ERROR_CORE_FLOW_IDENTIFIER))
ON_ERROR_CORE_TITLE = 'Retrieve error information - flows-sdk example'
ON_ERROR_CORE_DESCRIPTION = (
    'Core logic of the on-error flow. Compiles the data, relevant to the error. This includes '
    'metadata for the failed flow run, without any inputs, outputs or logs, which get excluded '
    'for performance reasons - their size is unpredictable.'
    '\n\n'
    'Any custom error handling logic should be added here. The main on-error flow should only be '
    'used for configuring the outputs in the Outputs block, i.e. where the compiled on-error data '
    'should be sent to.'
)


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)

def on_error_core_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{flow_run_uuid}/summary',
    )

    return Flow(
        uuid=UUID(ON_ERROR_CORE_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=ON_ERROR_CORE_TITLE,
        description=ON_ERROR_CORE_DESCRIPTION,
        manifest=Manifest(
            identifier=ON_ERROR_CORE_FLOW_IDENTIFIER,
            roles=['on_error_core', 'supporting', 'read_only'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER],
        ),
        output={'flow_run': flow_run_rest_api_block.output('result.data')},
        input={},
        blocks=[flow_run_rest_api_block],
    )


def entry_point_workflow() -> Flow:
    return on_error_core_flow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

V43

Name	Description	Release	Sample Docs	Download
Document Processing Starter V43	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v43.zip
OnError flow V43	Example implementation of an OnError flow that is triggered once the original flow fails.			on_error_flow_v43.zip

Document Processing Starter V43

The basic document processing flow. From V38 on, the IDP Document processing is split into a wrapper flow (IDP) and a subflow (IDP Core). The package includes:

• IDP Core Flow - a read-only subflow containing the steps to process documents into machine-readable output

• Document Processing Flow - a wrapper around the IDP Core flow, used for managing input and output connector configuration and document processing workflow settings

• IDP Submission Notify Flow - A flow that can be configured for some blocks of the IDP Core Flow, used to send notifications to external systems when a submission has been created or is waiting for supervision

• Submission-aware On Error Flow - When the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered. The submission-aware on-error flow calls a helper subflow which generates the error message to be sent to the output blocks.

• Submission-aware On Error Subflow - The submission-aware on-error subflow gathers relevant submission data and combines it with data about the failed flow run. In case this OnError implementation is used on a non-IDP flow (i.e. a submissionless flow), submission data will be omitted

In V42.3+ a new Document Processing with ORCA Subflow is available for Advanced and Premium license.

• IDP ORCA Flow - a read-only subflow similar to IDP Core Flow but ORCA is used for field identification. _Note: Currently this flow does not support Structured forms._

• ORCA QA Flow - Quality assurance for ORCA flow. Meant to be used together with IDP ORCA Flow to spawn QA tasks for human review. Configurable through IDP ORCA Flow settings.

The package includes also release and sample documents for processing Form W-9s.

Download idp_flow_v43.zip

Document Processing Sample

from uuid import UUID

from flows_sdk.error_handling import ErrorHandling
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v43 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v43.idp_blocks import IDPCoreBlock
from flows_sdk.implementations.idp_v43.idp_values import (
    DEFAULT_IDP_CORE_IDENTIFIER,
    DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER,
    IdpWorkflowConfig,
    IDPWrapperManifest,
    Settings,
    get_idp_wf_config,
)
from flows_sdk.package_utils import export_flow

IDP_UUID = 'feac0001-c17e-42e8-b367-0000feac0001'
IDP_IDENTIFIER = 'DOCUMENT_PROCESSING_EXAMPLE'
IDP_TITLE = 'Document Processing - flows-sdk example'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:
    # By default, all the notification flows are set to `None`.
    # I.e. the system will send no intermediate notifications for submission change events.
    # So let's specify the notification flows that we want to use explicitly here:
    idp_wf_config.notification_workflow_initialization = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_classification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_identification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_transcription = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_flexible_extraction = (
        DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    )

    idp_flow = IDPCoreBlock(idp_wf_config, identifier=DEFAULT_IDP_CORE_IDENTIFIER)

    outputs = idp_blocks.IDPOutputsBlock(inputs={'submission': idp_flow.output('submission')})

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=IDP_TITLE,
        description='Use this flow to manage your input and output connector configuration and '
        'document processing workflow settings.',
        manifest=IDPWrapperManifest(IDP_IDENTIFIER),
        output={'submission': idp_flow.output('submission')},
        triggers=idp_values.IDPTriggers(),
        input={
            Settings.LayoutReleaseUuid: None,
        },
        blocks=[
            idp_flow,
            outputs,
        ],
        error_handling=ErrorHandling(on_error_flow=None),
    )


def entry_point_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Core V43 Sample

from uuid import UUID

from flows_sdk.blocks import Routing
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v43 import additional_blocks, idp_blocks
from flows_sdk.implementations.idp_v43.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED
from flows_sdk.implementations.idp_v43.idp_blocks.defaults import IDP_CORE_FLOW_ROLE
from flows_sdk.implementations.idp_v43.idp_values import IDPCoreManifest
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into IDPCoreBlock(identifier=...) within your flow.
IDP_CORE_IDENTIFIER = f'IDP_CORE_{FLOWS_VERSION}_EXAMPLE'
IDP_CORE_UUID = str(str_to_deterministic_uuid_4(IDP_CORE_IDENTIFIER))
IDP_CORE_TITLE = f'Document Processing Subflow {FLOWS_VERSION_DOTTED} - flows-sdk example'
NO_LABEL = 'No'
YES_LABEL = 'Yes'


def idp_workflow() -> Flow:
    file_filter = additional_blocks.FileFilterBlock(reference_name='file_filter')

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        trigger=file_filter.output('result.filtered_trigger'),
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        nlc_metadata=machine_classification.output('nlc_metadata'),
    )

    manual_classification_routing = Routing(
        reference_name='manual_classification_routing',
        decision=machine_classification.output('needs_manual_classification'),
        title='Is Manual Classification needed?',
        branches=[
            Routing.Branch(
                case='false', label=NO_LABEL, blocks=[], output='machine_classification'
            ),
        ],
        default_branch=Routing.DefaultBranch(label=YES_LABEL, blocks=[manual_classification]),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification_routing = Routing(
        reference_name='manual_identification_routing',
        decision=machine_identification.output('needs_manual_identification'),
        title='Is Manual Identification needed?',
        branches=[
            Routing.Branch('false', blocks=[], label=NO_LABEL, output='machine_identification'),
        ],
        default_branch=Routing.DefaultBranch(blocks=[manual_identification], label=YES_LABEL),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription_routing = Routing(
        reference_name='manual_transcription_routing',
        decision=machine_transcription.output('needs_manual_transcription'),
        title='Is Manual Transcription needed?',
        branches=[
            Routing.Branch('false', blocks=[], label=NO_LABEL, output='machine_transcription'),
        ],
        default_branch=Routing.DefaultBranch(blocks=[manual_transcription], label=YES_LABEL),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction_routing = Routing(
        reference_name='flexible_extraction_routing',
        decision=machine_transcription.output('needs_flexible_extraction'),
        title='Is Flexible Extraction needed?',
        branches=[
            Routing.Branch(
                'false', blocks=[], label=NO_LABEL, output='manual_transcription_routing'
            ),
        ],
        default_branch=Routing.DefaultBranch(blocks=[flexible_extraction], label=YES_LABEL),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction_routing.output('submission'),
    )

    document_rendering = idp_blocks.DocumentRendererBlock(
        reference_name='document_rendering',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=document_rendering.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    return Flow(
        uuid=UUID(IDP_CORE_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=IDP_CORE_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output.',
        manifest=IDPCoreManifest(
            flow_identifier=IDP_CORE_IDENTIFIER,
            roles=['read_only', 'supporting', IDP_CORE_FLOW_ROLE],
        ),
        output={'submission': submission_complete.output('submission')},
        blocks=[
            file_filter,
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification_routing,
            machine_identification,
            manual_identification_routing,
            machine_transcription,
            manual_transcription_routing,
            flexible_extraction_routing,
            reprocessing,
            document_rendering,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Submission Notify Sample

from pathlib import Path
from uuid import UUID

from flows_sdk.blocks import Outputs
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import icon_path_to_base64, str_to_deterministic_uuid_4, workflow_input

BASE_DIR = Path(__file__).parents[0]

IDP_SUBMISSION_NOTIFY_IDENTIFIER = 'IDP_SUBMISSION_NOTIFY_EXAMPLE'
IDP_SUBMISSION_NOTIFY_UUID = str(str_to_deterministic_uuid_4(IDP_SUBMISSION_NOTIFY_IDENTIFIER))
IDP_SUBMISSION_NOTIFY_TITLE = 'Submission State Notifications - flows-sdk example'
IDP_OUTPUT_ROLE = 'idp_output'


class SubmissionNotifyInputKeys:
    submission: str = 'submission'


class SubmissionNotifyManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['notifications', 'supporting'],
            input=[
                Parameter(
                    name=SubmissionNotifyInputKeys.submission,
                    type='Submission',
                    title='Submission Object',
                    ui={
                        'hidden': True,
                    },
                )
            ],
            ui={
                'icon': icon_path_to_base64(str(BASE_DIR / 'submission_notify_icon.png')),
            },
        )


def idp_submission_notify_workflow() -> Flow:
    return Flow(
        title=IDP_SUBMISSION_NOTIFY_TITLE,
        owner_email='flows-sdk@hyperscience.ai',
        description=(
            'Send notifications to external systems when a submission has been '
            'created or is waiting for supervision. \n'
            f'If the "Document Processing" flow is live, this flow '
            'must also be live, but it can be empty.'
        ),
        manifest=SubmissionNotifyManifest(IDP_SUBMISSION_NOTIFY_IDENTIFIER),
        blocks=[
            Outputs(
                reference_name='outputs',
                title='Outputs',
                description=(
                    'Send submission data to external systems when a submission '
                    'has been created or is waiting for supervision'
                ),
                role_filter=[IDP_OUTPUT_ROLE],
                input_template={
                    'submission': {'id': workflow_input('submission.id')},
                    'enabled': True,
                },
                blocks=[],
            )
        ],
        uuid=UUID(IDP_SUBMISSION_NOTIFY_UUID),
        input={},
        output={},
    )


def entry_point_workflow() -> Flow:
    return idp_submission_notify_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

Submission-aware On Error Flow Sample

from uuid import UUID

from flows_sdk.blocks import Block
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v43.additional_blocks import JSONOutputsBlock
from flows_sdk.package_utils import export_flow
from flows_sdk.types import CompatibilitySpec
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER = 'SUBMISSION_AWARE_ON_ERROR_FLOW_3_EXAMPLE'
SUBMISSION_AWARE_ON_ERROR_FLOW_UUID = str(
    str_to_deterministic_uuid_4(SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER)
)
SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE = (
    'On-Error with included Submission data V3 - flows-sdk example'
)
SUBMISSION_AWARE_ON_ERROR_DESCRIPTION = (
    'Flow designed to be used as an "on-error" handler for submission-based flows.'
    '\n\n'
    'With the on-error flow configured, when the original flow fails for any reason, and all '
    'the configured auto-retries have been exhausted, the on-error flow will be triggered.'
    '\n\n'
    'The input it receives is the ID of the failed flow run, which can be used to query '
    'additional information about the failure.'
    '\n\n'
    'If you need more complex logic in this flow please consult the documentation of Flows SDK '
    '(https://flows-sdk.hyperscience.ai) and the available public APIs '
    '(https://docs.hyperscience.ai). Note that the custom logic should be implemented in the '
    'On-Error Subflow, so that it can be safely updated without losing the configured Output '
    'blocks. The Output blocks will receive the output of the On-Error Subflow as input.'
)

class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)
SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER = 'SUBMISSION_AWARE_ON_ERROR_CORE_EXAMPLE'

def submission_aware_on_error_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)

    on_error_core = Block(
        reference_name='on_error_core',
        identifier=SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER,
        input={'failed_run_uuid': flow_run_uuid},
        compatibility_spec=CompatibilitySpec(filter_roles=['on_error_core']),
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': on_error_core.output()})

    return Flow(
        uuid=UUID(SUBMISSION_AWARE_ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE,
        description=SUBMISSION_AWARE_ON_ERROR_DESCRIPTION,
        manifest=Manifest(
            identifier=SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER,
            roles=['on_error', 'supporting'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER],
        ),
        output={},
        input={},
        blocks=[
            on_error_core,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return submission_aware_on_error_flow()


if __name__ == '__main__':
    export_flow(submission_aware_on_error_flow())

Submission-aware On Error Subflow Sample

from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v43.additional_blocks import HyperscienceRestApiBlock
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsTask
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER = 'SUBMISSION_AWARE_ON_ERROR_CORE_EXAMPLE'
SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_UUID = str(
    str_to_deterministic_uuid_4(SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER)
)
SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_TITLE = (
    'Retrieve error and submission information - flows-sdk example'
)
SUBMISSION_AWARE_ON_ERROR_CORE_DESCRIPTION = (
    'Core logic of the on-error flow. Compiles the data, relevant to the error. This includes '
    'metadata for the failed flow run and the JSON serialized submission data, if the flow run '
    'is linked to a submission. The flow run inputs, outputs and logs are excluded for '
    'performance reasons - their size is unpredictable.'
    '\n\n'
    'In this On-Error Subflow implementation, we fetch the failed flow run, alongside with the '
    'full submission metadata (if the submission ID was found). The data format of the output '
    'conforms to the schema documented here: https://docs.hyperscience.ai/#flows-runs and here: '
    'https://docs.hyperscience.ai/#submissions, under the respective keys "flow_run" and '
    '"submission".'
    '\n\n'
    'Any custom error handling logic should be added here. The main on-error flow should only be '
    'used for configuring the outputs in the Outputs block, i.e. where the compiled on-error data '
    'should be sent to.'
)


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)


def submission_aware_on_error_core_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        title='Get Flow Run',
        description='Get the full metadata of the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{flow_run_uuid}/summary',
    )

    def _get_correlation_id(_hs_task: HsTask) -> dict:
        return {'correlation_id': _hs_task.correlation_id}

    get_correlation_id_block = PythonBlock(
        reference_name='get_correlation_id',
        title='Get Correlation ID',
        description='Get the correlation ID for the failed Submission / Flow Run',
        code=_get_correlation_id,
    )

    fetch_submissions_block = HyperscienceRestApiBlock(
        reference_name='fetch_submission',
        title='Fetch Submission',
        description='Fetch the submission data for the failed flow run',
        method='GET',
        app_endpoint='/api/v5/submissions?debug=true'
        f'&correlation_id={get_correlation_id_block.output('correlation_id')}',
    )

    return Flow(
        uuid=UUID(SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_TITLE,
        description=SUBMISSION_AWARE_ON_ERROR_CORE_DESCRIPTION,
        manifest=Manifest(
            identifier=SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER,
            roles=['on_error_core', 'supporting', 'read_only'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER]
        ),
        output={
            'flow_run': flow_run_rest_api_block.output('result.data'),
            'submission': fetch_submissions_block.output('result.data'),
        },
        input={},
        blocks=[
            flow_run_rest_api_block,
            get_correlation_id_block,
            fetch_submissions_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return submission_aware_on_error_core_flow()


if __name__ == '__main__':
    export_flow(submission_aware_on_error_core_flow())

IDP ORCA V43 Sample

from uuid import UUID

from flows_sdk.blocks import Routing
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v43 import additional_blocks, idp_blocks
from flows_sdk.implementations.idp_v43.additional_blocks import QASchedulerBlock
from flows_sdk.implementations.idp_v43.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED
from flows_sdk.implementations.idp_v43.idp_values import IDPCoreManifest, Settings
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into ORCACoreBlock(identifier=...) within your flow.
IDP_ORCA_IDENTIFIER = f'IDP_ORCA_{FLOWS_VERSION}_EXAMPLE'
IDP_ORCA_UUID = str(str_to_deterministic_uuid_4(IDP_ORCA_IDENTIFIER))
IDP_ORCA_TITLE = f'Document Processing with ORCA Subflow {FLOWS_VERSION_DOTTED} - flows-sdk example'
NO_LABEL = 'No'
YES_LABEL = 'Yes'


def idp_orca_workflow() -> Flow:
    file_filter = additional_blocks.FileFilterBlock(reference_name='file_filter')

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        trigger=file_filter.output('result.filtered_trigger'),
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        nlc_metadata=machine_classification.output('nlc_metadata'),
    )

    manual_classification_routing = Routing(
        reference_name='manual_classification_routing',
        decision=machine_classification.output('needs_manual_classification'),
        title='Is Manual Classification needed?',
        branches=[
            Routing.Branch(
                case='false', label=NO_LABEL, blocks=[], output='machine_classification'
            ),
        ],
        default_branch=Routing.DefaultBranch(label=YES_LABEL, blocks=[manual_classification]),
    )

    orca_extraction = idp_blocks.OpticalReasoningCognitionAgentBlock(
        reference_name='orca_extraction',
        submission=manual_classification_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=orca_extraction.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction_routing = Routing(
        reference_name='flexible_extraction_routing',
        decision=orca_extraction.output('needs_flexible_extraction'),
        title='Is Flexible Extraction needed?',
        branches=[
            Routing.Branch('false', blocks=[], label=NO_LABEL, output='orca_extraction'),
        ],
        default_branch=Routing.DefaultBranch(blocks=[flexible_extraction], label=YES_LABEL),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction_routing.output('submission'),
    )

    document_rendering = idp_blocks.DocumentRendererBlock(
        reference_name='document_rendering',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    schedule_orca_qa = QASchedulerBlock(
        reference_name='schedule_orca_qa',
        flow_identifier=workflow_input(Settings.OrcaQaFlowIdentifier),
        submission=document_rendering.output('submission'),
        # The keys are the input names of the ORCA QA flow (FlowInputs in orca_qa_flow.py).
        qa_config={
            'task_restrictions': workflow_input(Settings.OrcaQaTaskRestrictions),
            'orca_qa_sample_rate': workflow_input(Settings.OrcaQaSampleRate),
        },
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )
    schedule_orca_qa_routing = Routing(
        reference_name='schedule_orca_qa_routing',
        decision=workflow_input(Settings.OrcaQaFlowIdentifier),
        branches=[
            Routing.Branch(case='null', label='Skip QA', blocks=[], output='document_rendering')
        ],
        default_branch=Routing.DefaultBranch(label='Schedule QA', blocks=[schedule_orca_qa]),
        title='Should schedule QA?',
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=document_rendering.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    manifest = IDPCoreManifest(
        flow_identifier=IDP_ORCA_IDENTIFIER, roles=['read_only', 'supporting', 'idp_core', 'orca']
    )
    # Enable ORCA and ORCA only extraction for this flow.
    # (Shows ORCA Extraction parameters and hides Identification blocks related parameters from
    # IDPCoreManifest)
    for p in manifest.input:
        if p.name in [Settings.OrcaEnabled, Settings.OrcaOnlyExtraction]:
            p.value = True

    return Flow(
        uuid=UUID(IDP_ORCA_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=IDP_ORCA_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output using ORCA (Optical Reasoning & Cognition Agent).',
        manifest=manifest,
        output={'submission': submission_complete.output('submission')},
        blocks=[
            file_filter,
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification_routing,
            orca_extraction,
            flexible_extraction_routing,
            reprocessing,
            document_rendering,
            schedule_orca_qa_routing,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_orca_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

ORCA QA Flow Sample

from uuid import UUID

from flows_sdk.blocks import Block
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

ORCA_QA_FLOW_IDENTIFIER = 'ORCA_QA_FLOW_EXAMPLE'
ORCA_QA_FLOW_UUID = str(str_to_deterministic_uuid_4(ORCA_QA_FLOW_IDENTIFIER))
ORCA_QA_FLOW_TITLE = 'ORCA Quality Assurance Flow - flows-sdk example'
ORCA_QA_FLOW_DESCRIPTION = 'Optical Reasoning Cognition Agent Quality Assurance flow.'


class FlowInputs:
    Submission = 'submission'
    TaskRestrictions = 'task_restrictions'
    OrcaQASampleRate = 'orca_qa_sample_rate'


def orca_qa_flow() -> Flow:
    submission_ref = workflow_input(FlowInputs.Submission)
    task_restrictions_ref = workflow_input(FlowInputs.TaskRestrictions)
    orca_qa_sample_rate_ref = workflow_input(FlowInputs.OrcaQASampleRate)

    orca_qa_block = Block(
        identifier='VISION_LANGUAGE_MODEL_QA_2',
        reference_name='orca_qa',
        input={
            'submission': submission_ref,
            'sample_rate': orca_qa_sample_rate_ref,
            'task_restrictions': task_restrictions_ref,
        },
        title='ORCA Quality Assurance',
        description='Quality assurance for ORCA document processing',
    )

    return Flow(
        uuid=UUID(ORCA_QA_FLOW_UUID),
        owner_email='flows.sdk@hyperscience.com',
        title=ORCA_QA_FLOW_TITLE,
        manifest=Manifest(
            identifier=ORCA_QA_FLOW_IDENTIFIER,
            input=[
                Parameter(
                    name=FlowInputs.Submission,
                    type='Submission',
                    title='Submission Object',
                    ui={'hidden': True},
                ),
                Parameter(
                    name=FlowInputs.TaskRestrictions,
                    type='TaskRestrictions',
                    title='Task Restrictions',
                    description=(
                        'Defines what users can access Supervision tasks created by this flow.'
                    ),
                    optional=True,
                    ui={'hidden': True},
                ),
                Parameter(
                    name=FlowInputs.OrcaQASampleRate,
                    type='Percentage',
                    schema={'type': 'number', 'minimum': 0, 'maximum': 1},
                    title='ORCA QA Sample Rate',
                    description='Defines the percentage of documents sampled for QA.',
                    value=1,
                    ui={'hidden': True},
                ),
            ],
            roles=['supporting', 'quality_assurance', 'read_only'],
            enable_overrides=True,
        ),
        blocks=[orca_qa_block],
        description=ORCA_QA_FLOW_DESCRIPTION,
    )


def entry_point_workflow() -> Flow:
    return orca_qa_flow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

OnError flow V43

This example shows how to produce an error message when a flow fails. The package includes the source code for an On-Error flow and its helper subflow. The On-Error flow should be configured to the IDP flow. The helper subflow generates the error message to be sent to the output blocks.

With the on-error flow configured, when the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered.

on_error_flow_v43.zip

OnError flow V43

from uuid import UUID

from flows_sdk.blocks import Block
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v43.additional_blocks import (
    JSONOutputsBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.types import CompatibilitySpec
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

ON_ERROR_FLOW_IDENTIFIER = 'ON_ERROR_3_EXAMPLE'
ON_ERROR_FLOW_UUID = str(str_to_deterministic_uuid_4(ON_ERROR_FLOW_IDENTIFIER))
ON_ERROR_TITLE = 'On Error V3 - flows-sdk example'
ON_ERROR_DESCRIPTION = (
    'Flow designed to be used as an "on-error" handler for other flows.'
    '\n\n'
    'With the on-error flow configured, when the original flow fails for any reason, and all '
    'the configured auto-retries have been exhausted, the on-error flow will be triggered.'
    '\n\n'
    'The input it receives is the ID of the failed flow run, which can be used to query '
    'additional information about the failure.'
    '\n\n'
    'If you need more complex logic in this flow, '
    'please consult the documentation of Flows SDK (https://flows-sdk.hyperscience.ai) '
    'and the available public APIs (https://docs.hyperscience.ai). Note that the custom logic '
    'should be implemented in the On-Error Subflow, so that it can be safely updated without '
    'losing the configured Output blocks. The Output blocks will receive the output of the '
    'On-Error Subflow as input.'
)


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)
ON_ERROR_CORE_FLOW_IDENTIFIER = 'ON_ERROR_CORE_EXAMPLE'

def on_error_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)
    on_error_core = Block(
        reference_name='on_error_core',
        identifier=ON_ERROR_CORE_FLOW_IDENTIFIER,
        input={'failed_run_uuid': flow_run_uuid},
        compatibility_spec=CompatibilitySpec(filter_roles=['on_error_core']),
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': on_error_core.output()})

    return Flow(
        uuid=UUID(ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=ON_ERROR_TITLE,
        description=ON_ERROR_DESCRIPTION,
        manifest=Manifest(
            identifier=ON_ERROR_FLOW_IDENTIFIER,
            roles=['on_error', 'supporting'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER],
        ),
        output={},
        input={},
        blocks=[
            on_error_core,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return on_error_flow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

OnError Subflow V43

from uuid import UUID

from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v43.additional_blocks import (
    HyperscienceRestApiBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

ON_ERROR_CORE_FLOW_IDENTIFIER = 'ON_ERROR_CORE_EXAMPLE'
ON_ERROR_CORE_FLOW_UUID = str(str_to_deterministic_uuid_4(ON_ERROR_CORE_FLOW_IDENTIFIER))
ON_ERROR_CORE_TITLE = 'Retrieve error information - flows-sdk example'
ON_ERROR_CORE_DESCRIPTION = (
    'Core logic of the on-error flow. Compiles the data, relevant to the error. This includes '
    'metadata for the failed flow run, without any inputs, outputs or logs, which get excluded '
    'for performance reasons - their size is unpredictable.'
    '\n\n'
    'Any custom error handling logic should be added here. The main on-error flow should only be '
    'used for configuring the outputs in the Outputs block, i.e. where the compiled on-error data '
    'should be sent to.'
)


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)

def on_error_core_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{flow_run_uuid}/summary',
    )

    return Flow(
        uuid=UUID(ON_ERROR_CORE_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=ON_ERROR_CORE_TITLE,
        description=ON_ERROR_CORE_DESCRIPTION,
        manifest=Manifest(
            identifier=ON_ERROR_CORE_FLOW_IDENTIFIER,
            roles=['on_error_core', 'supporting', 'read_only'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER],
        ),
        output={'flow_run': flow_run_rest_api_block.output('result.data')},
        input={},
        blocks=[flow_run_rest_api_block],
    )


def entry_point_workflow() -> Flow:
    return on_error_core_flow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

V42

Name	Description	Release	Sample Docs	Download
Document Processing Starter V42	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v42.zip
OnError flow V42	Example implementation of an OnError flow that is triggered once the original flow fails.			on_error_flow_v42.zip

Document Processing Starter V42

The basic document processing flow. From V38 on, the IDP Document processing is split into a wrapper flow (IDP) and a subflow (IDP Core). The package includes:

• IDP Core Flow - a read-only subflow containing the steps to process documents into machine-readable output

• Document Processing Flow - a wrapper around the IDP Core flow, used for managing input and output connector configuration and document processing workflow settings

• IDP Submission Notify Flow - A flow that can be configured for some blocks of the IDP Core Flow, used to send notifications to external systems when a submission has been created or is waiting for supervision

• Submission-aware On Error Flow - When the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered. The submission-aware on-error flow calls a helper subflow which generates the error message to be sent to the output blocks.

• Submission-aware On Error Subflow - The submission-aware on-error subflow gathers relevant submission data and combines it with data about the failed flow run. In case this OnError implementation is used on a non-IDP flow (i.e. a submissionless flow), submission data will be omitted

In V42.3+ a new Document Processing with ORCA Subflow is available for Advanced and Premium license.

• IDP ORCA Flow - a read-only subflow similar to IDP Core Flow but ORCA is used for field identification. _Note: Currently this flow does not support Structured forms._

• ORCA QA Flow - Quality assurance for ORCA flow. Meant to be used together with IDP ORCA Flow to spawn QA tasks for human review. Configurable through IDP ORCA Flow settings.

The package includes also release and sample documents for processing Form W-9s.

Download idp_flow_v42.zip

Document Processing Sample

from uuid import UUID

from flows_sdk.error_handling import ErrorHandling
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v42 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v42.idp_blocks import IDPCoreBlock
from flows_sdk.implementations.idp_v42.idp_values import (
    DEFAULT_IDP_CORE_IDENTIFIER,
    DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER,
    IdpWorkflowConfig,
    IDPWrapperManifest,
    Settings,
    get_idp_wf_config,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4

IDP_IDENTIFIER = 'DOCUMENT_PROCESSING_EXAMPLE'
IDP_UUID = str(str_to_deterministic_uuid_4(IDP_IDENTIFIER))
IDP_TITLE = 'Document Processing - flows-sdk example'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:
    # By default, all the notification flows are set to `None`.
    # I.e. the system will send no intermediate notifications for submission change events.
    # So let's specify the notification flows that we want to use explicitly here:
    idp_wf_config.notification_workflow_initialization = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_classification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_identification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_transcription = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_flexible_extraction = (
        DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    )

    idp_flow = IDPCoreBlock(idp_wf_config, identifier=DEFAULT_IDP_CORE_IDENTIFIER)

    outputs = idp_blocks.IDPOutputsBlock(inputs={'submission': idp_flow.output('submission')})

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=IDP_TITLE,
        description='Use this flow to manage your input and output connector configuration and '
        'document processing workflow settings.',
        manifest=IDPWrapperManifest(IDP_IDENTIFIER),
        output={'submission': idp_flow.output('submission')},
        triggers=idp_values.IDPTriggers(),
        input={
            Settings.LayoutReleaseUuid: None,
        },
        blocks=[
            idp_flow,
            outputs,
        ],
        error_handling=ErrorHandling(on_error_flow=None),
    )


def entry_point_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Core V42 Sample

from uuid import UUID

from flows_sdk.blocks import Routing
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v42 import additional_blocks, idp_blocks
from flows_sdk.implementations.idp_v42.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED
from flows_sdk.implementations.idp_v42.idp_values import IDPCoreManifest
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into IDPCoreBlock(identifier=...) within your flow.
IDP_CORE_IDENTIFIER = f'IDP_CORE_{FLOWS_VERSION}_EXAMPLE'
IDP_CORE_UUID = str(str_to_deterministic_uuid_4(IDP_CORE_IDENTIFIER))
IDP_CORE_TITLE = f'Document Processing Subflow {FLOWS_VERSION_DOTTED} - flows-sdk example'
NO_LABEL = 'No'
YES_LABEL = 'Yes'


def idp_workflow() -> Flow:
    file_filter = additional_blocks.FileFilterBlock(reference_name='file_filter')

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        trigger=file_filter.output('result.filtered_trigger'),
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification_routing = Routing(
        reference_name='manual_classification_routing',
        decision=machine_classification.output('needs_manual_classification'),
        title='Is Manual Classification needed?',
        branches=[
            Routing.Branch(
                case='false', label=NO_LABEL, blocks=[], output='machine_classification'
            ),
        ],
        default_branch=Routing.DefaultBranch(label=YES_LABEL, blocks=[manual_classification]),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification_routing = Routing(
        reference_name='manual_identification_routing',
        decision=machine_identification.output('needs_manual_identification'),
        title='Is Manual Identification needed?',
        branches=[
            Routing.Branch('false', blocks=[], label=NO_LABEL, output='machine_identification'),
        ],
        default_branch=Routing.DefaultBranch(blocks=[manual_identification], label=YES_LABEL),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription_routing = Routing(
        reference_name='manual_transcription_routing',
        decision=machine_transcription.output('needs_manual_transcription'),
        title='Is Manual Transcription needed?',
        branches=[
            Routing.Branch('false', blocks=[], label=NO_LABEL, output='machine_transcription'),
        ],
        default_branch=Routing.DefaultBranch(blocks=[manual_transcription], label=YES_LABEL),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction_routing = Routing(
        reference_name='flexible_extraction_routing',
        decision=machine_transcription.output('needs_flexible_extraction'),
        title='Is Flexible Extraction needed?',
        branches=[
            Routing.Branch(
                'false', blocks=[], label=NO_LABEL, output='manual_transcription_routing'
            ),
        ],
        default_branch=Routing.DefaultBranch(blocks=[flexible_extraction], label=YES_LABEL),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction_routing.output('submission'),
    )

    document_rendering = idp_blocks.DocumentRendererBlock(
        reference_name='document_rendering',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=document_rendering.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    return Flow(
        uuid=UUID(IDP_CORE_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=IDP_CORE_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output.',
        manifest=IDPCoreManifest(
            flow_identifier=IDP_CORE_IDENTIFIER, roles=['read_only', 'supporting', 'idp_core']
        ),
        output={'submission': submission_complete.output('submission')},
        blocks=[
            file_filter,
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification_routing,
            machine_identification,
            manual_identification_routing,
            machine_transcription,
            manual_transcription_routing,
            flexible_extraction_routing,
            reprocessing,
            document_rendering,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Submission Notify Sample

from pathlib import Path
from uuid import UUID

from flows_sdk.blocks import Outputs
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import icon_path_to_base64, str_to_deterministic_uuid_4, workflow_input

BASE_DIR = Path(__file__).parents[0]

IDP_SUBMISSION_NOTIFY_IDENTIFIER = 'IDP_SUBMISSION_NOTIFY_EXAMPLE'
IDP_SUBMISSION_NOTIFY_UUID = str(str_to_deterministic_uuid_4(IDP_SUBMISSION_NOTIFY_IDENTIFIER))
IDP_SUBMISSION_NOTIFY_TITLE = 'Submission State Notifications - flows-sdk example'
IDP_OUTPUT_ROLE = 'idp_output'


class SubmissionNotifyInputKeys:
    submission: str = 'submission'


class SubmissionNotifyManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['notifications', 'supporting'],
            input=[
                Parameter(
                    name=SubmissionNotifyInputKeys.submission,
                    type='Submission',
                    title='Submission Object',
                    ui={
                        'hidden': True,
                    },
                )
            ],
            ui={
                'icon': icon_path_to_base64(str(BASE_DIR / 'submission_notify_icon.png')),
            },
        )


def idp_submission_notify_workflow() -> Flow:
    return Flow(
        title=IDP_SUBMISSION_NOTIFY_TITLE,
        owner_email='flows-sdk@hyperscience.ai',
        description=(
            'Send notifications to external systems when a submission has been '
            'created or is waiting for supervision. \n'
            'If the "Document Processing" flow is live, this flow '
            'must also be live, but it can be empty.'
        ),
        manifest=SubmissionNotifyManifest(IDP_SUBMISSION_NOTIFY_IDENTIFIER),
        blocks=[
            Outputs(
                reference_name='outputs',
                title='Outputs',
                description=(
                    'Send submission data to external systems when a submission '
                    'has been created or is waiting for supervision'
                ),
                role_filter=[IDP_OUTPUT_ROLE],
                input_template={
                    'submission': {'id': workflow_input('submission.id')},
                    'enabled': True,
                },
                blocks=[],
            )
        ],
        uuid=UUID(IDP_SUBMISSION_NOTIFY_UUID),
        input={},
        output={},
    )


def entry_point_workflow() -> Flow:
    return idp_submission_notify_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

Submission-aware On Error Flow Sample

from uuid import UUID

from flows_sdk.blocks import Block
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v42.additional_blocks import JSONOutputsBlock
from flows_sdk.package_utils import export_flow
from flows_sdk.types import CompatibilitySpec
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER = 'SUBMISSION_AWARE_ON_ERROR_FLOW_3_EXAMPLE'
SUBMISSION_AWARE_ON_ERROR_FLOW_UUID = str(
    str_to_deterministic_uuid_4(SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER)
)
SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE = (
    'On-Error with included Submission data V3 - flows-sdk example'
)
SUBMISSION_AWARE_ON_ERROR_DESCRIPTION = (
    'Flow designed to be used as an "on-error" handler for submission-based flows.'
    '\n\n'
    'With the on-error flow configured, when the original flow fails for any reason, and all '
    'the configured auto-retries have been exhausted, the on-error flow will be triggered.'
    '\n\n'
    'The input it receives is the ID of the failed flow run, which can be used to query '
    'additional information about the failure.'
    '\n\n'
    'If you need more complex logic in this flow please consult the documentation of Flows SDK '
    '(https://flows-sdk.hyperscience.ai) and the available public APIs '
    '(https://docs.hyperscience.ai). Note that the custom logic should be implemented in the '
    'On-Error Subflow, so that it can be safely updated without losing the configured Output '
    'blocks. The Output blocks will receive the output of the On-Error Subflow as input.'
)

class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)
SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER = 'SUBMISSION_AWARE_ON_ERROR_CORE_EXAMPLE'

def submission_aware_on_error_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)

    on_error_core = Block(
        reference_name='on_error_core',
        identifier=SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER,
        input={'failed_run_uuid': flow_run_uuid},
        compatibility_spec=CompatibilitySpec(filter_roles=['on_error_core']),
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': on_error_core.output()})

    return Flow(
        uuid=UUID(SUBMISSION_AWARE_ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE,
        description=SUBMISSION_AWARE_ON_ERROR_DESCRIPTION,
        manifest=Manifest(
            identifier=SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER,
            roles=['on_error', 'supporting'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER],
        ),
        output={},
        input={},
        blocks=[
            on_error_core,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return submission_aware_on_error_flow()


if __name__ == '__main__':
    export_flow(submission_aware_on_error_flow())

Submission-aware On Error Subflow Sample

from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v42.additional_blocks import HyperscienceRestApiBlock
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsTask
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER = 'SUBMISSION_AWARE_ON_ERROR_CORE_EXAMPLE'
SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_UUID = str(
    str_to_deterministic_uuid_4(SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER)
)
SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_TITLE = (
    'Retrieve error and submission information - flows-sdk example'
)
SUBMISSION_AWARE_ON_ERROR_CORE_DESCRIPTION = (
    'Core logic of the on-error flow. Compiles the data, relevant to the error. This includes '
    'metadata for the failed flow run and the JSON serialized submission data, if the flow run '
    'is linked to a submission. The flow run inputs, outputs and logs are excluded for '
    'performance reasons - their size is unpredictable.'
    '\n\n'
    'In this On-Error Subflow implementation, we fetch the failed flow run, alongside with the '
    'full submission metadata (if the submission ID was found). The data format of the output '
    'conforms to the schema documented here: https://docs.hyperscience.ai/#flows-runs and here: '
    'https://docs.hyperscience.ai/#submissions, under the respective keys "flow_run" and '
    '"submission".'
    '\n\n'
    'Any custom error handling logic should be added here. The main on-error flow should only be '
    'used for configuring the outputs in the Outputs block, i.e. where the compiled on-error data '
    'should be sent to.'
)


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)


def submission_aware_on_error_core_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        title='Get Flow Run',
        description='Get the full metadata of the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{flow_run_uuid}/summary',
    )

    def _get_correlation_id(_hs_task: HsTask) -> dict:
        return {'correlation_id': _hs_task.correlation_id}

    get_correlation_id_block = PythonBlock(
        reference_name='get_correlation_id',
        title='Get Correlation ID',
        description='Get the correlation ID for the failed Submission / Flow Run',
        code=_get_correlation_id,
    )

    fetch_submissions_block = HyperscienceRestApiBlock(
        reference_name='fetch_submission',
        title='Fetch Submission',
        description='Fetch the submission data for the failed flow run',
        method='GET',
        app_endpoint='/api/v5/submissions?debug=true'
        f'&correlation_id={get_correlation_id_block.output('correlation_id')}',
    )

    return Flow(
        uuid=UUID(SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_TITLE,
        description=SUBMISSION_AWARE_ON_ERROR_CORE_DESCRIPTION,
        manifest=Manifest(
            identifier=SUBMISSION_AWARE_ON_ERROR_CORE_FLOW_IDENTIFIER,
            roles=['on_error_core', 'supporting', 'read_only'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER]
        ),
        output={
            'flow_run': flow_run_rest_api_block.output('result.data'),
            'submission': fetch_submissions_block.output('result.data'),
        },
        input={},
        blocks=[
            flow_run_rest_api_block,
            get_correlation_id_block,
            fetch_submissions_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return submission_aware_on_error_core_flow()


if __name__ == '__main__':
    export_flow(submission_aware_on_error_core_flow())

IDP ORCA V42 Sample

from uuid import UUID

from flows_sdk.blocks import Routing
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v42 import additional_blocks, idp_blocks
from flows_sdk.implementations.idp_v42.additional_blocks import QASchedulerBlock
from flows_sdk.implementations.idp_v42.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED
from flows_sdk.implementations.idp_v42.idp_values import IDPCoreManifest, Settings
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into ORCACoreBlock(identifier=...) within your flow.
IDP_ORCA_IDENTIFIER = f'IDP_ORCA_{FLOWS_VERSION}_EXAMPLE'
IDP_ORCA_UUID = str(str_to_deterministic_uuid_4(IDP_ORCA_IDENTIFIER))
IDP_ORCA_TITLE = f'Document Processing with ORCA Subflow {FLOWS_VERSION_DOTTED} - flows-sdk example'
NO_LABEL = 'No'
YES_LABEL = 'Yes'


def idp_orca_workflow() -> Flow:
    file_filter = additional_blocks.FileFilterBlock(reference_name='file_filter')

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        trigger=file_filter.output('result.filtered_trigger'),
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification_routing = Routing(
        reference_name='manual_classification_routing',
        decision=machine_classification.output('needs_manual_classification'),
        title='Is Manual Classification needed?',
        branches=[
            Routing.Branch(
                case='false', label=NO_LABEL, blocks=[], output='machine_classification'
            ),
        ],
        default_branch=Routing.DefaultBranch(label=YES_LABEL, blocks=[manual_classification]),
    )

    orca_extraction = idp_blocks.OpticalReasoningCognitionAgentBlock(
        reference_name='orca_extraction',
        submission=manual_classification_routing.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=orca_extraction.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction_routing = Routing(
        reference_name='flexible_extraction_routing',
        decision=orca_extraction.output('needs_flexible_extraction'),
        title='Is Flexible Extraction needed?',
        branches=[
            Routing.Branch('false', blocks=[], label=NO_LABEL, output='orca_extraction'),
        ],
        default_branch=Routing.DefaultBranch(blocks=[flexible_extraction], label=YES_LABEL),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction_routing.output('submission'),
    )

    document_rendering = idp_blocks.DocumentRendererBlock(
        reference_name='document_rendering',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    schedule_orca_qa = QASchedulerBlock(
        reference_name='schedule_orca_qa',
        flow_identifier=workflow_input(Settings.OrcaQaFlowIdentifier),
        submission=document_rendering.output('submission'),
        # The keys are the input names of the ORCA QA flow (FlowInputs in orca_qa_flow.py).
        qa_config={
            'task_restrictions': workflow_input(Settings.OrcaQaTaskRestrictions),
            'orca_qa_sample_rate': workflow_input(Settings.OrcaQaSampleRate),
        },
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )
    schedule_orca_qa_routing = Routing(
        reference_name='schedule_orca_qa_routing',
        decision=workflow_input(Settings.OrcaQaFlowIdentifier),
        branches=[
            Routing.Branch(case='null', label='Skip QA', blocks=[], output='document_rendering')
        ],
        default_branch=Routing.DefaultBranch(label='Schedule QA', blocks=[schedule_orca_qa]),
        title='Should schedule QA?',
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=document_rendering.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    manifest = IDPCoreManifest(
        flow_identifier=IDP_ORCA_IDENTIFIER, roles=['read_only', 'supporting', 'idp_core', 'orca']
    )
    # Enable ORCA and ORCA only extraction for this flow.
    # (Shows ORCA Extraction parameters and hides Identification blocks related parameters from
    # IDPCoreManifest)
    for p in manifest.input:
        if p.name in [Settings.OrcaEnabled, Settings.OrcaOnlyExtraction]:
            p.value = True

    return Flow(
        uuid=UUID(IDP_ORCA_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=IDP_ORCA_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output using ORCA (Optical Reasoning & Cognition Agent).',
        manifest=manifest,
        output={'submission': submission_complete.output('submission')},
        blocks=[
            file_filter,
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification_routing,
            orca_extraction,
            flexible_extraction_routing,
            reprocessing,
            document_rendering,
            schedule_orca_qa_routing,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_orca_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

ORCA QA Flow Sample

from uuid import UUID

from flows_sdk.blocks import Block
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

ORCA_QA_FLOW_IDENTIFIER = 'ORCA_QA_FLOW_EXAMPLE'
ORCA_QA_FLOW_UUID = str(str_to_deterministic_uuid_4(ORCA_QA_FLOW_IDENTIFIER))
ORCA_QA_FLOW_TITLE = 'ORCA Quality Assurance Flow - flows-sdk example'
ORCA_QA_FLOW_DESCRIPTION = 'Optical Reasoning Cognition Agent Quality Assurance flow.'


class FlowInputs:
    Submission = 'submission'
    TaskRestrictions = 'task_restrictions'
    OrcaQASampleRate = 'orca_qa_sample_rate'


def orca_qa_flow() -> Flow:
    submission_ref = workflow_input(FlowInputs.Submission)
    task_restrictions_ref = workflow_input(FlowInputs.TaskRestrictions)
    orca_qa_sample_rate_ref = workflow_input(FlowInputs.OrcaQASampleRate)

    orca_qa_block = Block(
        identifier='VISION_LANGUAGE_MODEL_QA_2',
        reference_name='orca_qa',
        input={
            'submission': submission_ref,
            'sample_rate': orca_qa_sample_rate_ref,
            'task_restrictions': task_restrictions_ref,
        },
        title='ORCA Quality Assurance',
        description='Quality assurance for ORCA document processing',
    )

    return Flow(
        uuid=UUID(ORCA_QA_FLOW_UUID),
        owner_email='flows.sdk@hyperscience.com',
        title=ORCA_QA_FLOW_TITLE,
        manifest=Manifest(
            identifier=ORCA_QA_FLOW_IDENTIFIER,
            input=[
                Parameter(
                    name=FlowInputs.Submission,
                    type='Submission',
                    title='Submission Object',
                    ui={'hidden': True},
                ),
                Parameter(
                    name=FlowInputs.TaskRestrictions,
                    type='TaskRestrictions',
                    title='Task Restrictions',
                    description=(
                        'Defines what users can access Supervision tasks created by this flow.'
                    ),
                    optional=True,
                    ui={'hidden': True},
                ),
                Parameter(
                    name=FlowInputs.OrcaQASampleRate,
                    type='Percentage',
                    schema={'type': 'number', 'minimum': 0, 'maximum': 1},
                    title='ORCA QA Sample Rate',
                    description='Defines the percentage of documents sampled for QA.',
                    value=1,
                    ui={'hidden': True},
                ),
            ],
            roles=['supporting', 'quality_assurance', 'read_only'],
            enable_overrides=True,
        ),
        blocks=[orca_qa_block],
        description=ORCA_QA_FLOW_DESCRIPTION,
    )


def entry_point_workflow() -> Flow:
    return orca_qa_flow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

OnError flow V42

This example shows how to produce an error message when a flow fails. The package includes the source code for an On-Error flow and its helper subflow. The On-Error flow should be configured to the IDP flow. The helper subflow generates the error message to be sent to the output blocks.

With the on-error flow configured, when the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered.

on_error_flow_v42.zip

OnError flow V42

from uuid import UUID

from flows_sdk.blocks import Block
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v42.additional_blocks import (
    JSONOutputsBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.types import CompatibilitySpec
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

ON_ERROR_FLOW_IDENTIFIER = 'ON_ERROR_3_EXAMPLE'
ON_ERROR_FLOW_UUID = str(str_to_deterministic_uuid_4(ON_ERROR_FLOW_IDENTIFIER))
ON_ERROR_TITLE = 'On Error V3 - flows-sdk example'
ON_ERROR_DESCRIPTION = (
    'Flow designed to be used as an "on-error" handler for other flows.'
    '\n\n'
    'With the on-error flow configured, when the original flow fails for any reason, and all '
    'the configured auto-retries have been exhausted, the on-error flow will be triggered.'
    '\n\n'
    'The input it receives is the ID of the failed flow run, which can be used to query '
    'additional information about the failure.'
    '\n\n'
    'If you need more complex logic in this flow, '
    'please consult the documentation of Flows SDK (https://flows-sdk.hyperscience.ai) '
    'and the available public APIs (https://docs.hyperscience.ai). Note that the custom logic '
    'should be implemented in the On-Error Subflow, so that it can be safely updated without '
    'losing the configured Output blocks. The Output blocks will receive the output of the '
    'On-Error Subflow as input.'
)


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)
ON_ERROR_CORE_FLOW_IDENTIFIER = 'ON_ERROR_CORE_EXAMPLE'

def on_error_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)
    on_error_core = Block(
        reference_name='on_error_core',
        identifier=ON_ERROR_CORE_FLOW_IDENTIFIER,
        input={'failed_run_uuid': flow_run_uuid},
        compatibility_spec=CompatibilitySpec(filter_roles=['on_error_core']),
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': on_error_core.output()})

    return Flow(
        uuid=UUID(ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=ON_ERROR_TITLE,
        description=ON_ERROR_DESCRIPTION,
        manifest=Manifest(
            identifier=ON_ERROR_FLOW_IDENTIFIER,
            roles=['on_error', 'supporting'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER],
        ),
        output={},
        input={},
        blocks=[
            on_error_core,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return on_error_flow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

OnError Subflow V42

from uuid import UUID

from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v42.additional_blocks import (
    HyperscienceRestApiBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

ON_ERROR_CORE_FLOW_IDENTIFIER = 'ON_ERROR_CORE_EXAMPLE'
ON_ERROR_CORE_FLOW_UUID = str(str_to_deterministic_uuid_4(ON_ERROR_CORE_FLOW_IDENTIFIER))
ON_ERROR_CORE_TITLE = 'Retrieve error information - flows-sdk example'
ON_ERROR_CORE_DESCRIPTION = (
    'Core logic of the on-error flow. Compiles the data, relevant to the error. This includes '
    'metadata for the failed flow run, without any inputs, outputs or logs, which get excluded '
    'for performance reasons - their size is unpredictable.'
    '\n\n'
    'Any custom error handling logic should be added here. The main on-error flow should only be '
    'used for configuring the outputs in the Outputs block, i.e. where the compiled on-error data '
    'should be sent to.'
)


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'

FAILED_FLOW_RUN_UUID_PARAMETER = Parameter(
    name=OnErrorFlowInputKeys.failed_run_uuid,
    type='string',
    title='Failed Flow Run UUID',
    description='UUID of the failed flow run that triggered this on_error flow',
    optional=False,
    ui={'hidden': True},
)

def on_error_core_flow() -> Flow:
    flow_run_uuid = workflow_input(OnErrorFlowInputKeys.failed_run_uuid)
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{flow_run_uuid}/summary',
    )

    return Flow(
        uuid=UUID(ON_ERROR_CORE_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.ai',
        title=ON_ERROR_CORE_TITLE,
        description=ON_ERROR_CORE_DESCRIPTION,
        manifest=Manifest(
            identifier=ON_ERROR_CORE_FLOW_IDENTIFIER,
            roles=['on_error_core', 'supporting', 'read_only'],
            input=[FAILED_FLOW_RUN_UUID_PARAMETER],
        ),
        output={'flow_run': flow_run_rest_api_block.output('result.data')},
        input={},
        blocks=[flow_run_rest_api_block],
    )


def entry_point_workflow() -> Flow:
    return on_error_core_flow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

V41

Name	Description	Release	Sample Docs	Download
Document Processing Starter V41	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v41.zip
OnError flow	Example implementation of an OnError flow that is triggered once the original flow fails.			on_error_flow_v41.zip

Document Processing Starter V41

The basic document processing flow. From V38 on, the IDP Document processing is split into a wrapper flow (IDP) and a subflow (IDP Core). The package includes:

• IDP Core Flow - a read-only subflow containing the steps to process documents into machine-readable output

• Document Processing Flow - a wrapper around the IDP Core flow, used for managing input and output connector configuration and document processing workflow settings

• IDP Submission Notify Flow - A flow that can be configured for some blocks of the IDP Core Flow, used to send notifications to external systems when a submission has been created or is waiting for supervision

• Submission-aware On Error Flow - When the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered. The submission-aware on-error flow enriches the error message with submission data (compared with the plain OnError flow shown below). In case this OnError implementation is used on a non-IDP flow (i.e. a submissionless flow), submission data will be omitted

The package includes also release and sample documents for processing Form W-9s.

Download idp_flow_v41.zip

Document Processing Sample

from uuid import UUID

from flows_sdk.error_handling import ErrorHandling
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v41 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v41.idp_blocks import IDPCoreBlock
from flows_sdk.implementations.idp_v41.idp_values import (
    DEFAULT_IDP_CORE_IDENTIFIER,
    DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER,
    IdpWorkflowConfig,
    IDPWrapperManifest,
    Settings,
    get_idp_wf_config,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4

IDP_IDENTIFIER = 'DOCUMENT_PROCESSING_EXAMPLE'
IDP_UUID = str(str_to_deterministic_uuid_4(IDP_IDENTIFIER))
IDP_TITLE = 'Document Processing - flows-sdk example'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:
    # By default, all the notification flows are set to `None`.
    # I.e. the system will send no intermediate notifications for submission change events.
    # So let's specify the notification flows that we want to use explicitly here:
    idp_wf_config.notification_workflow_initialization = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_classification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_identification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_transcription = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_flexible_extraction = (
        DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    )

    idp_flow = IDPCoreBlock(idp_wf_config, identifier=DEFAULT_IDP_CORE_IDENTIFIER)

    outputs = idp_blocks.IDPOutputsBlock(inputs={'submission': idp_flow.output('submission')})

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_TITLE,
        description='Use this flow to manage your input and output connector configuration and '
        'document processing workflow settings.',
        manifest=IDPWrapperManifest(IDP_IDENTIFIER),
        output={'submission': idp_flow.output('submission')},
        triggers=idp_values.IDPTriggers(),
        input={
            Settings.LayoutReleaseUuid: None,
        },
        blocks=[
            idp_flow,
            outputs,
        ],
        error_handling=ErrorHandling(on_error_flow=None),
    )


def entry_point_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Core V41 Sample

from uuid import UUID

from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v41 import additional_blocks, idp_blocks
from flows_sdk.implementations.idp_v41.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED
from flows_sdk.implementations.idp_v41.idp_values import IDPCoreManifest
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into IDPCoreBlock(identifier=...) within your flow.
IDP_CORE_IDENTIFIER = f'IDP_CORE_{FLOWS_VERSION}_EXAMPLE'
IDP_CORE_UUID = str(str_to_deterministic_uuid_4(IDP_CORE_IDENTIFIER))
IDP_CORE_TITLE = f'Document Processing Subflow {FLOWS_VERSION_DOTTED} - flows-sdk example'


def idp_workflow() -> Flow:
    file_filter = additional_blocks.FileFilterBlock(reference_name='file_filter')

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        trigger=file_filter.output('result.filtered_trigger'),
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction.output('submission'),
    )

    document_rendering = idp_blocks.DocumentRendererBlock(
        reference_name='document_rendering',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=document_rendering.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    return Flow(
        uuid=UUID(IDP_CORE_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_CORE_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output.',
        manifest=IDPCoreManifest(
            flow_identifier=IDP_CORE_IDENTIFIER, roles=['read_only', 'supporting', 'idp_core']
        ),
        output={'submission': submission_complete.output('submission')},
        blocks=[
            file_filter,
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            flexible_extraction,
            reprocessing,
            document_rendering,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Submission Notify Sample

import os
from uuid import UUID

from flows_sdk.blocks import Outputs
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import icon_path_to_base64, str_to_deterministic_uuid_4, workflow_input

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

IDP_SUBMISSION_NOTIFY_IDENTIFIER = 'IDP_SUBMISSION_NOTIFY_EXAMPLE'
IDP_SUBMISSION_NOTIFY_UUID = str(str_to_deterministic_uuid_4(IDP_SUBMISSION_NOTIFY_IDENTIFIER))
IDP_SUBMISSION_NOTIFY_TITLE = 'Submission State Notifications - flows-sdk example'
IDP_OUTPUT_ROLE = 'idp_output'


class SubmissionNotifyInputKeys:
    submission: str = 'submission'


class SubmissionNotifyManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['notifications', 'supporting'],
            input=[
                Parameter(
                    name=SubmissionNotifyInputKeys.submission,
                    type='Submission',
                    title='Submission Object',
                    ui={
                        'hidden': True,
                    },
                )
            ],
            ui={
                'icon': icon_path_to_base64(os.path.join(BASE_DIR, 'submission_notify_icon.png')),
            },
        )


def idp_submission_notify_workflow() -> Flow:
    return Flow(
        title=IDP_SUBMISSION_NOTIFY_TITLE,
        owner_email='flows.sdk@hyperscience.com',
        description=(
            'Send notifications to external systems when a submission has been '
            'created or is waiting for supervision. \n'
            f'If the "Document Processing" flow is live, this flow '
            'must also be live, but it can be empty.'
        ),
        manifest=SubmissionNotifyManifest(IDP_SUBMISSION_NOTIFY_IDENTIFIER),
        blocks=[
            Outputs(
                reference_name='outputs',
                title='Outputs',
                description=(
                    'Send submission data to external systems when a submission '
                    'has been created or is waiting for supervision'
                ),
                role_filter=[IDP_OUTPUT_ROLE],
                input_template={
                    'submission': {'id': workflow_input('submission.id')},
                    'enabled': True,
                },
                blocks=[],
            )
        ],
        uuid=UUID(IDP_SUBMISSION_NOTIFY_UUID),
        input={},
        output={},
    )


def entry_point_workflow() -> Flow:
    return idp_submission_notify_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

Submission-aware On Error Flow Sample

from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v41.additional_blocks import (
    HyperscienceRestApiBlock,
    JSONOutputsBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsTask
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER = f'SUBMISSION_AWARE_ON_ERROR_FLOW_EXAMPLE'
SUBMISSION_AWARE_ON_ERROR_FLOW_UUID = str(
    str_to_deterministic_uuid_4(SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER)
)
SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE = f'On-Error with included Submission data - flows-sdk example'
SUBMISSION_AWARE_ON_ERROR_DESCRIPTION = '''Flow designed to be used as an "on-error" handler for
 submission-based flows.\n
With the on-error flow configured, when the original flow fails for any reason, and all
    the configured auto-retries have been exhausted, the on-error flow will be triggered.
\n
In this on-error flow implementation,
    we fetch the full metadata of the failed flow run, alongside with the full submission metadata
    (if the submission ID was found), and then send it to a user-configured output.
    The data format of the output conforms to
    the schema documented here: https://docs.hyperscience.ai/#flows-runs and
    here: https://docs.hyperscience.ai/#submissions, under the respective keys
    "flow_run" and "submission".
\n
If you need more complex logic in this flow
    please consult the documentation of Flows SDK (https://flows-sdk.hyperscience.ai)
    and the available public APIs (https://docs.hyperscience.ai)
'''  # noqa


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'


class OnErrorManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['on_error', 'supporting'],
            input=[
                Parameter(
                    name=OnErrorFlowInputKeys.failed_run_uuid,
                    type='string',
                    title='Failed Flow Run UUID',
                    description='UUID of the failed flow run that triggered this on_error flow',
                    optional=False,
                    ui={'hidden': True},
                )
            ],
        )


def submission_aware_on_error_flow() -> Flow:
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        title='Get Flow Run',
        description='Get the full metadata of the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{workflow_input(OnErrorFlowInputKeys.failed_run_uuid)}',
    )

    def _get_correlation_id(_hs_task: HsTask) -> dict:
        return {'correlation_id': _hs_task.correlation_id}

    get_correlation_id_block = PythonBlock(
        reference_name='get_correlation_id',
        title='Get Correlation ID',
        description='Get the correlation ID for the failed Submission / Flow Run',
        code=_get_correlation_id,
    )

    fetch_submissions_block = HyperscienceRestApiBlock(
        reference_name='fetch_submission',
        title='Fetch Submission',
        description='Fetch the submission data for the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/submissions'
        '?debug=true'
        f'&correlation_id={get_correlation_id_block.output("correlation_id")}',
    )

    def _prep_output_data(flow_run: dict, get_submissions_response: dict) -> dict:
        submissions_in_response = get_submissions_response.get('results', [])
        submission = None if not submissions_in_response else submissions_in_response[0]
        return {'flow_run': flow_run, 'submission': submission}

    prep_output_data_block = PythonBlock(
        title='Format Submission Response',
        description='Prepare the output data to be sent to the user-configured output'
        'under the keys "flow_run" and "submission"',
        code=_prep_output_data,
        code_input={
            'flow_run': flow_run_rest_api_block.output('result.data'),
            'get_submissions_response': fetch_submissions_block.output('result.data'),
        },
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': prep_output_data_block.output()})

    return Flow(
        uuid=UUID(SUBMISSION_AWARE_ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE,
        description=SUBMISSION_AWARE_ON_ERROR_DESCRIPTION,
        manifest=OnErrorManifest(SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER),
        output={},
        input={},
        blocks=[
            flow_run_rest_api_block,
            get_correlation_id_block,
            fetch_submissions_block,
            prep_output_data_block,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return submission_aware_on_error_flow()


if __name__ == '__main__':
    export_flow(submission_aware_on_error_flow())

V40

Name	Description	Release	Sample Docs	Download
Document Processing Starter V40	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v40.zip
OnError flow	Example implementation of an OnError flow that is triggered once the original flow fails.			on_error_flow_v40.zip

Document Processing Starter V40

The basic document processing flow. From V38 on, the IDP Document processing is split into a wrapper flow (IDP) and a subflow (IDP Core). The package includes:

• IDP Core Flow - a read-only subflow containing the steps to process documents into machine-readable output

• Document Processing Flow - a wrapper around the IDP Core flow, used for managing input and output connector configuration and document processing workflow settings

• IDP Submission Notify Flow - A flow that can be configured for some blocks of the IDP Core Flow, used to send notifications to external systems when a submission has been created or is waiting for supervision

• Submission-aware On Error Flow - When the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered. The submission-aware on-error flow enriches the error message with submission data (compared with the plain OnError flow shown below). In case this OnError implementation is used on a non-IDP flow (i.e. a submissionless flow), submission data will be omitted

The package includes also release and sample documents for processing Form W-9s.

Download idp_flow_v40.zip

Document Processing Sample

from uuid import UUID

from flows_sdk.error_handling import ErrorHandling
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v40 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v40.idp_blocks import IDPCoreBlock
from flows_sdk.implementations.idp_v40.idp_values import (
    DEFAULT_IDP_CORE_IDENTIFIER,
    DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER,
    IdpWorkflowConfig,
    IDPWrapperManifest,
    Settings,
    get_idp_wf_config,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4

IDP_IDENTIFIER = 'DOCUMENT_PROCESSING_EXAMPLE'
IDP_UUID = str(str_to_deterministic_uuid_4(IDP_IDENTIFIER))
IDP_TITLE = 'Document Processing - flows-sdk example'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:
    # By default, all the notification flows are set to `None`.
    # I.e. the system will send no intermediate notifications for submission change events.
    # So let's specify the notification flows that we want to use explicitly here:
    idp_wf_config.notification_workflow_initialization = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_classification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_identification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_transcription = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_flexible_extraction = (
        DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    )

    idp_flow = IDPCoreBlock(idp_wf_config, identifier=DEFAULT_IDP_CORE_IDENTIFIER)

    outputs = idp_blocks.IDPOutputsBlock(inputs={'submission': idp_flow.output('submission')})

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_TITLE,
        description='Use this flow to manage your input and output connector configuration and '
        'document processing workflow settings.',
        manifest=IDPWrapperManifest(IDP_IDENTIFIER),
        output={'submission': idp_flow.output('submission')},
        triggers=idp_values.IDPTriggers(),
        input={
            Settings.LayoutReleaseUuid: None,
        },
        blocks=[
            idp_flow,
            outputs,
        ],
        error_handling=ErrorHandling(on_error_flow=None),
        # You can use the following error handling configuration to enable the submission aware on
        # error flow if your system has either Advanced or Premium license package
        # error_handling=ErrorHandling(
        #    on_error_flow=OnErrorFlow(identifier=DEFAULT_SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER)
        # ),
    )


def entry_point_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Core V40 Sample

from uuid import UUID

from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v40 import idp_blocks
from flows_sdk.implementations.idp_v40.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED
from flows_sdk.implementations.idp_v40.idp_values import IDPCoreManifest
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into IDPCoreBlock(identifier=...) within your flow.
IDP_CORE_IDENTIFIER = f'IDP_CORE_{FLOWS_VERSION}_EXAMPLE'
IDP_CORE_UUID = str(str_to_deterministic_uuid_4(IDP_CORE_IDENTIFIER))
IDP_CORE_TITLE = f'Document Processing Subflow {FLOWS_VERSION_DOTTED} - flows-sdk example'


def idp_workflow() -> Flow:
    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction.output('submission'),
    )

    document_rendering = idp_blocks.DocumentRendererBlock(
        reference_name='document_rendering',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=document_rendering.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    return Flow(
        uuid=UUID(IDP_CORE_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_CORE_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output.',
        manifest=IDPCoreManifest(
            flow_identifier=IDP_CORE_IDENTIFIER, roles=['read_only', 'supporting', 'idp_core']
        ),
        output={'submission': submission_complete.output('submission')},
        blocks=[
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            flexible_extraction,
            reprocessing,
            document_rendering,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Submission Notify Sample

import os
from uuid import UUID

from flows_sdk.blocks import Outputs
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import icon_path_to_base64, str_to_deterministic_uuid_4, workflow_input

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

IDP_SUBMISSION_NOTIFY_IDENTIFIER = 'IDP_SUBMISSION_NOTIFY_EXAMPLE'
IDP_SUBMISSION_NOTIFY_UUID = str(str_to_deterministic_uuid_4(IDP_SUBMISSION_NOTIFY_IDENTIFIER))
IDP_SUBMISSION_NOTIFY_TITLE = 'Submission State Notifications - flows-sdk example'
IDP_OUTPUT_ROLE = 'idp_output'


class SubmissionNotifyInputKeys:
    submission: str = 'submission'


class SubmissionNotifyManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['notifications', 'supporting'],
            input=[
                Parameter(
                    name=SubmissionNotifyInputKeys.submission,
                    type='Submission',
                    title='Submission Object',
                    ui={
                        'hidden': True,
                    },
                )
            ],
            ui={
                'icon': icon_path_to_base64(os.path.join(BASE_DIR, 'submission_notify_icon.png')),
            },
        )


def idp_submission_notify_workflow() -> Flow:
    return Flow(
        title=IDP_SUBMISSION_NOTIFY_TITLE,
        owner_email='flows.sdk@hyperscience.com',
        description=(
            'Send notifications to external systems when a submission has been '
            'created or is waiting for supervision. \n'
            f'If the "Document Processing" flow is live, this flow '
            'must also be live, but it can be empty.'
        ),
        manifest=SubmissionNotifyManifest(IDP_SUBMISSION_NOTIFY_IDENTIFIER),
        blocks=[
            Outputs(
                reference_name='outputs',
                title='Outputs',
                description=(
                    'Send submission data to external systems when a submission '
                    'has been created or is waiting for supervision'
                ),
                role_filter=[IDP_OUTPUT_ROLE],
                input_template={
                    'submission': {'id': workflow_input('submission.id')},
                    'enabled': True,
                },
                blocks=[],
            )
        ],
        uuid=UUID(IDP_SUBMISSION_NOTIFY_UUID),
        input={},
        output={},
    )


def entry_point_workflow() -> Flow:
    return idp_submission_notify_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

Submission-aware On Error Flow Sample

from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v40.additional_blocks import (
    HyperscienceRestApiBlock,
    JSONOutputsBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsTask
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER = f'SUBMISSION_AWARE_ON_ERROR_FLOW_EXAMPLE'
SUBMISSION_AWARE_ON_ERROR_FLOW_UUID = str(
    str_to_deterministic_uuid_4(SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER)
)
SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE = f'On-Error with included Submission data - flows-sdk example'
SUBMISSION_AWARE_ON_ERROR_DESCRIPTION = """Flow designed to be used as an "on-error" handler for
 submission-based flows.\n
With the on-error flow configured, when the original flow fails for any reason, and all
    the configured auto-retries have been exhausted, the on-error flow will be triggered.
\n
In this on-error flow implementation,
    we fetch the full metadata of the failed flow run, alongside with the full submission metadata
    (if the submission ID was found), and then send it to a user-configured output.
    The data format of the output conforms to
    the schema documented here: https://docs.hyperscience.ai/#flows-runs and
    here: https://docs.hyperscience.ai/#submissions, under the respective keys
    "flow_run" and "submission".
\n
If you need more complex logic in this flow
    please consult the documentation of Flows SDK (https://flows-sdk.hyperscience.ai)
    and the available public APIs (https://docs.hyperscience.ai)
"""


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'


class OnErrorManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['on_error', 'supporting'],
            input=[
                Parameter(
                    name=OnErrorFlowInputKeys.failed_run_uuid,
                    type='string',
                    title='Failed Flow Run UUID',
                    description='UUID of the failed flow run that triggered this on_error flow',
                    optional=False,
                    ui={'hidden': True},
                )
            ],
        )


def submission_aware_on_error_flow() -> Flow:
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        title='Get Flow Run',
        description='Get the full metadata of the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{workflow_input(OnErrorFlowInputKeys.failed_run_uuid)}',
    )

    def _get_correlation_id(_hs_task: HsTask) -> dict:
        return {'correlation_id': _hs_task.correlation_id}

    get_correlation_id_block = PythonBlock(
        reference_name='get_correlation_id',
        title='Get Correlation ID',
        description='Get the correlation ID for the failed Submission / Flow Run',
        code=_get_correlation_id,
    )

    fetch_submissions_block = HyperscienceRestApiBlock(
        reference_name='fetch_submission',
        title='Fetch Submission',
        description='Fetch the submission data for the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/submissions'
        '?debug=true'
        f'&correlation_id={get_correlation_id_block.output("correlation_id")}',
    )

    def _prep_output_data(flow_run: dict, get_submissions_response: dict) -> dict:
        submissions_in_response = get_submissions_response.get('results', [])
        submission = None if not submissions_in_response else submissions_in_response[0]
        return {'flow_run': flow_run, 'submission': submission}

    prep_output_data_block = PythonBlock(
        title='Format Submission Response',
        description='Prepare the output data to be sent to the user-configured output'
        'under the keys "flow_run" and "submission"',
        code=_prep_output_data,
        code_input={
            'flow_run': flow_run_rest_api_block.output('result.data'),
            'get_submissions_response': fetch_submissions_block.output('result.data'),
        },
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': prep_output_data_block.output()})

    return Flow(
        uuid=UUID(SUBMISSION_AWARE_ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE,
        description=SUBMISSION_AWARE_ON_ERROR_DESCRIPTION,
        manifest=OnErrorManifest(SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER),
        output={},
        input={},
        blocks=[
            flow_run_rest_api_block,
            get_correlation_id_block,
            fetch_submissions_block,
            prep_output_data_block,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return submission_aware_on_error_flow()


if __name__ == '__main__':
    export_flow(submission_aware_on_error_flow())

Code Snippets

Note

While code snippets are great, the Hyperscience Platform provides easy tools to help you build your own solutions. Make sure to read our Testing & Debugging guide below before diving in.

How do I iterate over multiple documents in a submission?

Block subclasses provided in IDP Library generally take a Submission object as a mandatory input parameter as well as output a modified copy of the Submission object. Each block provides a convenience method to access the Submission object:

# This example shows the output of MachineTranscriptionBlock (i.e. machine_transcription) being used as the input to a ManualTranscriptionBlock
manual_transcription = ManualTranscriptionBlock(
  reference_name='manual_transcription',
  submission=machine_transcription.output('submission'),
  ...
)

Warning

Always check the exact output of the preceding block! Some block output includes additional metadata and may require a different key, such as .output(‘results.submission’), to access the Submission object.

The Submission object is constructed as:

{
   "submission": {
      "id": "12345",
      "documents": [
         {
             "id": "1",
             "pages": [...] // list of pages
             "document_fields": [...] // list of document fields
         },
         ...
      ]
      "cases": [],
   }
}

And so iterating over multiple documents in a submission is as easy as:

def _iter_docs(submission):
   for document in submission.get('documents', []):
      print(document["id"]) # Print the ID of each document in the submission
   return submission

custom_code = CodeBlock(
   reference_name='code_block',
   code=_iter_docs,
   code_input={'submission': manual_transcription.output('submission')}
)

How do I compare fields across different documents?

Let’s assume we’re trying to find the maximum value of the “income” field across multiple documents in the same submission. Once transcribed, we iterate over all the transcribed fields in the submission and compare each value to the previous maximum:

def _compare_fields(submission):
   max_income = 0.0
   for document in submission.get('documents', []):
      for field in document.get('document_fields', []):
         if field.get('field_name') == 'income' and float(field.get('transcription')) > max_income:
            max_income = float(field.get('transcription'))

      print(max_income) # Print the maximum income across all documents
   return submission

custom_code = CodeBlock(
   reference_name='code_block',
   code=_compare_fields,
   code_input={'submission': previous_block.output('submission')}
)

How do I store binary runtime data in the system?

Available in v35 and later. A code block may need to store some binary data such as images in order to transfer it to another block in the flow. To facilitate this code blocks provide the _hs_block_instance system argument that exposes the store_blob and fetch_blob methods (for more info see flows_sdk.types.HsBlockInstance). store_blob will store the passed in binary data and return an object that contains an identifier that can later be used to retrieve it using fetch_blob. This API can be used to store images and any type of data serialized into binary format, for example large JSONs or other text-based documents. See example code below:

from flows_sdk.types import HsBlockInstance
from flows_sdk.utils import workflow_input

def _transform_image(image_reference: str, _hs_block_instance: HsBlockInstance):
    from flows_sdk.types import StoreBlobRequest

    original_image = _hs_block_instance.fetch_blob(image_reference)

    transformed_image = transform_image_lib_function(original_image.content)

    stored_blob = _hs_block_instance.store_blob(
            StoreBlobRequest(name=f'transformed-{image_reference}', content=transformed_image)
        )
    # the returned StoreBlobResponse has name and uuid parameters
    return stored_blob.uuid

transform_image = CodeBlock(
   reference_name='transform_image',
   code=_transform_image,
   code_input={'image_reference': workflow_input('input_image_uuid')}
)

def _use_transformed_image(blob_reference: str, _hs_block_instance: HsBlockInstance) -> Any:
    transformed_image = _hs_block_instance.fetch_blob(blob_reference)
    return len(transformed_image.content)

use_transformed_image = CodeBlock(
    reference_name='use_transformed_image',
    code=_use_transformed_image,
    code_input={'blob_reference': transform_image.output()}
)

Note

System arguments don’t need to be present in code_input, just the function definition. They will be automatically populated by the block at runtime.

How do I access flow metadata inside of a code block?

Available in v35 and later. Each flow run is associated with its unique execution context in the system that is described by certain metadata. Code blocks provide access to this metadata through the _hs_task system argument (for more info see flows_sdk.types.HsTask). Example usage:

from flows_sdk.types import HsTask

def use_flow_metadata(in_data: dict, _hs_task: HsTask):
    return f"{_hs_task.correlation_id}-{_hs_task.flow_run_id}-{in_data['value']}"

custom_code = CodeBlock(
   reference_name='use_flow_metadata',
   code=use_flow_metadata,
   code_input={'in_data': another_block.output('data')}
)

Note

System arguments don’t need to be present in code_input, just the function definition. They will be automatically populated by the block at runtime.

How do I log from a code block?

Available in v35 and later. By default, messages from a generic logger / print statements are visible in the application container. However, they do not appear in the UI, nor do they contain the details that are typically included in log message (structured logging - e.g., ID of the submission being processed).

To add these features to your messages, you can use the _hs_block_instance parameter, which provides the log method (for more details, see flows_sdk.types.HsBlockInstance.log)

from flows_sdk.types import HsBlockInstance
from flows_sdk.utils import workflow_input

# we need a _hs_block_instance to call the logging methods
def log_text(text: str, _hs_block_instance: HsBlockInstance) -> None:
    _hs_block_instance.log(f'INFO level: {text}')  # by default, logs have the INFO level.
    # raise Exception('something went wrong')  # this will log an exception if uncommented

log_task_ccb = CodeBlock(
    reference_name='log_task_ccb',
    code=log_text,
    code_input={FlowInputs.TEXT_TO_LOG: workflow_input(FlowInputs.TEXT_TO_LOG)},
)

Available in v42.2 and later. The log method also allows for arguments and keyword arguments to format the log message, using {} style formatting:

from flows_sdk.types import HsBlockInstance

# we need a _hs_block_instance to call the logging methods
def log_text(_hs_block_instance: HsBlockInstance) -> None:
    msg_1 = "This is a debug message"
    msg_2 = "This is an info message"
    _hs_block_instance.log('INFO level: {}', msg_1)
    _hs_block_instance.log('DEBUG level: {text}', text=msg_2, level=HsBlockInstance.LogLevel.DEBUG)

log_task_ccb = CodeBlock(
    reference_name='log_task_ccb',
    code=log_text,
)

You can also download this example flow that showcases logging.

Logging flow

import sys
from uuid import UUID

from flows_sdk.blocks import CodeBlock, PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsBlockInstance
from flows_sdk.utils import workflow_input

LOGGING_FLOW_IDENTIFIER = 'LOGGING_FLOW'
LOGGING_FLOW_UUID = UUID('cee432a8-30a3-4d07-a924-e6d87c923325')


def entry_point_flow() -> Flow:
    return sample_logging_flow()


class FlowInputs:
    TEXT_TO_LOG = 'text'


def sample_logging_flow() -> Flow:
    # we need a _hs_block_instance to call the logging methods
    def log_text(text: str, _hs_block_instance: HsBlockInstance) -> None:
        _hs_block_instance.log(f'DEBUG level: {text}', HsBlockInstance.LogLevel.DEBUG)
        _hs_block_instance.log(f'INFO level: {text}', HsBlockInstance.LogLevel.INFO)
        _hs_block_instance.log(f'WARNING level: {text}', HsBlockInstance.LogLevel.WARN)
        _hs_block_instance.log(f'ERROR level: {text}', HsBlockInstance.LogLevel.ERROR)
        # raise Exception('something went wrong')  # this will log an exception if uncommented

    log_task_ccb = CodeBlock(
        reference_name='log_task_ccb',
        code=log_text,
        code_input={FlowInputs.TEXT_TO_LOG: workflow_input(FlowInputs.TEXT_TO_LOG)},
    )
    log_task_python = PythonBlock(
        reference_name='log_task_python',
        code=log_text,
        code_input={FlowInputs.TEXT_TO_LOG: workflow_input(FlowInputs.TEXT_TO_LOG)},
    )

    return Flow(
        title='Logging sample flow',
        description='A simple Flow showcasing how to log in code blocks',
        blocks=[log_task_ccb, log_task_python],
        owner_email='flows.sdk@hyperscience.com',
        manifest=Manifest(
            identifier=LOGGING_FLOW_IDENTIFIER,
            input=[
                (
                    Parameter(
                        name=FlowInputs.TEXT_TO_LOG,
                        title='Text to log',
                        type='string',
                        optional=False,
                    )
                )
            ],
        ),
        uuid=LOGGING_FLOW_UUID,
        input={FlowInputs.TEXT_TO_LOG: 'default-text'},
    )


if __name__ == '__main__':
    export_filename = None
    if len(sys.argv) > 1:
        export_filename = sys.argv[1]

    export_flow(flow=entry_point_flow(), filename=export_filename)

How do I access an email stored by the Email Listener?

Available in v41 and later. When configuring an email listener, it can be set to store the original email as a blob before processing it. To access this blob inside a code block (using the _hs_block_instance system argument), you can use the fetch_blob method (for more info see flows_sdk.types.HsBlockInstance). In order to do this you will need the uuid of the stored blob, which can be accessed from the email listener block’s output.

Example that fetches the email blob and attempts to forward it:

from flows_sdk.types import HsBlockInstance
from flows_sdk.utils import workflow_input

def send_email_func(trigger, _hs_block_instance: HsBlockInstance):
    # in this example we are using smtplib in order to send the email via SMTP
    import smtplib
    from email import message_from_bytes
    to = 'email@example.com'

    # blob metadata is stored in the 'raw_email' field of the trigger payload
    raw_email_metadata = trigger.get('payload', {}).get('raw_email')
    if raw_email_metadata:
        blob = _hs_block_instance.fetch_blob(raw_email_metadata['uuid'])
        msg = message_from_bytes(blob.content)
        # using a simple local SMTP server for example purposes
        with smtplib.SMTP('localhost', 8025) as smtp_server:
            smtp_server.sendmail(to, [to], msg.as_string())
    else:
        _hs_block_instance.log('no raw email found in trigger')
    return

send_email = PythonBlock(
    reference_name='send_email',
    code=send_email_func,
    # the workflow receives the email listener's output under '$trigger'
    code_input={'trigger': workflow_input('$trigger')}
)

How do I access attachment metadata stored by the Email Listener?

Available in v41 and later. When configuring an email listener, it can be set to record the headers of each attachment. This data can be accessed in a code block from the email listener block’s output.

Example that processes this metadata to determine whether a file attachment is inline or not:

from flows_sdk.types import HsBlockInstance
from flows_sdk.utils import workflow_input

def is_inline_func(trigger, _hs_block_instance: HsBlockInstance):
    # attachment metadata is stored in the 'origin_metadata' part of the trigger payload
    attachment_metadata = trigger.get('payload', {}).get('origin_metadata', {}).get('additional_props', {}).get('attachment_metadata')
    if attachment_metadata:
        for data in attachment_metadata:
            # determine the uuid of the stored attachment; we can use this to fetch the blob itself if needed
            file_uuid = data['blob_uuid']
            # check the Content-Disposition header as it has this information
            disposition = data['headers'].get('Content-Disposition', '')
            inline = 'inline' in disposition.lower()
            _hs_block_instance.log(f"File with uuid: {file_uuid} is inline: {inline}.")
    else:
        _hs_block_instance.log('no attachment metadata found in trigger')
    return

is_inline = PythonBlock(
    reference_name='is_inline',
    code=is_inline_func,
    # the workflow receives the email listener's output under '$trigger'
    code_input={'trigger': workflow_input('$trigger')}
)

How do I parse nested/forwarded emails and deduplicate attachments?

Available in v41 and later. When emails arrive as multi-level forwards, the platform’s email listener may expose the same attachment at every nesting level. This example shows a complete PythonBlock implementation (parse_emails) that:

• recursively unpacks nested message/rfc822 and .msg parts

• deduplicates attachments by SHA-256 digest across all nesting levels

• strips inline signature images from body blobs, emitting each once as a standalone blob

• preserves the HTML body, original charset, and Content-Transfer-Encoding at every level

The custom block with the parsing logic is implemented at the very start of an otherwise standard IDP flow, so the rest of the IDP pipeline receives clean, deduplicated blobs. The sample also includes self-contained tests and three example .eml fixture files demonstrating the three-layer nesting scenario.

You can download the full sample (flow, tests, and fixture emails).

Custom Email Parsing Flow V41

from typing import Any
from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v41 import additional_blocks, idp_blocks
from flows_sdk.implementations.idp_v41.idp_values import IDPCoreManifest, Inputs
from flows_sdk.mocks import MockHsBlockInstance
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsBlockInstance, StoreBlobRequest
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

CUSTOM_EMAIL_PARSING_IDENTIFIER = 'CUSTOM_EMAIL_PARSING_V41'
IDP_CORE_IDENTIFIER = CUSTOM_EMAIL_PARSING_IDENTIFIER
IDP_CORE_UUID = str(str_to_deterministic_uuid_4(CUSTOM_EMAIL_PARSING_IDENTIFIER))
IDP_CORE_TITLE = 'Custom Email Parsing Subflow V41'


def parse_emails_func(  # noqa: PLR0915
    trigger: dict[str, Any],
    _hs_block_instance: HsBlockInstance,
) -> dict[str, Any]:
    import base64
    import hashlib
    import itertools
    import os
    import tempfile
    from email.message import Message

    import magic  # type: ignore[import-not-found]  # pants: no-infer-dep
    import mailparser  # type: ignore[import-not-found]  # pants: no-infer-dep

    from flows_sdk.types import HsBlockInstance, StoreBlobRequest, StoreBlobResponse

    MSG_MIME_TYPE = 'application/vnd.ms-outlook'
    EMAIL_MIME_TYPES = frozenset({'message/rfc822', MSG_MIME_TYPE})

    # ---------------------------------------------------------------------------
    # Helpers
    # ---------------------------------------------------------------------------

    def detect_mime_type(content: bytes) -> str:
        return magic.Magic(mime=True).from_buffer(content)

    def is_email_content(content: bytes) -> bool:
        return detect_mime_type(content) in EMAIL_MIME_TYPES

    def is_msg_content(content: bytes) -> bool:
        return detect_mime_type(content) == MSG_MIME_TYPE

    def get_disposition_filename(content_disposition: str) -> str | None:
        """Extract the filename parameter from a Content-Disposition header, or None if absent."""
        m = Message()
        m['content-disposition'] = content_disposition
        param = m.get_param('filename', header='content-disposition')
        if isinstance(param, tuple):
            return param[2]
        return param

    def attachment_as_bytes(attachment: dict) -> bytes:
        """Decode a mailparser attachment payload to raw bytes."""
        payload = attachment.get('payload', '')
        if attachment.get('binary', False):
            return base64.b64decode(payload)
        if isinstance(payload, str):
            return payload.encode('utf-8')
        return bytes(payload)

    def parse_bytes(content: bytes) -> Any:
        """Parse raw email bytes into a mailparser object, detecting type via libmagic."""
        if is_msg_content(content):
            # extract-msg (used by mailparser for MSG) requires a file path
            tmp_fd, tmp_path = tempfile.mkstemp(suffix='.msg')
            try:
                with os.fdopen(tmp_fd, 'wb') as f:
                    f.write(content)
                return mailparser.parse_from_file_msg(tmp_path)
            finally:
                os.remove(tmp_path)
        return mailparser.parse_from_bytes(content)

    def parse_email(  # noqa: PLR0915
            raw_email: bytes,
            blobs: list[StoreBlobResponse],
            _hs_block_instance: HsBlockInstance,
            prefix: str = '',
            _untitled_counter: itertools.count | None = None,
            _processed_email_payloads: set[str] | None = None,
    ) -> set[str]:
        """
        Parse raw email bytes, recursing into any nested emails.
        All stored blob names are prefixed with the chain of containing email filenames,
        e.g. "outer.msg:inner.eml:$email_body".
        """
        if _untitled_counter is None:
            _untitled_counter = itertools.count()
        if _processed_email_payloads is None:
            _processed_email_payloads = set()

        # Avoid re-processing the same nested email payload when parent parsers
        # expose flattened descendants in addition to direct children.
        email_payload_digest = hashlib.sha256(raw_email).hexdigest()
        if email_payload_digest in _processed_email_payloads:
            return set()
        _processed_email_payloads.add(email_payload_digest)

        parsed = parse_bytes(raw_email)

        # --- body ---
        # Build the body blob by stripping only attachment parts from the original email,
        # preserving all headers, the HTML body (with formatting), plain text body,
        # and inline images unchanged.
        #
        # compat32 policy preserves original header folding and encoding byte-for-byte.
        import email as _stdlib_email
        import io as _io
        from email import policy as _email_policy
        from email.generator import BytesGenerator as _BytesGenerator

        _orig_msg = _stdlib_email.message_from_bytes(raw_email, policy=_email_policy.compat32)

        def _is_strippable(part: Any) -> bool:
            """Return True if this MIME part should be excluded from the body blob."""
            # Explicit file attachments (PDFs, ZIPs, etc.)
            if part.get_content_disposition() == 'attachment':
                return True
            # Nested email parts (message/rfc822, message/vnd.ms-outlook, etc.)
            # These are recursed into separately and produce their own $email_body blobs.
            # Note: also catches unnamed message/* parts that might lack a Content-Disposition.
            if part.get_content_maintype() == 'message':
                return True
            # Inline non-text parts that carry a filename — these are CID-referenced media
            # (e.g. Outlook signature images embedded inside multipart/related via cid: URLs).
            # They must not be left inside every level's body blob, because the downstream
            # body-parsing logic would pick them up as separate documents and show 3 copies.
            # They are handled exactly once via the standalone-attachment deduplication path.
            if (
                part.get_content_maintype() not in ('text', 'multipart')
                and part.get_filename()
            ):
                return True
            return False

        def _strip_inplace(part: Any) -> None:
            """Recursively remove strippable subparts from a multipart message in-place."""
            if not part.is_multipart():
                return
            kept = [sp for sp in part.get_payload() if not _is_strippable(sp)]
            for sp in kept:
                _strip_inplace(sp)
            part.set_payload(kept)

        _strip_inplace(_orig_msg)

        # Only emit a body blob when there is at least one text leaf remaining.
        _has_text = any(
            lp.get_content_maintype() == 'text'
            for lp in _orig_msg.walk()
            if not lp.is_multipart()
        )
        if _has_text:
            _buf = _io.BytesIO()
            _BytesGenerator(_buf, mangle_from_=False, maxheaderlen=998).flatten(_orig_msg)
            blob = StoreBlobRequest(name=f'{prefix}$email_body', content=_buf.getvalue())
            blobs.append(_hs_block_instance.store_blob(blob))

        # --- attachments ---
        # Pre-compute per-attachment info so we can inspect the full list before acting.
        attachments_info = [
            (
                get_disposition_filename(a.get('content-disposition', '')),
                attachment_as_bytes(a),
                a.get('mail_content_type', '').lower(),
            )
            for a in parsed.attachments
        ]
        has_named_email = any(
            fname and (is_email_content(data) or ctype in EMAIL_MIME_TYPES)
            for fname, data, ctype in attachments_info
        )
        # Note: nested emails from Outlook often arrive as message/rfc822 with
        # Content-Disposition: attachment but NO filename= parameter, so fname is None/''
        # and has_named_email stays False.  That is intentional: we recurse into ALL
        # unnamed email parts via the `not has_named_email` branch below.  Unnamed email
        # parts that appear alongside a *named* email attachment are treated as body
        # artifacts (inline-forwarded content) and stored as raw bytes instead.

        nested_emails: list[tuple[str, bytes]] = []
        regular_attachments: list[tuple[str, bytes]] = []

        for disposition_filename, attachment_bytes, content_type_header in attachments_info:
            is_email = is_email_content(attachment_bytes) or content_type_header in EMAIL_MIME_TYPES
            # Recurse into named emails always; recurse into unnamed emails only when no
            # named email sibling exists (otherwise the unnamed part is a body artifact).
            if is_email and (disposition_filename or not has_named_email):
                filename = disposition_filename or f'untitled_attachment_{next(_untitled_counter)}'
                nested_emails.append((filename, attachment_bytes))
            else:
                filename = disposition_filename or f'untitled_attachment_{next(_untitled_counter)}'
                regular_attachments.append((filename, attachment_bytes))

        descendant_leaf_hashes: set[str] = set()
        for filename, attachment_bytes in nested_emails:
            descendant_leaf_hashes.update(
                parse_email(
                    attachment_bytes,
                    blobs,
                    _hs_block_instance,
                    prefix=f'{prefix}{filename}:',
                    _untitled_counter=_untitled_counter,
                    _processed_email_payloads=_processed_email_payloads,
                )
            )

        emitted_leaf_hashes: set[str] = set()
        for filename, attachment_bytes in regular_attachments:
            attachment_digest = hashlib.sha256(attachment_bytes).hexdigest()
            if attachment_digest in descendant_leaf_hashes:
                continue
            blob = StoreBlobRequest(name=f'{prefix}{filename}', content=attachment_bytes)
            blobs.append(_hs_block_instance.store_blob(blob))
            emitted_leaf_hashes.add(attachment_digest)

        return descendant_leaf_hashes | emitted_leaf_hashes

    # ---------------------------------------------------------------------------
    # Entry point
    # ---------------------------------------------------------------------------

    def parse_email_entrypoint(
        trigger: dict[str, Any],
        _hs_block_instance: HsBlockInstance,
    ) -> dict[str, Any]:
        """
        Parse a raw email file and extract its body and attachments to disk.

        Args:
            trigger:  Input from the email listener.
            _hs_block_instance:  HsBlockInstance object.

        Returns:
            Output with new.
        """
        email_file_uuid = trigger.get('payload', {}).get('raw_email').get('uuid')
        if not email_file_uuid:
            _hs_block_instance.log(
                'No raw_email found in trigger payload',
                level=HsBlockInstance.LogLevel.ERROR,
            )
            return {}
        raw_email_blob = _hs_block_instance.fetch_blob(email_file_uuid)
        blobs: list[StoreBlobResponse] = []
        parse_email(raw_email_blob.content, blobs, _hs_block_instance)
        trigger['payload']['blob_storage_files'] = blobs
        return {'parsed_trigger': trigger}

    return parse_email_entrypoint(trigger, _hs_block_instance=_hs_block_instance)


def idp_workflow() -> Flow:

    parse_emails = PythonBlock(
        reference_name='parse_emails',
        code=parse_emails_func,
        code_input={'trigger': workflow_input(Inputs.Trigger)},
    )

    file_filter = additional_blocks.FileFilterBlock(
        trigger=parse_emails.output('parsed_trigger'),
        reference_name='file_filter'
    )

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        trigger=file_filter.output('result.filtered_trigger'),
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction.output('submission'),
    )

    document_rendering = idp_blocks.DocumentRendererBlock(
        reference_name='document_rendering',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=document_rendering.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    return Flow(
        uuid=UUID(IDP_CORE_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_CORE_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output.',
        manifest=IDPCoreManifest(
            flow_identifier=IDP_CORE_IDENTIFIER, roles=['read_only', 'supporting', 'idp_core']
        ),
        output={'submission': submission_complete.output('submission')},
        blocks=[
            parse_emails,
            file_filter,
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            flexible_extraction,
            reprocessing,
            document_rendering,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_workflow()


# ============================================================================
# Test Infrastructure
# ============================================================================

def _run_parse_emails_real_libs(email_bytes: bytes) -> list[StoreBlobRequest]:
    """Run parse_emails_func with REAL libraries (magic, mailparser) on given bytes."""
    mock = MockHsBlockInstance()
    mock.blobs['root'] = email_bytes
    trigger = {'payload': {'raw_email': {'uuid': 'root'}}}
    parsed_trigger = parse_emails_func(trigger, mock)  # type: ignore[arg-type]
    blob_responses = parsed_trigger['parsed_trigger']['payload']['blob_storage_files']
    # StoreBlobResponse carries uuid+name but not content; reconstruct as StoreBlobRequest
    # (which has both name and content) for use in test assertions.
    return [StoreBlobRequest(name=r.name, content=mock.blobs[r.uuid]) for r in blob_responses]



def _load_example_1st_layer() -> bytes:
    """Load example 1st layer email."""
    from pathlib import Path
    fixtures_dir = Path(__file__).parent / 'idp_core_flow_v41_test_data'
    return (fixtures_dir / 'original' / '1st layer email with a pdf 1.eml').read_bytes()


def _load_example_2nd_layer() -> bytes:
    """Load example 2nd layer email."""
    from pathlib import Path
    fixtures_dir = Path(__file__).parent / 'idp_core_flow_v41_test_data'
    return (fixtures_dir / 'original' / 'Fw 2nd layer email with a pdf 2.eml').read_bytes()


def _load_example_3rd_layer() -> bytes:
    """Load example 3rd layer email."""
    from pathlib import Path
    fixtures_dir = Path(__file__).parent / 'idp_core_flow_v41_test_data'
    return (fixtures_dir / 'original' / '3rd layer email with a pdf 3.eml').read_bytes()


# ============================================================================
# Tests using data from idp_core_flow_v41_test_data/original/
# ============================================================================

def test_parse_emails_example_1st_layer_no_nested_duplication() -> None:
    """Test example 1st layer email doesn't duplicate nested attachments.

    The 1st layer email contains:
    - pdf1.pdf (direct attachment)
    - pdf2.pdf (nested in 2nd layer email)
    - pdf3.pdf (nested in 3rd layer email)

    With the fix, each PDF should appear exactly once despite mailparser
    potentially exposing them in a flattened list.
    """
    email_bytes = _load_example_1st_layer()
    blobs = _run_parse_emails_real_libs(email_bytes)
    names = [b.name for b in blobs]

    # The 1st layer should have pdf1 + nested pdf2/3
    # With the fix, pdf2 and pdf3 should appear only once despite flattening
    pdf1_count = sum('pdf_1' in name for name in names)
    pdf2_count = sum('pdf_2' in name for name in names)
    pdf3_count = sum('pdf_3' in name for name in names)

    # Each PDF should appear exactly once
    assert pdf1_count == 1, \
        f'Example 1st layer: expected pdf1 1x, got {pdf1_count}. Names: {names}'
    assert pdf2_count == 1, \
        f'Example 1st layer: expected pdf2 1x, got {pdf2_count}. Names: {names}'
    assert pdf3_count == 1, \
        f'Example 1st layer: expected pdf3 1x, got {pdf3_count}. Names: {names}'


def test_parse_emails_example_1st_layer_has_headers() -> None:
    """Test example 1st layer email body includes From/Subject headers."""
    email_bytes = _load_example_1st_layer()
    blobs = _run_parse_emails_real_libs(email_bytes)

    # Find the top-level email body (not nested under file paths)
    top_bodies = [b for b in blobs if b.name == '$email_body']
    assert len(top_bodies) >= 1, f'Expected at least 1 top-level email body, got {len(top_bodies)}'

    top_body = top_bodies[0].content.decode('utf-8')

    # Should contain from/subject/date headers
    has_from = 'From:' in top_body
    has_subject = 'Subject:' in top_body

    assert has_from or has_subject, \
        f'Example 1st layer body should have headers, got: {top_body[:400]}'


def test_parse_emails_example_2nd_layer_single_pdf() -> None:
    """Test example 2nd layer email (which only has pdf2+3 as nested).

    The 2nd layer (Fw 2nd layer email with a pdf 2) contains:
    - pdf2.pdf (direct attachment)
    - pdf3.pdf (nested in 3rd layer email)

    Each should appear exactly once.
    """
    email_bytes = _load_example_2nd_layer()
    blobs = _run_parse_emails_real_libs(email_bytes)
    names = [b.name for b in blobs]

    # 2nd layer should have pdf2 at top + nested pdf3
    pdf2_count = sum('pdf_2' in name for name in names)
    pdf3_count = sum('pdf_3' in name for name in names)

    assert pdf2_count == 1, \
        f'Example 2nd layer: expected pdf2 1x, got {pdf2_count}. Names: {names}'
    assert pdf3_count == 1, \
        f'Example 2nd layer: expected pdf3 1x, got {pdf3_count}. Names: {names}'


def test_parse_emails_example_3rd_layer_single_pdf() -> None:
    """Test example 3rd layer email (which only has pdf3 as attachment).

    The 3rd layer (3rd layer email with a pdf 3) contains only:
    - pdf3.pdf (direct attachment, no further nesting)
    """
    email_bytes = _load_example_3rd_layer()
    blobs = _run_parse_emails_real_libs(email_bytes)
    names = [b.name for b in blobs]

    # 3rd layer should only have pdf3
    pdf3_count = sum('pdf_3' in name for name in names)

    assert pdf3_count == 1, \
        f'Example 3rd layer: expected pdf3 1x, got {pdf3_count}. Names: {names}'


def _make_cyrillic_simple_eml() -> bytes:
    """Build an in-memory RFC2822 email with Cyrillic subject and body."""
    from email.message import EmailMessage
    msg = EmailMessage()
    msg['From'] = 'sender@example.com'
    msg['To'] = 'recipient@example.com'
    msg['Subject'] = 'Събджект на кирилица'
    msg['Date'] = 'Tue, 5 May 2026 09:41:59 +0000'
    msg.set_content('Боди на кирилица 1', charset='utf-8')
    return bytes(msg)


def _make_cyrillic_with_attachment_eml() -> bytes:
    """Build an in-memory multipart RFC2822 email with Cyrillic body and a PDF attachment."""
    from email import encoders
    from email.mime.base import MIMEBase
    from email.mime.multipart import MIMEMultipart
    from email.mime.text import MIMEText
    outer = MIMEMultipart('mixed')
    outer['From'] = 'sender@example.com'
    outer['To'] = 'recipient@example.com'
    outer['Subject'] = 'Събджект с прикачен файл'
    outer['Date'] = 'Tue, 5 May 2026 10:00:00 +0000'
    outer.attach(MIMEText('Тяло с кирилица и прикачен файл', 'plain', 'utf-8'))
    att = MIMEBase('application', 'pdf')
    att.set_payload(b'%PDF-1.4\nCYRILLIC_ATTACH_FIXTURE')
    encoders.encode_base64(att)
    att.add_header('Content-Disposition', 'attachment', filename='cyrillic_doc.pdf')
    outer.attach(att)
    return outer.as_bytes()


def _make_cyrillic_with_html_eml() -> bytes:
    """Build an in-memory multipart/alternative email with HTML + plain bodies, Cyrillic content."""
    from email.mime.multipart import MIMEMultipart
    from email.mime.text import MIMEText
    outer = MIMEMultipart('alternative')
    outer['From'] = 'sender@example.com'
    outer['To'] = 'recipient@example.com'
    outer['Subject'] = 'Събджект на кирилица'
    outer['Date'] = 'Tue, 5 May 2026 09:41:59 +0000'
    outer.attach(MIMEText('Боди на кирилица 1', 'plain', 'utf-8'))
    outer.attach(MIMEText(
        '<html><body><b>Боди</b> на <i>кирилица</i> 1</body></html>',
        'html', 'utf-8',
    ))
    return outer.as_bytes()


def _make_cyrillic_with_html_and_attachment_eml() -> bytes:
    """Multipart/mixed email: HTML+plain body AND a PDF attachment."""
    from email import encoders
    from email.mime.base import MIMEBase
    from email.mime.multipart import MIMEMultipart
    from email.mime.text import MIMEText
    outer = MIMEMultipart('mixed')
    outer['From'] = 'sender@example.com'
    outer['To'] = 'recipient@example.com'
    outer['Subject'] = 'Събджект с HTML и прикачен файл'
    outer['Date'] = 'Tue, 5 May 2026 10:00:00 +0000'
    alt = MIMEMultipart('alternative')
    alt.attach(MIMEText('Боди на кирилица (plain)', 'plain', 'utf-8'))
    alt.attach(MIMEText(
        '<html><body><b>Боди</b> на <i>кирилица</i> (html)</body></html>',
        'html', 'utf-8',
    ))
    outer.attach(alt)
    att = MIMEBase('application', 'pdf')
    att.set_payload(b'%PDF-1.4\nHTML_ATTACH_FIXTURE')
    encoders.encode_base64(att)
    att.add_header('Content-Disposition', 'attachment', filename='doc.pdf')
    outer.attach(att)
    return outer.as_bytes()


def _decode_email_body_blob(blob: StoreBlobRequest) -> tuple[dict[str, str], str]:
    """
    Parse a $email_body blob as RFC2822 and return (headers_dict, plain_body_text).
    """
    import email as _email
    from email.header import decode_header as _decode_header

    msg = _email.message_from_bytes(blob.content)

    def _decode_hdr(raw: str | None) -> str:
        if not raw:
            return ''
        parts = []
        for chunk, charset in _decode_header(raw):
            if isinstance(chunk, bytes):
                parts.append(chunk.decode(charset or 'utf-8'))
            else:
                parts.append(chunk)
        return ''.join(parts)

    headers = {
        'From': _decode_hdr(msg.get('From')),
        'To': _decode_hdr(msg.get('To')),
        'Subject': _decode_hdr(msg.get('Subject')),
        'Date': _decode_hdr(msg.get('Date')),
        'Content-Type': msg.get('Content-Type', ''),
    }

    body = ''
    if msg.is_multipart():
        for part in msg.walk():
            if part.get_content_type() == 'text/plain':
                charset = part.get_content_charset() or 'utf-8'
                raw_payload = part.get_payload(decode=True)
                if isinstance(raw_payload, bytes):
                    body = raw_payload.decode(charset)
                break
    else:
        charset = msg.get_content_charset() or 'utf-8'
        payload = msg.get_payload(decode=True)
        if payload is not None and isinstance(payload, bytes):
            body = payload.decode(charset)

    return headers, body


# ============================================================================
# Encoding & formatting integrity tests (inline fixtures, no external files)
# ============================================================================

def test_email_body_blob_is_valid_rfc2822() -> None:
    """$email_body blob must be parseable RFC2822."""
    import email as _email
    raw = _make_cyrillic_simple_eml()
    blobs = _run_parse_emails_real_libs(raw)
    body_blobs = [b for b in blobs if b.name == '$email_body']
    assert len(body_blobs) == 1, 'Expected exactly one $email_body blob'
    msg = _email.message_from_bytes(body_blobs[0].content)
    assert msg.get_content_maintype() in ('text', 'multipart'), \
        f'Unexpected top-level Content-Type: {msg.get_content_type()}'


def test_email_body_blob_cyrillic_subject_decoded_correctly() -> None:
    """Subject containing Cyrillic must survive the round-trip without corruption."""
    raw = _make_cyrillic_simple_eml()
    blobs = _run_parse_emails_real_libs(raw)
    body_blobs = [b for b in blobs if b.name == '$email_body']
    assert body_blobs, 'Expected $email_body blob'
    headers, _ = _decode_email_body_blob(body_blobs[0])
    assert 'Събджект' in headers['Subject'] and 'кирилица' in headers['Subject'], \
        f"Cyrillic subject corrupted: {headers['Subject']!r}"


def test_email_body_blob_cyrillic_body_decoded_correctly() -> None:
    """Body text containing Cyrillic must survive the round-trip without corruption."""
    raw = _make_cyrillic_simple_eml()
    blobs = _run_parse_emails_real_libs(raw)
    body_blobs = [b for b in blobs if b.name == '$email_body']
    assert body_blobs, 'Expected $email_body blob'
    _, body = _decode_email_body_blob(body_blobs[0])
    assert 'Боди' in body and 'кирилица' in body, \
        f'Cyrillic body text corrupted: {body!r}'


def test_email_body_blob_cyrillic_addresses_are_valid_rfc5322() -> None:
    """From/To addresses must be valid RFC5322, not Python tuple repr."""
    raw = _make_cyrillic_simple_eml()
    blobs = _run_parse_emails_real_libs(raw)
    body_blobs = [b for b in blobs if b.name == '$email_body']
    assert body_blobs, 'Expected $email_body blob'
    headers, _ = _decode_email_body_blob(body_blobs[0])
    for field in ('From', 'To'):
        value = headers[field]
        assert '(' not in value and "'" not in value, \
            f'{field} looks like a Python tuple repr: {value!r}'
    assert 'sender@example.com' in headers['From'], \
        f"From address not found: {headers['From']!r}"


def test_email_body_blob_html_body_preserved() -> None:
    """HTML body part must be preserved in the $email_body blob unchanged.

    The original approach reconstructed a plain-text-only email, discarding the HTML
    part entirely. This caused bold/italic formatting and proper line breaks to disappear
    in downstream rendering. With the strip-attachments approach the HTML part is kept.
    """
    import email as _email
    raw = _make_cyrillic_with_html_eml()
    blobs = _run_parse_emails_real_libs(raw)
    body_blobs = [b for b in blobs if b.name == '$email_body']
    assert body_blobs, 'Expected $email_body blob'
    msg = _email.message_from_bytes(body_blobs[0].content)
    html_parts = [p for p in msg.walk() if p.get_content_type() == 'text/html']
    assert html_parts, 'HTML part must be preserved in $email_body blob'
    html_raw = html_parts[0].get_payload(decode=True)
    html_content = html_raw.decode(
        html_parts[0].get_content_charset() or 'utf-8'
    ) if isinstance(html_raw, bytes) else ''
    assert '<b>' in html_content or '<strong>' in html_content, \
        f'Bold HTML tag not found in preserved HTML body: {html_content!r}'
    assert 'Боди' in html_content and 'кирилица' in html_content, \
        f'Cyrillic missing from HTML body: {html_content!r}'


def test_email_body_blob_attachment_stripped_html_preserved() -> None:
    """Attachment must be stripped but HTML body must remain in the $email_body blob."""
    import email as _email
    raw = _make_cyrillic_with_html_and_attachment_eml()
    blobs = _run_parse_emails_real_libs(raw)
    body_blobs = [b for b in blobs if b.name == '$email_body']
    attachment_blobs = [b for b in blobs if b.name != '$email_body']
    assert len(body_blobs) == 1, f'Expected 1 $email_body, got {len(body_blobs)}'
    assert len(attachment_blobs) >= 1, 'Expected at least 1 attachment blob'
    # HTML must be preserved
    msg = _email.message_from_bytes(body_blobs[0].content)
    html_parts = [p for p in msg.walk() if p.get_content_type() == 'text/html']
    assert html_parts, 'HTML part must survive attachment stripping'
    # Attachment payload must NOT be in body blob
    assert b'HTML_ATTACH_FIXTURE' not in body_blobs[0].content, \
        'Attachment payload leaked into the body blob'
    # No Content-Disposition: attachment part must remain
    att_parts = [
        p for p in msg.walk()
        if p.get_content_disposition() == 'attachment'
    ]
    assert not att_parts, (
        f'Attachment MIME parts found in body blob: {[p.get_filename() for p in att_parts]}'
    )


def test_email_body_blob_with_attachment_body_only() -> None:
    """Plain-text body email with attachment: body preserved, attachment stripped."""
    raw = _make_cyrillic_with_attachment_eml()
    blobs = _run_parse_emails_real_libs(raw)
    body_blobs = [b for b in blobs if b.name == '$email_body']
    attachment_blobs = [b for b in blobs if b.name != '$email_body']
    assert len(body_blobs) == 1, f'Expected 1 $email_body, got {len(body_blobs)}'
    assert len(attachment_blobs) >= 1, 'Expected at least 1 attachment blob'
    _, body = _decode_email_body_blob(body_blobs[0])
    assert 'Тяло' in body and 'кирилица' in body, \
        f'Cyrillic body not found in body blob: {body!r}'
    assert 'CYRILLIC_ATTACH_FIXTURE' not in body, \
        'Attachment payload leaked into the body blob'



def _make_multiline_plain_eml() -> bytes:
    """Build a plain-text email whose body contains multiple distinct lines.
    Used to verify that line breaks are preserved in the $email_body blob.
    """
    from email.mime.text import MIMEText
    msg = MIMEText(
        'Line one\nРед две на кирилица\nLine three\nЧетвърти ред на кирилица',
        'plain',
        'utf-8',
    )
    msg['From'] = 'sender@example.com'
    msg['To'] = 'recipient@example.com'
    msg['Subject'] = 'Multiline test'
    msg['Date'] = 'Tue, 5 May 2026 09:41:59 +0000'
    return msg.as_bytes()


def test_email_body_blob_plain_text_line_breaks_preserved() -> None:
    """Line breaks in the plain-text body must be preserved in the $email_body blob.

    The 'no_html' regression showed all lines merged into one paragraph.  With the
    strip-attachments approach the original body bytes are kept verbatim, so newlines
    must survive the round-trip.
    """
    raw = _make_multiline_plain_eml()
    blobs = _run_parse_emails_real_libs(raw)
    body_blobs = [b for b in blobs if b.name == '$email_body']
    assert body_blobs, 'Expected $email_body blob'
    _, body = _decode_email_body_blob(body_blobs[0])
    lines = [line for line in body.splitlines() if line.strip()]
    assert len(lines) >= 4, (  # noqa: PLR2004
        f'Expected >=4 non-empty lines, got {len(lines)}. Body: {body!r}'
    )
    assert 'Line one' in lines[0], f'First line wrong: {lines[0]!r}'
    assert 'Ред две' in lines[1] and 'кирилица' in lines[1], \
        f'Second Cyrillic line corrupted: {lines[1]!r}'
    assert 'Line three' in lines[2], f'Third line wrong: {lines[2]!r}'
    assert 'Четвърти' in lines[3], f'Fourth Cyrillic line corrupted: {lines[3]!r}'


def test_email_body_blob_preserves_original_charset_and_cte() -> None:
    """The body blob must preserve the original email's charset and Content-Transfer-Encoding.

    The strip-attachments approach should never re-encode or transcode the body.
    Whatever charset/CTE was set in the original email must appear unchanged in the blob,
    so that downstream renderers receive exactly the same encoding signal.
    """
    import email as _email
    raw = _make_cyrillic_with_html_eml()
    # Parse original to capture its charset and CTE
    orig_msg = _email.message_from_bytes(raw)
    # Find the text/html part's CTE in the original
    orig_html_cte = None
    for part in orig_msg.walk():
        if part.get_content_type() == 'text/html':
            orig_html_cte = (part.get('Content-Transfer-Encoding') or '').lower()
            break

    blobs = _run_parse_emails_real_libs(raw)
    body_blobs = [b for b in blobs if b.name == '$email_body']
    assert body_blobs, 'Expected $email_body blob'
    blob_msg = _email.message_from_bytes(body_blobs[0].content)

    # The HTML part's CTE in the body blob must match what was in the original
    blob_html_cte = None
    for part in blob_msg.walk():
        if part.get_content_type() == 'text/html':
            blob_html_cte = (part.get('Content-Transfer-Encoding') or '').lower()
            break

    assert blob_html_cte is not None, 'HTML part missing from body blob'
    assert blob_html_cte == orig_html_cte, (
        f'HTML part CTE changed: original={orig_html_cte!r}, blob={blob_html_cte!r}. '
        'Body blob must not re-encode the original email.'
    )


def test_email_body_blob_example_emails_have_html() -> None:
    """Emails must have their HTML body preserved.
    """
    import email as _email
    for label, loader in [
        ('1st layer', _load_example_1st_layer),
        ('2nd layer', _load_example_2nd_layer),
        ('3rd layer', _load_example_3rd_layer),
    ]:
        blobs = _run_parse_emails_real_libs(loader())
        body_blobs = [b for b in blobs if b.name == '$email_body']
        assert body_blobs, f'{label}: expected at least one $email_body blob'
        top_body_msg = _email.message_from_bytes(body_blobs[0].content)
        html_parts = [p for p in top_body_msg.walk() if p.get_content_type() == 'text/html']
        assert html_parts, (
            f'{label}: $email_body blob has no text/html part. '
            'Outlook emails always include HTML; stripping must preserve it.'
        )


def _make_email_with_inline_signature_image_eml() -> bytes:
    """
    Build a multipart/related email that has:
    - text/plain body
    - text/html body referencing an inline image via cid:
    - image/png with Content-Disposition: inline; filename="logo.png"  (the signature)
    This mimics a real Outlook email with a company signature.
    """
    from email.mime.image import MIMEImage
    from email.mime.multipart import MIMEMultipart
    from email.mime.text import MIMEText

    outer = MIMEMultipart('related')
    outer['From'] = 'sender@example.com'
    outer['To'] = 'recipient@example.com'
    outer['Subject'] = 'Email with inline signature image'
    outer['Date'] = 'Tue, 5 May 2026 09:41:59 +0000'

    alt = MIMEMultipart('alternative')
    alt.attach(MIMEText('Body text plain', 'plain', 'utf-8'))
    alt.attach(MIMEText(
        '<html><body>Body text <b>html</b><img src="cid:sig_img"></body></html>',
        'html', 'utf-8',
    ))
    outer.attach(alt)

    # Inline signature image (same bytes in all levels — this is what was duplicating)
    img_bytes = b'\x89PNG_SIGNATURE_FIXTURE_BYTES'
    sig_img = MIMEImage(img_bytes, 'png')
    sig_img.add_header('Content-Disposition', 'inline', filename='logo.png')
    sig_img.add_header('Content-ID', '<sig_img>')
    outer.attach(sig_img)

    return outer.as_bytes()


def test_email_body_blob_inline_images_stripped() -> None:
    """Inline images with filenames must NOT appear inside the $email_body blob.
    """
    import email as _email
    raw = _make_email_with_inline_signature_image_eml()
    blobs = _run_parse_emails_real_libs(raw)
    body_blobs = [b for b in blobs if b.name == '$email_body']
    assert body_blobs, 'Expected $email_body blob'
    msg = _email.message_from_bytes(body_blobs[0].content)
    image_parts = [p for p in msg.walk() if p.get_content_maintype() == 'image']
    assert not image_parts, (
        f'Body blob must not contain image parts, found: '
        f'{[(p.get_filename(), p.get_content_type()) for p in image_parts]}'
    )


def test_email_body_blob_inline_image_emitted_as_standalone_blob() -> None:
    """The inline signature image must appear exactly once as a standalone attachment blob."""
    raw = _make_email_with_inline_signature_image_eml()
    blobs = _run_parse_emails_real_libs(raw)
    non_body_blobs = [b for b in blobs if b.name != '$email_body']
    img_blobs = [b for b in non_body_blobs if b'PNG_SIGNATURE_FIXTURE_BYTES' in b.content]
    assert len(img_blobs) == 1, (
        f'Inline signature image must appear exactly once as a standalone blob, '
        f'got {len(img_blobs)}. All blob names: {[b.name for b in blobs]}'
    )


def test_email_body_blob_example_inline_png_not_in_body_blobs() -> None:
    """In the example 1st-layer email, the signature PNG must not appear in body blobs.
    """
    import email as _email
    blobs = _run_parse_emails_real_libs(_load_example_1st_layer())
    body_blobs = [b for b in blobs if b.name == '$email_body']
    for bb in body_blobs:
        msg = _email.message_from_bytes(bb.content)
        image_parts_with_filename = [
            p for p in msg.walk()
            if p.get_content_maintype() == 'image' and p.get_filename()
        ]
        assert not image_parts_with_filename, (
            f"Body blob '{bb.name}' must not contain inline image parts with filenames, "
            f"found: {[(p.get_filename(), p.get_content_type()) for p in image_parts_with_filename]}"  # noqa: E501
        )


if __name__ == '__main__':
    export_flow(entry_point_workflow())

How do I write extracted data to Salesforce with my own logic?

Available in v44 and later. The updated Salesforce Notifier output block runs a flow of yours for every payload of a completed submission, passing down the Salesforce credentials, the submission id and a reference to the rendered payload. Your flow reads the payload and decides what to write, so several objects, a related record, an upsert or a value computed from two fields are all ordinary Python.

The sample is that flow: it creates a Salesforce Account object from an extracted Form W-9, looking the record up first and creating it only when nothing matches. It ships with dry run on, so it queries Salesforce without writing to it. The output block itself is added to a document processing flow you already have, in Flow Studio, so nothing in the sample is tied to an IDP version.

You can download the sample, and see Salesforce Integration Flow for the walkthrough and the expected results. On platforms before v44 the output block does not exist; the same page’s “Before v44” section ships a second sample that wires a Salesforce writer subflow into a document processing flow by hand, and can also write submission metadata.

Salesforce Integration Flow

"""Integration flow driven by the Salesforce Notifier output block.

The Salesforce Notifier output block does not write to Salesforce itself. For every payload of a
completed submission it runs the flow you select in its "Integration Flow" picker, passing down the
Salesforce credentials, the submission id and a reference to the rendered payload. Everything that
happens in Salesforce is therefore ordinary Python you own and can change: this sample looks up a
Salesforce record by the fields it is about to write and creates it only when nothing matches.

To be selectable in that picker a flow needs two things, both below:

* the ``salesforce_integration`` role, and
* the inputs the output block passes in - ``submission_id``, ``payload_uuid``, ``username``,
  ``consumer_key``, ``private_key`` and ``sandbox_environment``.

The credentials are entered once, on the output block, and forwarded from there, so they are never
stored in this file.

Everything else is configured here, through the manifest defaults: which Salesforce object to
write, how document fields and submission metadata map onto its fields, and whether to run in dry
run mode. A supporting flow has no settings panel of its own, so a manifest default is the only
place these values can come from.

Prerequisites:

* ``simple-salesforce`` and its dependencies uploaded under **Flows -> Python Packages**, for the
  same Python version this flow is built for. It is not a single wheel, and packages are tracked
  per Python version.
* A Salesforce connected app set up for JWT bearer authentication.

Nothing here is tied to an IDP version, because the flow never touches IDP: it takes a payload
reference and writes to Salesforce. Import it as it is, then add the output block to whichever
document processing flow you already run::

    HS_PYTHON_CODE_BLOCK_PYTHON_VERSION=3.13 hs-flows bundle salesforce_writer_flow.py \
        -o salesforce_writer_flow.zip

See `Salesforce Integration Flow <../pages/salesforce_integration_flow.html>`_ for the full walkthrough.
"""

from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsBlockInstance, HsTask
from flows_sdk.utils import workflow_input

IDENTIFIER = 'SALESFORCE_INTEGRATION_FLOW_EXAMPLE'
FLOW_UUID = UUID('9c1d4b7e-2f83-4a56-b0d1-7e5c8a3f42b6')

# Must match the output block's role filter, which is what narrows its "Integration Flow" picker.
SALESFORCE_INTEGRATION_ROLE = 'salesforce_integration'


class Inputs:
    # Passed in by the Salesforce Notifier output block (hidden, never configured here).
    SubmissionId = 'submission_id'
    PayloadUuid = 'payload_uuid'
    Username = 'username'
    ConsumerKey = 'consumer_key'
    PrivateKey = 'private_key'
    SandboxEnvironment = 'sandbox_environment'
    # Configured on this flow, through the manifest defaults below.
    SfObject = 'sf_object'
    MetadataFieldsMapping = 'metadata_fields_mapping'
    DocumentFieldsMapping = 'document_fields_mapping'
    DryRun = 'dry_run'


def _write_payload_to_salesforce(
    submission_id: int,
    payload_uuid: str,
    username: str,
    consumer_key: str,
    private_key: str,
    sandbox_environment: bool,
    sf_object: str,
    metadata_fields_mapping: dict,
    document_fields_mapping: dict,
    dry_run: bool,
    _hs_block_instance: HsBlockInstance,
    _hs_task: HsTask,
) -> dict:
    import json

    # `simple-salesforce` has to be installed for the platform's code-block interpreter, see the
    # module docstring. The import lives inside the function because it runs on the block, not
    # when the flow is built.
    from simple_salesforce import (  # type: ignore[import-not-found]
        Salesforce,  # pants: no-infer-dep
    )

    # A Python code block ships only THIS function's source, so every helper has to be defined
    # inside it. A module-level helper would raise NameError at runtime.

    def log_info(msg: str, *args: object) -> None:
        # `log()` applies {}-style formatting to the message, so values that may themselves
        # contain braces (a dict, a SOQL string) are passed as ARGS rather than interpolated
        # into `msg`: args are substituted, not re-parsed.
        _hs_block_instance.log(msg, *args, level=HsBlockInstance.LogLevel.INFO)

    def soql_literal(value: object) -> str:
        # SOQL string literals are single-quoted; escape backslashes and single quotes.
        escaped = str(value).replace('\\', '\\\\').replace("'", "\\'")
        return f"'{escaped}'"

    # The output block hands over a blob reference, one per rendered payload. The payload is the
    # same JSON the platform renders for its output blocks: a top-level `metadata` (submission
    # metadata) plus `output` (the serialized submission). With the "Full individual object" export
    # type each payload covers one document, with `output.documents` narrowed to it.
    blob = _hs_block_instance.fetch_blob(payload_uuid)
    payload = json.loads(blob.content.decode('utf-8'))
    metadata = payload.get('metadata') or {}
    documents = (payload.get('output') or {}).get('documents') or []

    def get_extracted_field(field_name: str) -> object:
        # A field in a v5 payload looks like:
        #   {"name": "Name", "output_name": "name", "id": 1, "state": "complete",
        #    "transcription": {"raw": "Jane Doe", "normalized": "Jane Doe",
        #                      "source": "machine_transcription", ...}, ...}
        # so a field is matched on either the layout field name or its output name, and
        # `transcription` is an object rather than a string. A plain string transcription is
        # accepted too, since that is what the flat export types send.
        for document in documents:
            for field in document.get('document_fields') or []:
                names = (field.get('name'), field.get('output_name'), field.get('field_name'))
                if field_name not in names:
                    continue
                transcription = field.get('transcription')
                if isinstance(transcription, dict):
                    # Prefer the normalized value, which is the platform's canonical one:
                    # "12-0719651" normalizes to "120719651" and text is upper-cased. Use
                    # `raw` instead where the verbatim string matters.
                    return transcription.get('normalized') or transcription.get('raw')
                return field.get('transcription_normalized') or transcription
        return None

    # Build the Salesforce record from the two mappings, source name -> Salesforce field API name.
    record: dict = {}

    for metadata_key, sf_field in (metadata_fields_mapping or {}).items():
        if metadata_key not in metadata:
            raise ValueError(
                f'Submission metadata has no key {metadata_key!r}. '
                f'Available keys: {sorted(metadata)}'
            )
        record[sf_field] = metadata[metadata_key]

    for document_field_name, sf_field in (document_fields_mapping or {}).items():
        value = get_extracted_field(document_field_name)
        if value is None:
            raise ValueError(
                f'No extracted value for document field {document_field_name!r} in payload '
                f'{payload_uuid} of submission {submission_id}.'
            )
        record[sf_field] = value

    if not sf_object:
        raise ValueError('sf_object is required (e.g. "Account").')
    if not record:
        raise ValueError(
            'Nothing to write: both metadata_fields_mapping and document_fields_mapping are empty.'
        )

    # JWT bearer authentication with the connected app credentials the output block passes in:
    # username is the JWT subject, consumer_key the issuer, private_key the RS256 signing key.
    # A PEM pasted into a form field can arrive with literal "\n", so normalize it first.
    private_key = private_key.replace('\\n', '\n')
    domain = 'test' if sandbox_environment else 'login'
    salesforce = Salesforce(
        username=username,
        consumer_key=consumer_key,
        privatekey=private_key,
        domain=domain,
    )

    # Look the record up by the very fields that are about to be written, AND'ed together, and
    # create it only when nothing matches. That keeps a re-run from duplicating records.
    where = ' AND '.join(f'{sf_field} = {soql_literal(v)}' for sf_field, v in record.items())
    query = f'SELECT Id FROM {sf_object} WHERE {where}'
    # Log the query before running it, so it is visible even if the query itself errors.
    log_info('Salesforce lookup query: {}', query)
    result = salesforce.query_all(query)
    count = result.get('totalSize', len(result.get('records', [])))
    log_info('Salesforce lookup on {} with {} -> {} record(s).', sf_object, record, count)

    created_id = None
    if dry_run:
        log_info('Dry run: not creating a {} record. Would have written {}.', sf_object, record)
    elif count == 0:
        created = getattr(salesforce, sf_object).create(record)
        created_id = created.get('id')
        log_info('No existing {}; created record {} with {}.', sf_object, created_id, record)
    else:
        log_info('{} existing {} record(s) matched; skipping create.', count, sf_object)

    return {
        'submission_id': submission_id,
        'payload_uuid': payload_uuid,
        'result_count': count,
        'lookup_fields': record,
        'query': query,
        'dry_run': dry_run,
        'created': created_id is not None,
        'salesforce_id': created_id,
    }


def workflow() -> Flow:
    write_to_salesforce = PythonBlock(
        reference_name='write_payload_to_salesforce',
        code=_write_payload_to_salesforce,
        # No `python_version` here: the block compiles to the Python version of the venv this
        # is built in, or to `HS_PYTHON_CODE_BLOCK_PYTHON_VERSION` if that is set. Pick the one
        # your platform runs, as described in the walkthrough.
        code_input={
            'submission_id': workflow_input(Inputs.SubmissionId),
            'payload_uuid': workflow_input(Inputs.PayloadUuid),
            'username': workflow_input(Inputs.Username),
            'consumer_key': workflow_input(Inputs.ConsumerKey),
            'private_key': workflow_input(Inputs.PrivateKey),
            'sandbox_environment': workflow_input(Inputs.SandboxEnvironment),
            'sf_object': workflow_input(Inputs.SfObject),
            'metadata_fields_mapping': workflow_input(Inputs.MetadataFieldsMapping),
            'document_fields_mapping': workflow_input(Inputs.DocumentFieldsMapping),
            'dry_run': workflow_input(Inputs.DryRun),
        },
        title='Write payload to Salesforce',
        description='Fetches the rendered submission payload, maps its metadata and extracted '
        'document fields to Salesforce field API names, and writes them to a Salesforce record '
        'using simple-salesforce.',
    )

    return Flow(
        uuid=FLOW_UUID,
        owner_email='flows-sdk@hyperscience.com',
        title='Salesforce Integration Flow - flows-sdk example',
        description='Writes one rendered submission payload to Salesforce. Driven by the '
        'Salesforce Notifier output block, which passes in the credentials, the submission id '
        'and the payload reference.',
        manifest=Manifest(
            identifier=IDENTIFIER,
            # 'supporting' marks this as a subflow, so the validator does not require values for
            # the inputs the caller supplies. The salesforce role is what makes the output
            # block's "Integration Flow" picker offer this flow.
            roles=['supporting', SALESFORCE_INTEGRATION_ROLE],
            input=[
                # Passed in by the output block, hidden because they are not configured here.
                Parameter(
                    name=Inputs.SubmissionId,
                    type='integer',
                    title='Submission ID',
                    ui={'hidden': True},
                ),
                Parameter(
                    name=Inputs.PayloadUuid,
                    type='string',
                    title='Payload UUID',
                    ui={'hidden': True},
                ),
                Parameter(
                    name=Inputs.Username,
                    type='string',
                    title='Associated Username',
                    ui={'hidden': True},
                ),
                Parameter(
                    name=Inputs.ConsumerKey,
                    type='Password',
                    title='Consumer Key',
                    secret=True,
                    ui={'hidden': True},
                ),
                Parameter(
                    name=Inputs.PrivateKey,
                    type='Password',
                    title='Private Key',
                    secret=True,
                    ui={'hidden': True},
                ),
                Parameter(
                    name=Inputs.SandboxEnvironment,
                    type='boolean',
                    title='Sandbox Environment',
                    value=False,
                    ui={'hidden': True},
                ),
                # Configured here, through these defaults. The output block passes a fixed input
                # set, and a supporting flow has no settings panel, so a default is the only way
                # to set them.
                Parameter(
                    name=Inputs.SfObject,
                    type='string',
                    title='Salesforce Object',
                    description='API name of the Salesforce object to look up and create, '
                    'e.g. Account.',
                    value='Account',
                ),
                Parameter(
                    name=Inputs.MetadataFieldsMapping,
                    type='json',
                    title='Metadata Field Mappings',
                    description='Map submission metadata keys to Salesforce field API names, e.g. '
                    '{"customer_id": "AccountNumber"}. Submission metadata is empty for documents '
                    'uploaded through the UI.',
                    value={},
                    optional=True,
                ),
                Parameter(
                    name=Inputs.DocumentFieldsMapping,
                    type='json',
                    title='Document Field Mappings',
                    description='Map extracted document field names to Salesforce field API '
                    'names, e.g. {"Name": "Name"}. A mapped field with no extracted value fails '
                    'the flow rather than writing a partial record.',
                    # "Name" comes from the W-9 layout shipped with the Document Processing
                    # Starter and maps onto a standard Account field, so nothing has to be created
                    # in Salesforce first. It is printed on the sample document, so machine
                    # transcription reads it without a human keyer - unlike its "Address", which
                    # is handwritten and comes back illegible.
                    value={'Name': 'Name'},
                ),
                Parameter(
                    name=Inputs.DryRun,
                    type='boolean',
                    title='Dry Run',
                    description='Run the Salesforce lookup and log what would be written, but do '
                    'not create the record. Turn this off to write for real.',
                    value=True,
                ),
            ],
        ),
        blocks=[write_to_salesforce],
        output={'result': write_to_salesforce.output()},
    )


def entry_point_workflow() -> Flow:
    return workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

Unsupported Versions

Warning

The following Hyperscience versions (V39.1 and below) are deprecated and no longer actively supported. Please upgrade to V40 or later for the latest features and bug fixes.

V39.1

Name	Description	Release	Sample Docs	Download
IDP Starter V39.1	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v39_1.zip
OnError flow	Example implementation of an OnError flow that is triggered once the original flow fails.			on_error_flow_v39_1.zip

IDP Starter V39.1

The basic document processing flow. From V38 on, the IDP Document processing is split into a wrapper flow (IDP) and a subflow (IDP Core). The package includes:

• IDP Core Flow - a read-only subflow containing the steps to process documents into machine-readable output

• Document Processing Flow - a wrapper around the IDP Core flow, used for managing input and output connector configuration and document processing workflow settings

• IDP Submission Notify Flow - A flow that can be configured for some blocks of the IDP Core Flow, used to send notifications to external systems when a submission has been created or is waiting for supervision

• Submission-aware On Error Flow - When the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered. The submission-aware on-error flow enriches the error message with submission data (compared with the plain OnError flow shown below). In case this OnError implementation is used on a non-IDP flow (i.e. a submissionless flow), submission data will be omitted

The package includes also release and sample documents for processing Form W-9s.

Download idp_flow_v39_1.zip

IDP V39.1 Sample

from uuid import UUID

from flows_sdk.error_handling import ErrorHandling, OnErrorFlow
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v39_1 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v39_1.idp_blocks import IDPCoreBlock
from flows_sdk.implementations.idp_v39_1.idp_values import (
    DEFAULT_IDP_CORE_IDENTIFIER,
    DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER,
    DEFAULT_SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER,
    IdpWorkflowConfig,
    IDPWrapperManifest,
    Settings,
    get_idp_wf_config,
)
from flows_sdk.package_utils import export_flow

IDP_UUID = '71d32ae8-65be-4518-932a-e865be95186e'
IDP_IDENTIFIER = 'DOCUMENT_PROCESSING_EXAMPLE'
IDP_TITLE = 'Document Processing - flows-sdk example'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:
    # By default, all the notification flows are set to `None`.
    # I.e. the system will send no intermediate notifications for submission change events.
    # So let's specify the notification flows that we want to use explicitly here:
    idp_wf_config.notification_workflow_initialization = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_classification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_identification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_transcription = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_flexible_extraction = (
        DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    )

    idp_flow = IDPCoreBlock(idp_wf_config, identifier=DEFAULT_IDP_CORE_IDENTIFIER)

    outputs = idp_blocks.IDPOutputsBlock(inputs={'submission': idp_flow.output('submission')})

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_TITLE,
        description='Use this flow to manage your input and output connector configuration and '
        'document processing workflow settings.',
        manifest=IDPWrapperManifest(IDP_IDENTIFIER),
        output={'submission': idp_flow.output('submission')},
        triggers=idp_values.IDPTriggers(),
        input={
            Settings.LayoutReleaseUuid: None,
        },
        blocks=[
            idp_flow,
            outputs,
        ],
        error_handling=ErrorHandling(
            on_error_flow=OnErrorFlow(identifier=DEFAULT_SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER)
        ),
    )


def entry_point_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Core V39.1 Sample

from uuid import UUID

from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v39_1 import idp_blocks
from flows_sdk.implementations.idp_v39_1.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED
from flows_sdk.implementations.idp_v39_1.idp_values import IDPCoreManifest
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import str_to_deterministic_uuid_4

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into IDPCoreBlock(identifier=...) within your flow.
IDP_CORE_IDENTIFIER = f'IDP_CORE_{FLOWS_VERSION}_EXAMPLE'
IDP_CORE_UUID = str(str_to_deterministic_uuid_4(IDP_CORE_IDENTIFIER))
IDP_CORE_TITLE = f'Document Processing Subflow {FLOWS_VERSION_DOTTED} - flows-sdk example'


def idp_workflow() -> Flow:
    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction.output('submission'),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    return Flow(
        uuid=UUID(IDP_CORE_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_CORE_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output.',
        manifest=IDPCoreManifest(
            flow_identifier=IDP_CORE_IDENTIFIER, roles=['read_only', 'supporting', 'idp_core']
        ),
        output={'submission': submission_complete.output('submission')},
        blocks=[
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            flexible_extraction,
            reprocessing,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Submission Notify V39.1 Sample

import os
from uuid import UUID

from flows_sdk.blocks import Outputs
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v39_1.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import icon_path_to_base64, str_to_deterministic_uuid_4, workflow_input

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

IDP_SUBMISSION_NOTIFY_IDENTIFIER = f'IDP_SUBMISSION_NOTIFY_{FLOWS_VERSION}_EXAMPLE'
IDP_SUBMISSION_NOTIFY_UUID = str(str_to_deterministic_uuid_4(IDP_SUBMISSION_NOTIFY_IDENTIFIER))
IDP_SUBMISSION_NOTIFY_TITLE = (
    f'Submission State Notifications {FLOWS_VERSION_DOTTED} - flows-sdk example'
)
IDP_OUTPUT_ROLE = 'idp_output'


class SubmissionNotifyInputKeys:
    submission: str = 'submission'


class SubmissionNotifyManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['notifications', 'supporting'],
            input=[
                Parameter(
                    name=SubmissionNotifyInputKeys.submission,
                    type='Submission',
                    title='Submission Object',
                    ui={
                        'hidden': True,
                    },
                )
            ],
            ui={
                'icon': icon_path_to_base64(os.path.join(BASE_DIR, 'submission_notify_icon.png')),
            },
        )


def idp_submission_notify_workflow() -> Flow:
    return Flow(
        title=IDP_SUBMISSION_NOTIFY_TITLE,
        owner_email='flows.sdk@hyperscience.com',
        description=(
            'Send notifications to external systems when a submission has been '
            'created or is waiting for supervision. \n'
            f'If the "Document Processing" flow is live, this flow '
            'must also be live, but it can be empty.'
        ),
        manifest=SubmissionNotifyManifest(IDP_SUBMISSION_NOTIFY_IDENTIFIER),
        blocks=[
            Outputs(
                reference_name='outputs',
                title='Outputs',
                description=(
                    'Send submission data to external systems when a submission '
                    'has been created or is waiting for supervision'
                ),
                role_filter=[IDP_OUTPUT_ROLE],
                input_template={
                    'submission': {'id': workflow_input('submission.id')},
                    'enabled': True,
                },
                blocks=[],
            )
        ],
        uuid=UUID(IDP_SUBMISSION_NOTIFY_UUID),
        input={},
        output={},
    )


def entry_point_workflow() -> Flow:
    return idp_submission_notify_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

Submission-aware On Error Flow V39.1 Sample

from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v39_1.additional_blocks import (
    HyperscienceRestApiBlock,
    JSONOutputsBlock,
)
from flows_sdk.implementations.idp_v39_1.flows_version import FLOWS_VERSION, FLOWS_VERSION_DOTTED
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsTask
from flows_sdk.utils import str_to_deterministic_uuid_4, workflow_input

SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER = f'SUBMISSION_AWARE_ON_ERROR_FLOW_{FLOWS_VERSION}'
SUBMISSION_AWARE_ON_ERROR_FLOW_UUID = str(
    str_to_deterministic_uuid_4(SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER)
)
SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE = (
    f'On-Error with included Submission data {FLOWS_VERSION_DOTTED}'
)
SUBMISSION_AWARE_ON_ERROR_DESCRIPTION = """Flow designed to be used as an "on-error" handler for
 submission-based flows.\n
With the on-error flow configured, when the original flow fails for any reason, and all
    the configured auto-retries have been exhausted, the on-error flow will be triggered.
\n
In this on-error flow implementation,
    we fetch the full metadata of the failed flow run, alongside with the full submission metadata
    (if the submission ID was found), and then send it to a user-configured output.
    The data format of the output conforms to
    the schema documented here: https://docs.hyperscience.ai/#flows-runs and
    here: https://docs.hyperscience.ai/#submissions, under the respective keys
    "flow_run" and "submission".
\n
If you need more complex logic in this flow
    please consult the documentation of Flows SDK (https://flows-sdk.hyperscience.ai)
    and the available public APIs (https://docs.hyperscience.ai)
"""


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'


class OnErrorManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['on_error', 'supporting'],
            input=[
                Parameter(
                    name=OnErrorFlowInputKeys.failed_run_uuid,
                    type='string',
                    title='Failed Flow Run UUID',
                    description='UUID of the failed flow run that triggered this on_error flow',
                    optional=False,
                    ui={'hidden': True},
                )
            ],
        )


def submission_aware_on_error_flow() -> Flow:
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        title='Get Flow Run',
        description='Get the full metadata of the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{workflow_input(OnErrorFlowInputKeys.failed_run_uuid)}',
    )

    def _get_correlation_id(_hs_task: HsTask) -> dict:
        return {'correlation_id': _hs_task.correlation_id}

    get_correlation_id_block = PythonBlock(
        reference_name='get_correlation_id',
        title='Get Correlation ID',
        description='Get the correlation ID for the failed Submission / Flow Run',
        code=_get_correlation_id,
    )

    fetch_submissions_block = HyperscienceRestApiBlock(
        reference_name='fetch_submission',
        title='Fetch Submission',
        description='Fetch the submission data for the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/submissions'
        '?debug=true'
        f'&correlation_id={get_correlation_id_block.output("correlation_id")}',
    )

    def _prep_output_data(flow_run: dict, get_submissions_response: dict) -> dict:
        submissions_in_response = get_submissions_response.get('results', [])
        submission = None if not submissions_in_response else submissions_in_response[0]
        return {'flow_run': flow_run, 'submission': submission}

    prep_output_data_block = PythonBlock(
        title='Format Submission Response',
        description='Prepare the output data to be sent to the user-configured output'
        'under the keys "flow_run" and "submission"',
        code=_prep_output_data,
        code_input={
            'flow_run': flow_run_rest_api_block.output('result.data'),
            'get_submissions_response': fetch_submissions_block.output('result.data'),
        },
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': prep_output_data_block.output()})

    return Flow(
        uuid=UUID(SUBMISSION_AWARE_ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE,
        description=SUBMISSION_AWARE_ON_ERROR_DESCRIPTION,
        manifest=OnErrorManifest(SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER),
        output={},
        input={},
        blocks=[
            flow_run_rest_api_block,
            get_correlation_id_block,
            fetch_submissions_block,
            prep_output_data_block,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return submission_aware_on_error_flow()


if __name__ == '__main__':
    export_flow(submission_aware_on_error_flow())

V39

Name	Description	Release	Sample Docs	Download
IDP Starter V39	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v39.zip
OnError flow	Example implementation of an OnError flow that is triggered once the original flow fails.			on_error_flow_v39.zip

IDP Starter V39

The basic document processing flow. From V38 on, the IDP Document processing is split into a wrapper flow (IDP) and a subflow (IDP Core). The package includes:

• IDP Core Flow - a read-only subflow containing the steps to process documents into machine-readable output

• Document Processing Flow - a wrapper around the IDP Core flow, used for managing input and output connector configuration and document processing workflow settings

• IDP Submission Notify Flow - A flow that can be configured for some blocks of the IDP Core Flow, used to send notifications to external systems when a submission has been created or is waiting for supervision

• Submission-aware On Error Flow - When the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered. The submission-aware on-error flow enriches the error message with submission data (compared with the plain OnError flow shown below). In case this OnError implementation is used on a non-IDP flow (i.e. a submissionless flow), submission data will be omitted

The package includes also release and sample documents for processing Form W-9s.

Download idp_flow_v39.zip

IDP V39 Sample

from uuid import UUID

from flows_sdk.error_handling import ErrorHandling, OnErrorFlow
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v39 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v39.idp_blocks import IDPCoreBlock
from flows_sdk.implementations.idp_v39.idp_values import (
    DEFAULT_IDP_CORE_IDENTIFIER,
    DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER,
    DEFAULT_SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER,
    IdpWorkflowConfig,
    IDPWrapperManifest,
    Settings,
    get_idp_wf_config,
)
from flows_sdk.package_utils import export_flow

IDP_UUID = '96a6606e-6c27-4423-a660-6e6c278423fb'
IDP_IDENTIFIER = 'DOCUMENT_PROCESSING_EXAMPLE'
IDP_TITLE = 'Document Processing - flows-sdk example'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:
    # By default, all the notification flows are set to `None`.
    # I.e. the system will send no intermediate notifications for submission change events.
    # So let's specify the notification flows that we want to use explicitly here:
    idp_wf_config.notification_workflow_initialization = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_classification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_identification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_transcription = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_flexible_extraction = (
        DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    )

    idp_flow = IDPCoreBlock(idp_wf_config, identifier=DEFAULT_IDP_CORE_IDENTIFIER)

    outputs = idp_blocks.IDPOutputsBlock(inputs={'submission': idp_flow.output('submission')})

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_TITLE,
        description='Use this flow to manage your input and output connector configuration and '
        'document processing workflow settings.',
        manifest=IDPWrapperManifest(IDP_IDENTIFIER),
        output={'submission': idp_flow.output('submission')},
        triggers=idp_values.IDPTriggers(),
        input={
            Settings.LayoutReleaseUuid: None,
        },
        blocks=[
            idp_flow,
            outputs,
        ],
        error_handling=ErrorHandling(
            on_error_flow=OnErrorFlow(identifier=DEFAULT_SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER)
        ),
    )


def entry_point_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Core V39 Sample

from uuid import UUID

from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v39 import idp_blocks
from flows_sdk.implementations.idp_v39.idp_values import IDPCoreManifest
from flows_sdk.package_utils import export_flow

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into IDPCoreBlock(identifier=...) within your flow.
IDP_CORE_UUID = '3c0e2602-c27d-45d5-8e26-02c27d05d58c'
IDP_CORE_IDENTIFIER = 'IDP_CORE_V39_EXAMPLE'
IDP_CORE_TITLE = 'Document Processing Subflow V39 - flows-sdk example'


def idp_workflow() -> Flow:
    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction.output('submission'),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    return Flow(
        uuid=UUID(IDP_CORE_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_CORE_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output.',
        manifest=IDPCoreManifest(
            flow_identifier=IDP_CORE_IDENTIFIER, roles=['read_only', 'supporting', 'idp_core']
        ),
        output={'submission': submission_complete.output('submission')},
        blocks=[
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            flexible_extraction,
            reprocessing,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Submission Notify V39 Sample

import os

from flows_sdk.blocks import Outputs
from flows_sdk.flows import Flow, Manifest
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import icon_path_to_base64, workflow_input

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

IDP_SUBMISSION_NOTIFY_IDENTIFIER = 'IDP_SUBMISSION_NOTIFY_V39_EXAMPLE'
IDP_SUBMISSION_NOTIFY_UUID = 'cf09ba7b-664d-48c5-89ba-7b664df8c518'
IDP_SUBMISSION_NOTIFY_TITLE = 'Submission State Notifications V39 - flows-sdk example'

IDP_OUTPUT_ROLE = 'idp_output'


def idp_submission_notify_workflow() -> Flow:
    return Flow(
        title=IDP_SUBMISSION_NOTIFY_TITLE,
        owner_email='flows.sdk@hyperscience.com',
        description=(
            'Send notifications to external systems when a submission has been '
            'created or is waiting for supervision. \n'
            f'If the "Document Processing" flow is live, this flow '
            'must also be live, but it can be empty.'
        ),
        is_user_facing=True,
        blocks=[
            Outputs(
                reference_name='outputs',
                title='Outputs',
                description=(
                    'Send submission data to external systems when a submission '
                    'has been created or is waiting for supervision'
                ),
                role_filter=[IDP_OUTPUT_ROLE],
                input_template={
                    'submission': {'id': workflow_input('submission.id')},
                    'enabled': True,
                },
                blocks=[],
            )
        ],
        uuid=IDP_SUBMISSION_NOTIFY_UUID,
        input={},
        output={},
        manifest=Manifest(
            identifier=IDP_SUBMISSION_NOTIFY_IDENTIFIER,
            input=[
                {
                    'name': 'submission',
                    'type': 'Submission',
                    'title': 'Submission Object',
                    'ui': {'hidden': True},
                }
            ],
            output=[],
            roles=['notifications', 'supporting'],
            ui={
                'hidden': True,
                'icon': icon_path_to_base64(os.path.join(BASE_DIR, 'notification_flow_icon.png')),
            },
        ),
    )


def entry_point_workflow() -> Flow:
    return idp_submission_notify_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

Submission-aware On Error Flow V39 Sample

from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v39.additional_blocks import (
    HyperscienceRestApiBlock,
    JSONOutputsBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsTask
from flows_sdk.utils import workflow_input

SUBMISSION_AWARE_ON_ERROR_FLOW_UUID = '909d73bc-c220-439d-9d73-bcc220f39d3d'
SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER = 'SUBMISSION_AWARE_ON_ERROR_FLOW_V39_EXAMPLE'
SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE = (
    'On-Error with included Submission data V39 - flows-sdk example'
)
SUBMISSION_AWARE_ON_ERROR_DESCRIPTION = """Flow designed to be used as an "on-error" handler for
 submission-based flows.\n
With the on-error flow configured, when the original flow fails for any reason, and all
    the configured auto-retries have been exhausted, the on-error flow will be triggered.
\n
In this on-error flow implementation,
    we fetch the full metadata of the failed flow run, alongside with the full submission metadata
    (if the submission ID was found), and then send it to a user-configured output.
    The data format of the output conforms to
    the schema documented here: https://docs.hyperscience.ai/#flows-runs and
    here: https://docs.hyperscience.ai/#submissions, under the respective keys
    "flow_run" and "submission".
\n
If you need more complex logic in this flow
    please consult the documentation of Flows SDK (https://flows-sdk.hyperscience.ai)
    and the available public APIs (https://docs.hyperscience.ai)
"""


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'


class OnErrorManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['on_error', 'supporting'],
            input=[
                Parameter(
                    name=OnErrorFlowInputKeys.failed_run_uuid,
                    type='string',
                    title='Failed Flow Run UUID',
                    description='UUID of the failed flow run that triggered this on_error flow',
                    optional=False,
                    ui={'hidden': True},
                )
            ],
        )


def submission_aware_on_error_flow() -> Flow:
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        title='Get Flow Run',
        description='Get the full metadata of the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{workflow_input(OnErrorFlowInputKeys.failed_run_uuid)}',
    )

    def _get_correlation_id(_hs_task: HsTask) -> dict:
        return {'correlation_id': _hs_task.correlation_id}

    get_correlation_id_block = PythonBlock(
        reference_name='get_correlation_id',
        title='Get Correlation ID',
        description='Get the correlation ID for the failed Submission / Flow Run',
        code=_get_correlation_id,
    )

    fetch_submissions_block = HyperscienceRestApiBlock(
        reference_name='fetch_submission',
        title='Fetch Submission',
        description='Fetch the submission data for the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/submissions'
        '?debug=true'
        f'&correlation_id={get_correlation_id_block.output("correlation_id")}',
    )

    def _prep_output_data(flow_run: dict, get_submissions_response: dict) -> dict:
        submissions_in_response = get_submissions_response.get('results', [])
        submission = None if not submissions_in_response else submissions_in_response[0]
        return {'flow_run': flow_run, 'submission': submission}

    prep_output_data_block = PythonBlock(
        title='Format Submission Response',
        description='Prepare the output data to be sent to the user-configured output'
        'under the keys "flow_run" and "submission"',
        code=_prep_output_data,
        code_input={
            'flow_run': flow_run_rest_api_block.output('result.data'),
            'get_submissions_response': fetch_submissions_block.output('result.data'),
        },
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': prep_output_data_block.output()})

    return Flow(
        uuid=UUID(SUBMISSION_AWARE_ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=SUBMISSION_AWARE_ON_ERROR_FLOW_TITLE,
        description=SUBMISSION_AWARE_ON_ERROR_DESCRIPTION,
        manifest=OnErrorManifest(SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER),
        output={},
        input={},
        blocks=[
            flow_run_rest_api_block,
            get_correlation_id_block,
            fetch_submissions_block,
            prep_output_data_block,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return submission_aware_on_error_flow()


if __name__ == '__main__':
    export_flow(submission_aware_on_error_flow())

OnError flow

This example shows how to produce an error message when a flow fails. The package includes the source code for an OnError flow, which can be configured to the IDP flow.

With the on-error flow configured, when the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered.

on_error_flow_v39.zip

OnError flow

from uuid import UUID

from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v39.additional_blocks import (
    HyperscienceRestApiBlock,
    JSONOutputsBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import workflow_input

ON_ERROR_FLOW_UUID = 'd1cce845-07ec-4bd4-9a9f-cb73b39b68e6'
ON_ERROR_FLOW_IDENTIFIER = 'ON_ERROR_V39'
ON_ERROR_TITLE = 'On Error V39'
ON_ERROR_DESCRIPTION = (
    'Flow designed to be used as an "on-error" handler for other flows.'
    '\n\n'
    'With the on-error flow configured, when the original flow fails for any reason, and all '
    'the configured auto-retries have been exhausted, the on-error flow will be triggered.'
    '\n\n'
    'The input it receives is the ID of the failed flow run, which can be used to query '
    'additional information about the failure.'
    '\n\n'
    'In this on-error flow implementation, '
    'we fetch the full metadata of the failed flow run and then send it '
    'to a user-configured output. '
    'The data format of the output conforms to '
    'the schema documented here: https://docs.hyperscience.ai/#flows-runs.'
    '\n\n'
    'If you need more complex logic in this flow, '
    'please consult the documentation of Flows SDK (https://flows-sdk.hyperscience.ai) '
    'and the available public APIs (https://docs.hyperscience.ai)'
)


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'


class OnErrorManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['on_error', 'supporting'],
            input=[
                Parameter(
                    name=OnErrorFlowInputKeys.failed_run_uuid,
                    type='string',
                    title='Failed Flow Run UUID',
                    description='UUID of the failed flow run that triggered this on_error flow',
                    optional=False,
                    ui={'hidden': True},
                ),
            ],
        )


def on_error_flow() -> Flow:
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{workflow_input(OnErrorFlowInputKeys.failed_run_uuid)}',
    )

    json_outputs_block = JSONOutputsBlock(
        inputs={'payload': flow_run_rest_api_block.output('result.data')}
    )

    return Flow(
        uuid=UUID(ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=ON_ERROR_TITLE,
        description=ON_ERROR_DESCRIPTION,
        manifest=OnErrorManifest(ON_ERROR_FLOW_IDENTIFIER),
        output={},
        input={},
        blocks=[
            flow_run_rest_api_block,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return on_error_flow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

V38

Name	Description	Release	Sample Docs	Download
IDP Starter V38	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v38.zip
Submission-aware OnError flow	Example implementation of an OnError flow that enriches the error message with submission information.			submission_aware_on_error.zip

IDP Starter V38

The basic document processing flow. From V38 on, the IDP Document processing is split into a wrapper flow (IDP) and a subflow (IDP Core). The package includes:

• IDP Core Flow - a read-only subflow containing the steps to process documents into machine-readable output

• Document Processing Flow - a wrapper around the IDP Core flow, used for managing input and output connector configuration and document processing workflow settings

• IDP Submission Notify Flow - A flow that can be configured for some blocks of the IDP Core Flow, used to send notifications to external systems when a submission has been created or is waiting for supervision

• On Error Flow - A flow that is used as an “on-error” handler for other flows. With the on-error flow configured, when the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered. It can be configured to the IDP flow.

The package includes also release and sample documents for processing Form W-9s.

Download idp_flow_v38.zip

IDP V38 Sample

from uuid import UUID

from flows_sdk.error_handling import ErrorHandling
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v38 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v38.idp_blocks import IDPCoreBlock
from flows_sdk.implementations.idp_v38.idp_values import (
    DEFAULT_IDP_CORE_IDENTIFIER,
    DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER,
    IdpWorkflowConfig,
    IDPWrapperManifest,
    Settings,
    get_idp_wf_config,
)
from flows_sdk.package_utils import export_flow

IDP_UUID = 'dab9f90b-e9ec-421f-a2cc-988ebdcf21f1'
IDP_IDENTIFIER = 'DOCUMENT_PROCESSING_EXAMPLE'
IDP_TITLE = 'Document Processing - flows-sdk example'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:
    # By default, all the notification flows are set to `None`.
    # I.e. the system will send no intermediate notifications for submission change events.
    # So let's specify the notification flows that we want to use explicitly here:
    idp_wf_config.notification_workflow_initialization = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_classification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_identification = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_transcription = DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    idp_wf_config.notification_workflow_flexible_extraction = (
        DEFAULT_IDP_SUBMISSION_NOTIFY_IDENTIFIER
    )

    idp_flow = IDPCoreBlock(idp_wf_config, identifier=DEFAULT_IDP_CORE_IDENTIFIER)

    outputs = idp_blocks.IDPOutputsBlock(inputs={'submission': idp_flow.output('submission')})

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_TITLE,
        description='Use this flow to manage your input and output connector configuration and '
        'document processing workflow settings.',
        manifest=IDPWrapperManifest(IDP_IDENTIFIER),
        output={'submission': idp_flow.output('submission')},
        triggers=idp_values.IDPTriggers(),
        input={
            Settings.LayoutReleaseUuid: None,
        },
        blocks=[
            idp_flow,
            outputs,
        ],
        error_handling=ErrorHandling(on_error_flow=None),
        # You can use the following error handling configuration to enable the submission aware on
        # error flow if your system has either Advanced or Premium license package
        # error_handling=ErrorHandling(
        #    on_error_flow=OnErrorFlow(identifier=DEFAULT_SUBMISSION_AWARE_ON_ERROR_FLOW_IDENTIFIER)
        # ),
    )


def entry_point_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Core V38 Sample

from uuid import UUID

from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v38 import idp_blocks
from flows_sdk.implementations.idp_v38.idp_values import IDPCoreManifest
from flows_sdk.package_utils import export_flow

# If you are creating a custom version of this flow, choose a different UUID and identifier
# here and pass the correct identifier into IDPCoreBlock(identifier=...) within your flow.
IDP_CORE_UUID = 'dca93e70-c742-419b-8d70-2ac41ea027c2'
IDP_CORE_IDENTIFIER = 'IDP_CORE_V38_EXAMPLE'
IDP_CORE_TITLE = 'Document Processing Subflow V38 - flows-sdk example'


def idp_workflow() -> Flow:
    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    reprocessing = idp_blocks.ReprocessingBlock(
        reference_name='reprocessing',
        submission=flexible_extraction.output('submission'),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=reprocessing.output('submission'),
        is_reprocessing=reprocessing.output('is_reprocessing'),
    )

    return Flow(
        uuid=UUID(IDP_CORE_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title=IDP_CORE_TITLE,
        description='A read-only subflow containing steps to process documents into '
        'machine-readable output.',
        manifest=IDPCoreManifest(
            flow_identifier=IDP_CORE_IDENTIFIER, roles=['read_only', 'supporting', 'idp_core']
        ),
        output={'submission': submission_complete.output('submission')},
        blocks=[
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            flexible_extraction,
            reprocessing,
            submission_complete,
        ],
    )


def entry_point_workflow() -> Flow:
    return idp_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

IDP Submission Notify V38 Sample

import os

from flows_sdk.blocks import Outputs
from flows_sdk.flows import Flow, Manifest
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import icon_path_to_base64, workflow_input

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

IDP_SUBMISSION_NOTIFY_IDENTIFIER = 'IDP_SUBMISSION_NOTIFY_V38_EXAMPLE'
IDP_SUBMISSION_NOTIFY_UUID = 'cf22dd1e-d013-4d09-b598-40696fa9a322'
IDP_SUBMISSION_NOTIFY_TITLE = 'Submission State Notifications V38 - flows-sdk example'

IDP_OUTPUT_ROLE = 'idp_output'


def idp_submission_notify_workflow() -> Flow:
    return Flow(
        title='Submission State Notifications V38',
        owner_email='flows.sdk@hyperscience.com',
        description=(
            'Send notifications to external systems when a submission has been '
            'created or is waiting for supervision. \n'
            f'If the "Document Processing" flow is live, this flow '
            'must also be live, but it can be empty.'
        ),
        is_user_facing=True,
        blocks=[
            Outputs(
                reference_name='outputs',
                title='Outputs',
                description=(
                    'Send submission data to external systems when a submission '
                    'has been created or is waiting for supervision'
                ),
                role_filter=[IDP_OUTPUT_ROLE],
                input_template={
                    'submission': {'id': workflow_input('submission.id')},
                    'enabled': True,
                },
                blocks=[],
            )
        ],
        uuid=IDP_SUBMISSION_NOTIFY_UUID,
        input={},
        output={},
        manifest=Manifest(
            identifier=IDP_SUBMISSION_NOTIFY_IDENTIFIER,
            input=[
                {
                    'name': 'submission',
                    'type': 'Submission',
                    'title': 'Submission Object',
                    'ui': {'hidden': True},
                }
            ],
            output=[],
            roles=['notifications', 'supporting'],
            ui={
                'hidden': True,
                'icon': icon_path_to_base64(os.path.join(BASE_DIR, 'notification_flow_icon.png')),
            },
        ),
    )


def entry_point_workflow() -> Flow:
    return idp_submission_notify_workflow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

On Error Flow V38 Sample

from uuid import UUID

from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v38.additional_blocks import (
    HyperscienceRestApiBlock,
    JSONOutputsBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.utils import workflow_input

ON_ERROR_FLOW_UUID = '46ad83ad-5a9c-46c5-a322-e2aa8e40bdce'
ON_ERROR_FLOW_TASK_NAME = 'ON_ERROR_V38_EXAMPLE'
ON_ERROR_DESCRIPTION = """Flow designed to be used as an "on-error" handler for other flows.
\n\n
With the on-error flow configured, when the original flow fails for any reason, and all
    the configured auto-retries have been exhausted, the on-error flow will be triggered.
\n\n
The input it receives is the ID of the failed flow run, which can be used to query
    additional information about the failure.
\n\n
In this on-error flow implementation,
    we fetch the full metadata of the failed flow run and then send it
    to a user-configured output.
    The data format of the output conforms to
    the schema documented here: https://docs.hyperscience.ai/#flows-runs.
\n\n
If you need more complex logic in this flow
    please consult the documentation of Flows SDK (https://flows-sdk.hyperscience.ai)
    and the available public APIs (https://docs.hyperscience.ai)
"""


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'


class OnErrorManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['on_error', 'supporting'],
            input=[
                Parameter(
                    name=OnErrorFlowInputKeys.failed_run_uuid,
                    type='string',
                    title='Failed Flow Run UUID',
                    description='UUID of the failed flow run that triggered this on_error flow',
                    optional=False,
                    ui={'hidden': True},
                ),
            ],
        )


def on_error_flow() -> Flow:

    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{workflow_input(OnErrorFlowInputKeys.failed_run_uuid)}',
    )

    json_outputs_block = JSONOutputsBlock(
        inputs={'payload': flow_run_rest_api_block.output('result.data')}
    )

    return Flow(
        uuid=UUID(ON_ERROR_FLOW_UUID),
        owner_email='flows.sdk@hyperscience.com',
        title='On-Error V38 - flows-sdk',
        description=ON_ERROR_DESCRIPTION,
        manifest=OnErrorManifest(ON_ERROR_FLOW_TASK_NAME),
        output={},
        input={},
        blocks=[
            flow_run_rest_api_block,
            json_outputs_block,
        ],
    )


def entry_point_workflow() -> Flow:
    return on_error_flow()


if __name__ == '__main__':
    export_flow(entry_point_workflow())

Submission-aware OnError flow

This example shows how to enrich an error message with submission information. The package includes the source code for a Submission-aware onError flow, which can be configured to the IDP flow.

With the on-error flow configured, when the original flow fails for any reason, and all the configured auto-retries have been exhausted, the on-error flow will be triggered. In case this OnError implementation is used on a non-IDP flow (i.e. a submissionless flow), submission data will be omitted.

submission_aware_on_error.zip

Submission-aware OnError flow

import os
from uuid import UUID

from flows_sdk.blocks import PythonBlock
from flows_sdk.flows import Flow, Manifest, Parameter
from flows_sdk.implementations.idp_v38.additional_blocks import (
    HyperscienceRestApiBlock,
    JSONOutputsBlock,
)
from flows_sdk.package_utils import export_flow
from flows_sdk.types import HsTask
from flows_sdk.utils import workflow_input

ON_ERROR_FLOW_UUID = '1e8605e7-db3f-4fae-ac25-029a7ffd4ba4'
ON_ERROR_FLOW_TASK_NAME = 'SUBMISSION_AWARE_ON_ERROR_FLOW'
ON_ERROR_DESCRIPTION = """
Flow designed to be used as an "on-error" handler for submission-based flows.
\n
With the on-error flow configured, when the original flow fails for any reason, and all
    the configured auto-retries have been exhausted, the on-error flow will be triggered.
\n
In this on-error flow implementation,
    we fetch the full metadata of the failed flow run, alongside with the full submission metadata
    (if the submission ID was found), and then send it to a user-configured output.
    The data format of the output conforms to
    the schema documented here: https://docs.hyperscience.ai/#flows-runs and
    here: https://docs.hyperscience.ai/#submissions, under the respective keys
    "flow_run" and "submission".
\n
If you need more complex logic in this flow
    please consult the documentation of Flows SDK (https://flows-sdk.hyperscience.ai)
    and the available public APIs (https://docs.hyperscience.ai)
""".lstrip()
CURRENT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))


class OnErrorFlowInputKeys:
    failed_run_uuid: str = 'failed_run_uuid'


class OnErrorManifest(Manifest):
    def __init__(self, flow_identifier: str):
        super().__init__(
            identifier=flow_identifier,
            roles=['on_error', 'supporting'],
            input=[
                Parameter(
                    name=OnErrorFlowInputKeys.failed_run_uuid,
                    type='string',
                    title='Failed Flow Run UUID',
                    description='UUID of the failed flow run that triggered this on_error flow',
                    optional=False,
                    ui={'hidden': True},
                )
            ],
        )


def on_error_flow() -> Flow:
    flow_run_rest_api_block = HyperscienceRestApiBlock(
        reference_name='flow_run_rest_api_block',
        title='Get Flow Run',
        description='Get the full metadata of the failed flow run',
        method='GET',
        app_endpoint=f'/api/v5/flow_runs/{workflow_input(OnErrorFlowInputKeys.failed_run_uuid)}',
    )

    def _get_correlation_id(_hs_task: HsTask) -> dict:
        return {'correlation_id': _hs_task.correlation_id}

    get_correlation_id_block = PythonBlock(
        reference_name='get_correlation_id',
        title='Get Correlation ID',
        description='Get the correlation ID for the failed Submission / Flow Run',
        code=_get_correlation_id,
    )

    fetch_submissions_block = HyperscienceRestApiBlock(
        reference_name='fetch_submission',
        title='Fetch Submission',
        description='Fetch the submission data for the failed flow run',
        method='GET',
        app_endpoint='/api/v5/submissions'
        '?debug=true'
        f'&correlation_id={get_correlation_id_block.output("correlation_id")}',
    )

    def _prep_output_data(flow_run: dict, get_submissions_response: dict) -> dict:
        submissions_in_response = get_submissions_response.get('results', [])
        submission = None if not submissions_in_response else submissions_in_response[0]
        return {'flow_run': flow_run, 'submission': submission}

    prep_output_data_block = PythonBlock(
        title='Format Submission Response',
        description='Prepare the output data to be sent to the user-configured output'
        'under the keys "flow_run" and "submission"',
        code=_prep_output_data,
        code_input={
            'flow_run': flow_run_rest_api_block.output('result.data'),
            'get_submissions_response': fetch_submissions_block.output('result.data'),
        },
    )

    json_outputs_block = JSONOutputsBlock(inputs={'payload': prep_output_data_block.output()})

    return Flow(
        uuid=UUID(ON_ERROR_FLOW_UUID),
        owner_email='flows-sdk@hyperscience.com',
        title='On-Error with included Submission data V38 - flows-sdk',
        description=ON_ERROR_DESCRIPTION,
        manifest=OnErrorManifest(ON_ERROR_FLOW_TASK_NAME),
        output={},
        input={},
        blocks=[
            flow_run_rest_api_block,
            get_correlation_id_block,
            fetch_submissions_block,
            prep_output_data_block,
            json_outputs_block,
        ],
    )


if __name__ == '__main__':
    export_flow(on_error_flow())

V37

Name	Description	Release	Sample Docs	Download
IDP Starter V37	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v37.zip

IDP Starter V37

This example is a snapshot of our V37 Intelligent Document Processing (IDP) flow. It uses both Machine and Manual blocks to:

1. Classify and collate pages

2. Identify and transcribe text

3. Additional manual transcription for any fields marked for review

4. Output results to a downstream system

Download idp_flow_v37.zip

IDP Starter V37 flow

from uuid import UUID

from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v37 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v37.idp_values import (
    IdpWorkflowConfig,
    get_idp_wf_config,
    get_idp_wf_inputs,
)
from flows_sdk.package_utils import export_flow

IDP_UUID = 'd5ba954c-8fde-437b-92ff-c87408723f91'
IDP_TASK_NAME = 'IDP_V37_EXAMPLE'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        rotation_correction_enabled=idp_wf_config.rotation_correction_enabled,
        mobile_processing_enabled=False,
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        table_output_manual_review=(
            idp_wf_config.manual_transcription_config.table_output_manual_review
        ),
        supervision_transcription_masking=(
            idp_wf_config.manual_transcription_config.supervision_transcription_masking
        ),
        task_restrictions=idp_wf_config.manual_transcription_config.task_restrictions,
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        task_restrictions=idp_wf_config.flexible_extraction_config.task_restrictions,
        supervision_transcription_masking=(
            idp_wf_config.flexible_extraction_config.supervision_transcription_masking
        ),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission', submission=flexible_extraction.output('submission')
    )

    outputs = idp_blocks.IDPOutputsBlock(
        inputs={'submission': bootstrap_submission.output('submission')}
    )

    inputs = get_idp_wf_inputs(idp_wf_config)

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows.sdk@hyperscience.com',
        title='Document Processing V37 - flows-sdk example',
        description='Document Processing V37 - flows-sdk example',
        manifest=idp_values.IDPManifest(flow_identifier=IDP_TASK_NAME),
        output={'submission': submission_complete.output()},
        triggers=idp_values.IDPTriggers(),
        input=inputs,
        blocks=[
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            flexible_extraction,
            submission_complete,
            outputs,
        ],
    )


def entry_point_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(flow=entry_point_workflow())

V36

Name	Description	Release	Sample Docs	Download
IDP Starter V36	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v36.zip

IDP Starter V36

This example is a snapshot of our V36 Intelligent Document Processing (IDP) flow. It uses both Machine and Manual blocks to:

1. Classify and collate pages

2. Identify and transcribe text

3. Additional manual transcription for any fields marked for review

4. Output results to a downstream system

Download idp_flow_v36.zip

IDP Starter V36 flow

from uuid import UUID

from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v36 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v36.idp_values import (
    IdpWorkflowConfig,
    get_idp_wf_config,
    get_idp_wf_inputs,
)
from flows_sdk.package_utils import export_flow

IDP_UUID = '2667344c-ffef-4021-86c1-108fd54b6654'
IDP_TASK_NAME = 'IDP_V36_EXAMPLE'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        rotation_correction_enabled=idp_wf_config.rotation_correction_enabled,
        mobile_processing_enabled=False,
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        table_output_manual_review=(
            idp_wf_config.manual_transcription_config.table_output_manual_review
        ),
        supervision_transcription_masking=(
            idp_wf_config.manual_transcription_config.supervision_transcription_masking
        ),
        task_restrictions=idp_wf_config.manual_transcription_config.task_restrictions,
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        task_restrictions=idp_wf_config.flexible_extraction_config.task_restrictions,
        supervision_transcription_masking=(
            idp_wf_config.flexible_extraction_config.supervision_transcription_masking
        ),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission', submission=flexible_extraction.output('submission')
    )

    outputs = idp_blocks.IDPOutputsBlock(
        inputs={'submission': bootstrap_submission.output('submission')}
    )

    inputs = get_idp_wf_inputs(idp_wf_config)

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows.sdk@hyperscience.com',
        title='Document Processing V36 - flows-sdk example',
        description='Document Processing V36 - flows-sdk example',
        manifest=idp_values.IDPManifest(flow_identifier=IDP_TASK_NAME),
        output={'submission': submission_complete.output()},
        triggers=idp_values.IDPTriggers(),
        input=inputs,
        blocks=[
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            flexible_extraction,
            submission_complete,
            outputs,
        ],
    )


def entry_point_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(flow=entry_point_workflow())

V35

Name	Description	Release	Sample Docs	Download
IDP Starter V35	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v35.zip
Full Page Transcription V35	Transcribes text from documents.			idp_fpt_flow_v35.zip

IDP Starter V35

This example is a snapshot of our V35 Intelligent Document Processing (IDP) flow. It uses both Machine and Manual blocks to:

1. Classify and collate pages

2. Identify and transcribe text

3. Additional manual transcription for any fields marked for review

4. Output results to a downstream system

Download idp_flow_v35.zip

IDP Starter V35 flow

from uuid import UUID

from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v35 import idp_blocks, idp_values
from flows_sdk.implementations.idp_v35.idp_values import (
    IdpWorkflowConfig,
    get_idp_wf_config,
    get_idp_wf_inputs,
)
from flows_sdk.package_utils import export_flow

IDP_UUID = 'c25cc470-e2c4-4696-aeaa-26ba6cac255a'
IDP_TASK_NAME = 'IDP_V35_FLOW_EXAMPLE'


def idp_workflow(idp_wf_config: IdpWorkflowConfig) -> Flow:

    bootstrap_submission = idp_blocks.SubmissionBootstrapBlock(
        reference_name='submission_bootstrap',
    )

    case_collation_task = idp_blocks.MachineCollationBlock(
        reference_name='machine_collation',
        submission=bootstrap_submission.output('submission'),
        cases=bootstrap_submission.output('api_params.cases'),
    )

    machine_classification = idp_blocks.MachineClassificationBlock(
        reference_name='machine_classification',
        submission=case_collation_task.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        rotation_correction_enabled=idp_wf_config.rotation_correction_enabled,
        mobile_processing_enabled=False,
    )

    manual_classification = idp_blocks.ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_identification = idp_blocks.MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_identification = idp_blocks.ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    machine_transcription = idp_blocks.MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
    )

    manual_transcription = idp_blocks.ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        table_output_manual_review=(
            idp_wf_config.manual_transcription_config.table_output_manual_review
        ),
        supervision_transcription_masking=(
            idp_wf_config.manual_transcription_config.supervision_transcription_masking
        ),
        task_restrictions=idp_wf_config.manual_transcription_config.task_restrictions,
    )

    flexible_extraction = idp_blocks.FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription.output('submission'),
        api_params=bootstrap_submission.output('api_params'),
        task_restrictions=idp_wf_config.flexible_extraction_config.task_restrictions,
        supervision_transcription_masking=(
            idp_wf_config.flexible_extraction_config.supervision_transcription_masking
        ),
    )

    submission_complete = idp_blocks.SubmissionCompleteBlock(
        reference_name='complete_submission', submission=flexible_extraction.output('submission')
    )

    outputs = idp_blocks.IDPOutputsBlock(
        inputs={'submission': bootstrap_submission.output('submission')}
    )

    inputs = get_idp_wf_inputs(idp_wf_config)

    return Flow(
        uuid=UUID(IDP_UUID),
        owner_email='flows.sdk@hyperscience.com',
        title='Document Processing V35',
        description='Document Processing V35',
        manifest=idp_values.IDPManifest(flow_identifier=IDP_TASK_NAME),
        output={'submission': submission_complete.output()},
        triggers=idp_values.IDPTriggers(),
        input=inputs,
        blocks=[
            bootstrap_submission,
            case_collation_task,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            flexible_extraction,
            submission_complete,
            outputs,
        ],
    )


def entry_point_idp_flow() -> Flow:
    idp_wf_config = get_idp_wf_config()
    return idp_workflow(idp_wf_config)


if __name__ == '__main__':
    export_flow(flow=entry_point_idp_flow())

Full Page Transcription V35

This example transcribes text from documents.

Download idp_fpt_flow_v35.zip

Full Page Transcription V35 flow

from uuid import UUID

from flows_sdk.flows import Flow, Manifest
from flows_sdk.implementations.idp_v35.idp_blocks import (
    IDPFullPageTranscriptionBlock,
    IDPImageCorrectionBlock,
    IDPOutputsBlock,
    SubmissionBootstrapBlock,
    SubmissionCompleteBlock,
)
from flows_sdk.implementations.idp_v35.idp_values import IDPTriggers
from flows_sdk.package_utils import export_flow

IDENTIFIER = 'IDP_FULL_PAGE_TRANSCRIPTION_V35'
FLOW_UUID = '9f27b706-e063-473e-8eea-43896d1e6214'


def idp_fpt_workflow() -> Flow:

    submission_bootstrap = SubmissionBootstrapBlock(reference_name='submission_bootstrap')

    image_correction = IDPImageCorrectionBlock(
        reference_name='image_correction', submission=submission_bootstrap.output('submission')
    )

    full_page_transcription = IDPFullPageTranscriptionBlock(
        reference_name='full_page_transcription', submission=image_correction.output('submission')
    )

    # Submission complete block finalizes submission processing and updates reporting data
    # Every flow needs a complete block because it initiates Quality Assurance tasks and
    # changes the submission's status to "Complete"
    # In this example, submission complete block receives the submission object from
    # marked_as_complete custom code block
    submission_complete = SubmissionCompleteBlock(
        reference_name='complete_submission',
        payload=full_page_transcription.output('submission'),
        submission=full_page_transcription.output('submission'),
        nlc_qa_sampling_ratio=0,
        field_id_qa_enabled=False,
        field_id_qa_sampling_ratio=0,
        table_id_qa_enabled=False,
        table_id_qa_sampling_ratio=0,
        transcription_qa_enabled=False,
        transcription_qa_sampling_ratio=0,
        table_cell_transcription_qa_enabled=False,
        table_cell_transcription_qa_sample_rate=0,
    )

    outputs = IDPOutputsBlock(
        inputs={'submission': submission_bootstrap.output('submission')}, blocks=[]
    )

    triggers = IDPTriggers(blocks=[])

    return Flow(
        uuid=UUID(FLOW_UUID),
        owner_email='flows.sdk@hyperscience.com',
        title='Full Page Transcription V35',
        manifest=Manifest(identifier=IDENTIFIER, input=[]),
        triggers=triggers,
        blocks=[
            submission_bootstrap,
            image_correction,
            full_page_transcription,
            submission_complete,
            outputs,
        ],
        description='IDP Full Page Transcription V35',
        output={'submission': submission_complete.output()},
    )


if __name__ == '__main__':
    export_flow(flow=idp_fpt_workflow())

V34

Name	Description	Release	Sample Docs	Download
IDP Starter V34	Basic document processing flow. Package includes release and sample documents for processing Form W-9s.			idp_flow_v34.zip
IDP Custom Supervision V34	Sends all fields and tables from the submission to custom supervision. It will automatically collate all classified submission documents into a newly generated case if no case ID was submitted.			idp_custom_supervision_v34.zip
IDP Full Page Transcription V34	Transcribes text from documents.			idp_fpt_flow_v34.zip

IDP Starter V34

This example is a snapshot of our V34 Intelligent Document Processing (IDP) flow. It uses both Machine and Manual blocks to:

1. Classify and collate pages

2. Identify and transcribe text

3. Additional manual transcription for any fields marked for review

4. Output results to a downstream system

Download idp_flow_v34.zip

IDP Starter V34 flow

from uuid import UUID

from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v34.idp_blocks import (
    FlexibleExtractionBlock,
    IDPOutputsBlock,
    MachineClassificationBlock,
    MachineCollationBlock,
    MachineIdentificationBlock,
    MachineTranscriptionBlock,
    ManualClassificationBlock,
    ManualIdentificationBlock,
    ManualTranscriptionBlock,
    SubmissionBootstrapBlock,
    SubmissionCompleteBlock,
)
from flows_sdk.implementations.idp_v34.idp_values import (
    IDPManifest,
    IDPTriggers,
    get_idp_wf_config,
    get_idp_wf_inputs,
)
from flows_sdk.package_utils import export_flow


def entry_point_flow() -> Flow:
    return idp_workflow()


def idp_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()

    # The idp flow basically processes, modifies and propagates the submission object from
    # block to block
    # Each block's processing result is usually included in the submission object

    # Submission bootstrap block initializes the submission object and prepares external images
    # or other submission data if needed
    submission_bootstrap = SubmissionBootstrapBlock(reference_name='submission_bootstrap')

    # Case collation block groups files, documents and pages (from the submission) into cases
    # In this example, case collation block receives the submission object and the cases
    # information from submission bootstrap block
    machine_collation = MachineCollationBlock(
        reference_name='machine_collation',
        submission=submission_bootstrap.output('submission'),
        cases=submission_bootstrap.output('api_params.cases'),
    )

    # Machine classification block automatically matches documents to structured, semi-structured
    # or additional layouts
    # In this example, machine classification block receives the submission object from
    # case collation block
    machine_classification = MachineClassificationBlock(
        reference_name='machine_classification',
        submission=machine_collation.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
        rotation_correction_enabled=idp_wf_config.rotation_correction_enabled,
    )

    # Manual classification block allows keyers to manually match submissions to their layouts.
    # Keyers may perform manual classification if machine classification cannot automatically
    # match a submission to a layout with high confidence
    # In this example, manual classification block receives the submission object from machine
    # classification block
    manual_classification = ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
    )

    # Machine identification automatically identify fields and tables in the submission
    # In this example, machine identification block receives the submission object from manual
    # classification
    machine_identification = MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
    )

    # Manual identification allows keyers to complete field identification or table identification
    # tasks, where they draw bounding boxes around the contents of certain fields, table columns
    # or table rows. This identification process ensures that the system will be able to
    # transcribe the correct content in the upcoming transcription process
    # In this example, manual identification block receives the submission object from machine
    # identification
    manual_identification = ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
        task_restrictions=idp_wf_config.manual_identification_config.task_restrictions,
    )

    # Machine transcription automatically transcribes the content of your submission
    # In this example, machine identification block receives the submission object from manual
    # identification
    machine_transcription = MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
    )

    # Manual transcription lets your keyers manually enter the text found in fields or tables
    # that could not be automatically transcribed
    # In this example, manual transcription block receives the submission object from machine
    # transcription block
    manual_transcription = ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
        supervision_transcription_masking=(
            idp_wf_config.manual_transcription_config.supervision_transcription_masking
        ),
        table_output_manual_review=(
            idp_wf_config.manual_transcription_config.table_output_manual_review
        ),
        task_restrictions=idp_wf_config.manual_transcription_config.task_restrictions,
    )

    # Flexible extraction manually transcribes fields marked for review
    # In this example, flexible extraction block receives the submission object from manual
    # transcription block
    flexible_extraction = FlexibleExtractionBlock(
        reference_name='flexible_extraction',
        submission=manual_transcription.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
        supervision_transcription_masking=(
            idp_wf_config.flexible_extraction_config.supervision_transcription_masking
        ),
        task_restrictions=idp_wf_config.flexible_extraction_config.task_restrictions,
    )

    # Submission complete block finalizes submission processing and updates reporting data
    # Every flow needs a complete block because it initiates Quality Assurance tasks and
    # changes the submission's status to "Complete"
    # In this example, submission complete block receives the submission object from
    # marked_as_complete custom code block
    submission_complete = SubmissionCompleteBlock(
        reference_name='complete_submission', submission=flexible_extraction.output('submission')
    )

    # Output block allows users to send data extracted by this idp flow to other systems
    # for downstream processing
    # In this example, this is an empty output block that does not do anything by default
    outputs = IDPOutputsBlock(inputs={'submission': submission_bootstrap.output('submission')})

    return Flow(
        # Flows should have a deterministic UUID ensuring cross-system consistency
        uuid=UUID('0c2ee4ee-0fcf-4d9b-b57a-d90519d2a002'),
        owner_email='flows.sdk@hyperscience.com',
        title='IDP Starter Example (V34)',
        # Flow identifiers are globally unique
        manifest=IDPManifest(flow_identifier='IDP_V34_FLOW_EXAMPLE'),
        # It is important to include all blocks that are instantiated here in the blocks
        # field and make sure they follow the order of the flow. For example, if machine
        # classification depends on the output of case collation, then machine_collation must
        # come before machine_classification in this blocks array
        blocks=[
            submission_bootstrap,
            machine_collation,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            flexible_extraction,
            submission_complete,
            outputs,
        ],
        input=get_idp_wf_inputs(idp_wf_config),
        description='A basic approach to extracting information from documents.',
        triggers=IDPTriggers(),
        output={'submission': submission_complete.output()},
    )


if __name__ == '__main__':
    export_flow(flow=entry_point_flow())

IDP Custom Supervision V34

This example sends all fields and tables in the submission to Custom Supervision. It will automatically collate all classified submission documents into a newly generated case if no case ID was submitted.

Download idp_custom_supervision_v34.zip

IDP Custom Supervision V34 flow

from typing import Any, List
from uuid import UUID

from flows_sdk.blocks import CodeBlock, Routing
from flows_sdk.flows import Flow
from flows_sdk.implementations.idp_v34.idp_blocks import (
    IdpCustomSupervisionBlock,
    IDPOutputsBlock,
    MachineClassificationBlock,
    MachineCollationBlock,
    MachineIdentificationBlock,
    MachineTranscriptionBlock,
    ManualClassificationBlock,
    ManualIdentificationBlock,
    ManualTranscriptionBlock,
    SubmissionBootstrapBlock,
    SubmissionCompleteBlock,
)
from flows_sdk.implementations.idp_v34.idp_values import (
    IDPManifest,
    IDPTriggers,
    get_idp_wf_config,
    get_idp_wf_inputs,
)
from flows_sdk.package_utils import export_flow


def entry_point_flow() -> Flow:
    return idp_workflow()


def _collate_classified_documents(manual_transcription: Any) -> List:
    """
    Helper method responsible for an optional flow that executes the following logic:
    - If any documents in the submission already have a case ID, then no operation is necessary
    - If none of the documents in the submission are assigned to a case,
        then create a new case containing all the CLASSIFIED documents in the submission.
        Unassigned pages should not receive case IDs
    """

    # Check the submission object's case field to see if a case has been created
    # thus far in the flow. If not, we should collate all the classified documents
    # in the submission. Otherwise, we can skip this operation.
    should_add_docs_to_case = CodeBlock(
        reference_name='should_add_docs_to_case',
        title='Should Add Classified Documents to a Case',
        description='If there are no cases assigned in the submission, group all classified '
        'documents into a newly created case for Custom Supervision',
        code=lambda cases: {'add_docs': 'true' if not cases else 'false'},
        code_input={'cases': manual_transcription.output('submission.cases')},
    )

    # Here we define the custom method used to grab all documents in the submission.
    def _grab_classified_doc_ids_fn(submission: Any) -> Any:
        doc_list = []
        for doc in submission.get('documents', []):
            doc_list.append(doc['id'])

        return doc_list

    # Custom Code Block to parse the submission object to find all documents. Classified
    # Documents automatically get categorized in this field.
    get_classified_doc_ids = CodeBlock(
        reference_name='get_classified_doc_ids',
        code=_grab_classified_doc_ids_fn,
        code_input={'submission': manual_transcription.output('submission')},
        title='Get All Document IDs',
        description='Grab all the Document IDs present in this submission '
        'to collate into a new case for Custom Supervision',
    )

    # We create and format the appropriate payload for our Machine Collation block to ingest.
    # Note that we are passing a null value for our external_case_id so that we can
    # automatically generate a new case ID.
    format_case_input = CodeBlock(
        reference_name='format_case_input',
        title='Format Classified Documents for Machine Collation',
        description='Creates the payload for Machine Collation',
        code=lambda docs: [{'external_case_id': None, 'documents': docs}],
        code_input={'docs': get_classified_doc_ids.output()},
    )

    # The Machine Collation Block creates the case and collates the classified documents
    # into the newly created case.
    group_into_cases = MachineCollationBlock(
        reference_name='pre_custom_sv_case_collation',
        submission=manual_transcription.output('submission'),
        cases=format_case_input.output(),
    )

    # The Routing block that is responsible for conditionally executing this flow.
    # Subsequent blocks are not reliant on the output of this Routing block.
    group_classified_doc_to_case = Routing(
        reference_name='group_classified_doc_to_case',
        decision=should_add_docs_to_case.output('add_docs'),
        branches=[
            Routing.Branch(
                case='true',
                blocks=[get_classified_doc_ids, format_case_input, group_into_cases],
                label='Collate Classified Documents',
            )
        ],
        default_branch=Routing.DefaultBranch(
            blocks=[
                CodeBlock(
                    reference_name='no_op_case_collation',
                    title='No Operation',
                    description='Pass the existing submission along without any changes',
                    code=lambda submission: {'submission': submission},
                    code_input={'submission': manual_transcription.output('submission')},
                )
            ]
        ),
    )

    return [should_add_docs_to_case, group_classified_doc_to_case]


def idp_workflow() -> Flow:
    idp_wf_config = get_idp_wf_config()

    # The idp flow basically processes, modifies and propagates the submission object from
    # block to block
    # Each block's processing result is usually included in the submission object

    # Submission bootstrap block initializes the submission object and prepares external images
    # or other submission data if needed
    submission_bootstrap = SubmissionBootstrapBlock(reference_name='submission_bootstrap')

    # Case collation block groups files, documents and pages (from the submission) into cases
    # In this example, case collation block receives the submission object and the cases
    # information from submission bootstrap block
    machine_collation = MachineCollationBlock(
        reference_name='machine_collation',
        submission=submission_bootstrap.output('submission'),
        cases=submission_bootstrap.output('api_params.cases'),
    )

    # Machine classification block automatically matches documents to structured, semi-structured
    # or additional layouts
    # In this example, machine classification block receives the submission object from
    # case collation block
    machine_classification = MachineClassificationBlock(
        reference_name='machine_classification',
        submission=machine_collation.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
        rotation_correction_enabled=idp_wf_config.rotation_correction_enabled,
    )

    # Manual classification block allows keyers to manually match submissions to their layouts.
    # Keyers may perform manual classification if machine classification cannot automatically
    # match a submission to a layout with high confidence
    # In this example, manual classification block receives the submission object from machine
    # classification block
    manual_classification = ManualClassificationBlock(
        reference_name='manual_classification',
        submission=machine_classification.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
    )

    # Machine identification automatically identify fields and tables in the submission
    # In this example, machine identification block receives the submission object from manual
    # classification
    machine_identification = MachineIdentificationBlock(
        reference_name='machine_identification',
        submission=manual_classification.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
    )

    # Manual identification allows keyers to complete field identification or table identification
    # tasks, where they draw bounding boxes around the contents of certain fields, table columns
    # or table rows. This identification process ensures that the system will be able to
    # transcribe the correct content in the upcoming transcription process
    # In this example, manual identification block receives the submission object from machine
    # identification
    manual_identification = ManualIdentificationBlock(
        reference_name='manual_identification',
        submission=machine_identification.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
        task_restrictions=idp_wf_config.manual_identification_config.task_restrictions,
    )

    # Machine transcription automatically transcribes the content of your submission
    # In this example, machine identification block receives the submission object from manual
    # identification
    machine_transcription = MachineTranscriptionBlock(
        reference_name='machine_transcription',
        submission=manual_identification.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
    )

    # Manual transcription lets your keyers manually enter the text found in fields or tables
    # that could not be automatically transcribed
    # In this example, manual transcription block receives the submission object from machine
    # transcription block
    manual_transcription = ManualTranscriptionBlock(
        reference_name='manual_transcription',
        submission=machine_transcription.output('submission'),
        api_params=submission_bootstrap.output('api_params'),
        supervision_transcription_masking=(
            idp_wf_config.manual_transcription_config.supervision_transcription_masking
        ),
        table_output_manual_review=(
            idp_wf_config.manual_transcription_config.table_output_manual_review
        ),
        task_restrictions=idp_wf_config.manual_transcription_config.task_restrictions,
    )

    # Here we define a function for a custom code block that gets all fields and tables from a
    # submission to display them in custom supervision. It makes sure they appear only when an
    # appropriate layout has been selected by using the dependencies settings.
    # We also create a decision for the first cell in each table.
    def _get_transcriptions_for_custom_sv_fn(submission: Any) -> Any:
        transcription_entities = []
        table_cell_decisions = []
        visited = set()
        for doc in submission.get('documents', []):
            fields = sorted(doc.get('document_fields', []), key=lambda f: f['id'])
            for field in fields:
                if field['layout_field_uuid'] in visited:
                    continue
                visited.add(field['layout_field_uuid'])
                transcription_entities.append(
                    {
                        'name': f"{field['field_name']}_{field['layout_field_uuid']}",
                        'type': 'transcription',
                        'data_identifier': field['layout_field_uuid'],
                        'ui': {'hidden': True},
                        'dependencies': [
                            {
                                'condition': {
                                    'properties': {'layoutId': {'const': doc['layout_uuid']}}
                                },
                                'override': {'ui': {'hidden': False}},
                            }
                        ],
                    }
                )
            for table in doc.get('tables', []):
                if table['layout_table_uuid'] in visited:
                    continue
                visited.add(table['layout_table_uuid'])
                transcription_entities.append(
                    {
                        'name': table['layout_table_uuid'],
                        'type': 'table_transcription',
                        'data_identifier': table['layout_table_uuid'],
                        'ui': {'hidden': True},
                        'dependencies': [
                            {
                                'condition': {
                                    'properties': {'layoutId': {'const': doc['layout_uuid']}}
                                },
                                'override': {'ui': {'hidden': False}},
                            }
                        ],
                    },
                )
                try:
                    table_cell_decisions.append(
                        {
                            'name': f'table_cell_decision_{table["layout_table_uuid"]}',
                            'type': 'decision',
                            'title': 'Table Cell Decision',
                            'ui': {'hidden': True},
                            'relation': {
                                'type': 'table_cell',
                                'match': table['columns'][0]['cells'][0]['uuid'],
                            },
                            'dependencies': [
                                {
                                    'condition': {
                                        'properties': {'layoutId': {'const': doc['layout_uuid']}}
                                    },
                                    'override': {'ui': {'hidden': False}},
                                }
                            ],
                            'schema': {
                                'oneOf': [
                                    {'const': 'accept', 'title': 'Accept Table Cell'},
                                    {'const': 'reject', 'title': 'Reject Table Cell'},
                                ]
                            },
                        },
                    )
                except (IndexError, KeyError):
                    pass
        return transcription_entities, table_cell_decisions

    # Create our Custom Code block using our function from above.
    get_transcriptions_for_custom_sv = CodeBlock(
        reference_name='get_transcriptions_for_custom_sv',
        code=_get_transcriptions_for_custom_sv_fn,
        code_input={'submission': manual_transcription.output('submission')},
        title='Get All Transcriptions',
        description='Find all transcriptions in this submission to include in Custom Supervision',
    )

    # Using the output of our Custom Code block, we can construct our three column template
    # to define our Custom Supervision task.
    format_sv_template = CodeBlock(
        title='Format Custom Supervision Template',
        description='Creates the supervision template to define our Custom Supervision task',
        reference_name='format_sv_template',
        code=lambda transcriptions: [
            {
                'name': 'three_column_template',
                'version': '1.0',
                'thumbnail': {'group_by_document': True, 'group_by_case': True},
                'action': [
                    {
                        'name': 'Transcriptions',
                        'display': 'All Identified Fields and Tables',
                        'input': transcriptions[0],
                    },
                    {
                        'name': 'Decision and Case',
                        'display': 'Make Decision - Add Case',
                        'input': [
                            *transcriptions[1],
                            {
                                'name': 'document_decision',
                                'type': 'decision',
                                'title': 'Document Decision',
                                'relation': {'type': 'document'},
                                'schema': {
                                    'oneOf': [
                                        {'const': 'accept', 'title': 'Accept Document'},
                                        {'const': 'reject', 'title': 'Reject Document'},
                                    ]
                                },
                            },
                            {'name': 'assign_to_case', 'type': 'case_dropdown', 'title': ''},
                        ],
                    },
                ],
            }
        ],
        code_input={
            'transcriptions': get_transcriptions_for_custom_sv.output(),
        },
    )

    # Create our Custom Supervision task using the template we constructed via Custom Code blocks.
    # This task will have all fields from the submission so that the user can edit them, as well
    # as a document level decision ("accept" or "reject" the document). It will also have the
    # option to add a document to a case.
    idp_custom_supervision = IdpCustomSupervisionBlock(
        reference_name='idp_custom_supervision',
        submission=manual_transcription.output('submission'),
        task_purpose='example_custom_supervision',
        supervision_template=format_sv_template.output(),
    )

    # Case collation block groups files, documents and pages (from the submission) into cases
    # In this example, case collation block receives the submission object and the cases
    # information from the custom supervision block. This block is required after Custom
    # Supervision in order to propagate case ID changes to the IDP database.
    custom_supervision_collation = MachineCollationBlock(
        reference_name='custom_supervision_collation',
        submission=idp_custom_supervision.output('submission'),
        cases=idp_custom_supervision.output('cases'),
        remove_from_cases=idp_custom_supervision.output('remove_from_cases'),
    )

    # Submission complete block finalizes submission processing and updates reporting data
    # Every flow needs a complete block because it initiates Quality Assurance tasks and
    # changes the submission's status to "Complete"
    # In this example, submission complete block receives the submission object from
    # marked_as_complete custom code block
    submission_complete = SubmissionCompleteBlock(
        reference_name='complete_submission',
        submission=custom_supervision_collation.output('submission'),
    )

    # Output block allows users to send data extracted by this idp flow to other systems
    # for downstream processing
    # In this example, this is an empty output block that does not do anything by default
    outputs = IDPOutputsBlock(inputs={'submission': submission_bootstrap.output('submission')})

    return Flow(
        # Flows should have a deterministic UUID ensuring cross-system consistency
        uuid=UUID('4cc70d7f-43c1-43ff-a7c7-708b43610558'),
        owner_email='flows.sdk@hyperscience.com',
        title='IDP Custom Supervision Block Flow Example (V34)',
        # Flow identifiers are globally unique
        manifest=IDPManifest(flow_identifier='IDP_CUSTOM_SUPERVISION_V34_EXAMPLE'),
        # It is important to include all blocks that are instantiated here in the blocks
        # field and make sure they follow the order of the flow. For example, if machine
        # classification depends on the output of case collation, then machine_collation must
        # come before machine_classification in this blocks array
        blocks=[
            submission_bootstrap,
            machine_collation,
            machine_classification,
            manual_classification,
            machine_identification,
            manual_identification,
            machine_transcription,
            manual_transcription,
            *_collate_classified_documents(manual_transcription),
            get_transcriptions_for_custom_sv,
            format_sv_template,
            idp_custom_supervision,
            custom_supervision_collation,
            submission_complete,
            outputs,
        ],
        input=get_idp_wf_inputs(idp_wf_config),
        description='IDP Custom Supervision - All Fields (V34)',
        triggers=IDPTriggers(),
        output={'submission': submission_complete.output()},
    )


if __name__ == '__main__':
    export_flow(flow=entry_point_flow())

IDP Full Page Transcription V34

This example transcribes text from documents.

Download idp_fpt_flow_v34.zip

IDP Full Page Transcription V34 flow

from uuid import UUID

from flows_sdk.flows import Flow, Manifest
from flows_sdk.implementations.idp_v34.idp_blocks import (
    IDPFullPageTranscriptionBlock,
    IDPImageCorrectionBlock,
    IDPOutputsBlock,
    SubmissionBootstrapBlock,
    SubmissionCompleteBlock,
)
from flows_sdk.implementations.idp_v34.idp_values import IDPTriggers
from flows_sdk.package_utils import export_flow


def entry_point_flow() -> Flow:
    return idp_fpt_workflow()


def idp_fpt_workflow() -> Flow:
    # The idp flow basically processes, modifies and propagates the submission object from
    # block to block
    # Each block's processing result is usually included in the submission object

    # Submission bootstrap block initializes the submission object and prepares external images
    # or other submission data if needed
    submission_bootstrap = SubmissionBootstrapBlock(reference_name='submission_bootstrap')

    image_correction = IDPImageCorrectionBlock(
        reference_name='image_correction', submission=submission_bootstrap.output('submission')
    )

    full_page_transcription = IDPFullPageTranscriptionBlock(
        reference_name='full_page_transcription', submission=image_correction.output('submission')
    )

    # Submission complete block finalizes submission processing and updates reporting data
    # Every flow needs a complete block because it initiates Quality Assurance tasks and
    # changes the submission's status to "Complete"
    # In this example, submission complete block receives the submission object from
    # marked_as_complete custom code block
    submission_complete = SubmissionCompleteBlock(
        reference_name='complete_submission',
        payload=full_page_transcription.output('submission'),
        submission=full_page_transcription.output('submission'),
        nlc_qa_sampling_ratio=0,
        field_id_qa_enabled=False,
        field_id_qa_sampling_ratio=0,
        table_id_qa_enabled=False,
        table_id_qa_sampling_ratio=0,
        transcription_qa_enabled=False,
        transcription_qa_sampling_ratio=0,
        table_cell_transcription_qa_enabled=False,
        table_cell_transcription_qa_sample_rate=0,
    )

    # Output block allows users to send data extracted by this idp flow to other systems
    # for downstream processing
    # In this example, no output block is instantiated (blocks=[])
    # Setting up output blocks via UI and leaving this empty is recommended
    outputs = IDPOutputsBlock(
        inputs={'submission': submission_bootstrap.output('submission')}, blocks=[]
    )

    # Trigger block allows users to send data to idp flow via sources other than the User Interface
    # In this example, no trigger block is instantiated (blocks=[])
    # Setting up trigger blocks via UI and leaving this empty is recommended
    triggers = IDPTriggers(blocks=[])

    return Flow(
        # Flows should have a deterministic UUID ensuring cross-system consistency
        uuid=UUID('0dd837ae-44da-425a-b4be-9ffa3fc40eab'),
        owner_email='flows.sdk@hyperscience.com',
        title='IDP Full Page Transcription (V34)',
        # Flow identifiers are globally unique
        manifest=Manifest(identifier='IDP_FULL_PAGE_TRANSCRIPTION_V34', input=[]),
        triggers=triggers,
        # It is important to include all blocks that are instantiated here in the blocks
        # field and make sure they follow the order of the flow. For example, if machine
        # classification depends on the output of case collation, then case_collation must
        # come before machine_classification in this blocks array
        blocks=[
            submission_bootstrap,
            image_correction,
            full_page_transcription,
            submission_complete,
            outputs,
        ],
        description='IDP Full Page Transcription (V34)',
        output={'submission': submission_complete.output()},
    )


if __name__ == '__main__':
    export_flow(flow=entry_point_flow())