> ## Documentation Index
> Fetch the complete documentation index at: https://developer.watson-orchestrate.ibm.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Testing agentic workflow

After importing an agentic workflow tool, you can test it using a main Python script before add it to an agent. The following steps guide you through testing a flow, including defining, compiling, and running the flow:

<Steps>
  <Step title="Create a test script">
    Agentic workflow are asynchronous, and so are the functions used to run them.

    Start by creating a script that defines a function to invoke your agentic workflow. In the example below, the script `hello_message_flow.py` contains a function `build_hello_message_flow`, decorated with `@flow`, which returns the `Flow` object you want to test.

    ```python Python [expandable] theme={null}
    import asyncio

    from .hello_message_flow import build_hello_message_flow

    async def main():

        my_flow_definition = build_hello_message_flow()

    if __name__ == "__main__":
        asyncio.run(main())  
    ```
  </Step>

  <Step title="Compile the agentic workflow">
    Next, compile the agentic workflow. This step generates the agentic workflow model and deploys it to the engine. It returns a `CompiledFlow` instance, which you can use to start an agentic workflow run. Use `compile_deploy()` to compile and deploy the agentic workflow.

    ```python Python theme={null}
    compiled_flow = await my_flow_definition.compile_deploy()
    ```

    If you only want to generate the JSON model without deploying it, use `compile()` instead.
  </Step>

  <Step title="Start a run">
    Agentic workflows are asynchronous, and the engine communicates with your client using events. You can start a run using methods on the `CompiledFlow` object. Depending on your needs, you can choose from the following approaches:

    <AccordionGroup>
      <Accordion title="A simple client">
        The simplest way to run an agentic workflows is to invoke it and wait for the result. Use `CompiledFlow.invoke()` to start the agentic workflow. You can optionally pass input data.

        ```python Python theme={null}
        flow_run = await compiled_flow.invoke({"first_name": "John", "last_name": "Doe"})
        ```

        <Expandable title="optional parameters to invoke() function">
          <ParamField path="input_data" type="dict">
            Input data to pass to the agentic workflow.
          </ParamField>

          <ParamField path="on_flow_end_handler" type="callable">
            Callback executed when the agentic workflow completes successfully.
          </ParamField>

          <ParamField path="on_flow_error_handler" type="callable">
            Callback executed if the agentic workflow encounters an error.
          </ParamField>

          <ParamField path="debug" type="bool">
            Enables debug mode if set to `True`.
          </ParamField>
        </Expandable>

        To monitor the agentic workflow and retrieve the output:

        ```python Python [expandable] theme={null}
        from ibm_watsonx_orchestrate.experimental.flow_builder.flows.flow import FlowRunStatus

        # Wait for the flow to complete
        while flow_run.status != FlowRunStatus.COMPLETED and flow_run.status != FlowRunStatus.FAILED:
            await asyncio.sleep(1)  # sleep 1 second between checks
            
        print(f"Flow ended with status: {flow_run.status}")
        print(f"Flow output: {flow_run.output}" if flow_run.status == FlowRunStatus.COMPLETED else f"Flow error: {flow_run.error}")
        ```

        <Note>
          **Note:** Always use asynchronous waits to avoid blocking the event loop.
        </Note>
      </Accordion>

      <Accordion title="Using event handlers">
        You can provide custom event handlers to `invoke()` for more control over agentic workflow completion and error handling.

        ```python Python [expandable] theme={null}
        flow_run = None

        def on_flow_end(result):
            """
            Callback function to be called when the agentic workflow is completed.
            """
            print(f"Custom Handler: flow `{flow_run.name}` completed with result: {result}")

        def on_flow_error(error):
            """
            Callback function to be called when the agentic workflow fails.
            """
            print(f"Custom Handler: flow `{flow_run.name}` failed: {error}")

        async def main():

            my_flow_definition = build_hello_message_flow()
            compiled_flow = await my_flow_definition.compile_deploy()
            
            global flow_run
            flow_run = await compiled_flow.invoke({"first_name": "John", "last_name": "Doe"}, on_flow_end_handler=on_flow_end, on_flow_error_handler=on_flow_error)
        ```
      </Accordion>

      <Accordion title="Processing raw events">
        To handle all emitted events, use `invoke_events()`, which returns an asynchronous iterator yielding (`FlowEvent`, `FlowRun`) tuples.

        ```python Python [expandable] theme={null}
        from ibm_watsonx_orchestrate.experimental.flow_builder.types import FlowEventType

        async for event, run in compiled_flow.invoke_events({"first_name": "John", "last_name": "Doe"}):
            if not event:
                continue
            
            if event.kind == FlowEventType.ON_FLOW_END:
                on_flow_end(event.context.data.output)
                print(f"Flow `{run.name}` completed with result: {run.output}")
                break

            elif event.kind == FlowEventType.ON_FLOW_ERROR:
                print(f"Flow `{run.name}` failed with error: {run.error}")
                break 
        ```

        <Expandable title="optional parameters to invoke() function">
          <ParamField path="input_data" type="dict">
            Input data to pass to the agentic workflow.
          </ParamField>

          <ParamField path="filters" type="Sequence[Union[FlowEventType, TaskEventType]]">
            Callback executed when the agentic workflow completes successfully.
          </ParamField>

          <ParamField path="debug" type="bool">
            Callback executed if the agentic workflow encounters an error.
          </ParamField>
        </Expandable>

        You can listen for task and agentic workflow events, as defined by `FlowEventType` and  `TaskEventType` in `ibm_watsonx_orchestrate.experimental.flow_builder.types`.
      </Accordion>
    </AccordionGroup>
  </Step>
</Steps>