Testing agentic workflow

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:

A simple client

The simplest way to run a 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

flow_run = await compiled_flow.invoke({"first_name": "John", "last_name": "Doe"})

Show optional parameters to invoke() function

Parameter	Type	Description
input_data	`dict`	Input data to pass to the agentic workflow.
on_flow_end_handler	`callable`	Callback executed when the agentic workflow completes successfully.
on_flow_error_handler	`callable`	Callback executed if the agentic workflow encounters an error.
debug	`bool`	Enables debug mode if set to `True`.

To monitor the agentic workflow and retrieve the output:

Python

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: Always use asynchronous waits to avoid blocking the event loop.

Using event handlers

You can provide custom event handlers to invoke() for more control over agentic workflow completion and error handling.

Python

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)

Processing raw events

To handle all emitted events, use invoke_events(), which returns an asynchronous iterator yielding (FlowEvent, FlowRun) tuples.

Python

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 

Show optional parameters to invoke() function

Parameter	Type	Description
input_data	`dict`	Input data to pass to the agentic workflow.
filters	`Sequence[Union[FlowEventType, TaskEventType]]`	Callback executed when the agentic workflow completes successfully.
debug	`bool`	Callback executed if the agentic workflow encounters an error.

You can listen for task and agentic workflow events, as defined by FlowEventType and TaskEventType in ibm_watsonx_orchestrate.experimental.flow_builder.types.

Release Notes

Get Started

Build

Analyze

watsonx Orchestrate Developer Edition

watsonx Orchestrate ADK MCP Server

Reference

Legal notices