Run a managed workflow

• Package versions

  The code on this page was developed using the following requirements. We recommend using these versions or newer.

  qiskit-serverless~=0.38.0
  

A managed workflow job works like a batch job that you submit to an HPC scheduler. run() submits the job and returns a Job object immediately. Use the Job object for the tasks that squeue, the job's output file, and scancel cover on a cluster. The job keeps running after you close your notebook.

The examples on this page use a small workflow that reads one angle, theta, and fails if the angle is out of range. Upload it as described in Deploy your first managed workflow.

import math

from qiskit_serverless import get_arguments, save_result

arguments = get_arguments()
theta = arguments["theta"]
if not 0 <= theta <= math.pi:
    raise ValueError(f"theta must be between 0 and pi, got {theta}")

print(f"Computing the expectation value for theta={theta}", flush=True)
save_result({"theta": theta, "expectation_value": math.cos(theta)})

Start a job

Get the workflow by its title, and call run() with the arguments that the workflow reads:

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
my_workflow = client.load("my-workflow")

job = my_workflow.run(theta=0.5)
job_id = job.job_id

run() returns after the workflow queues the job. Save the job ID so that you can find the job from a new session.

• To choose which arguments a workflow accepts and check them before the job starts, see Define inputs and outputs.
• To run the job on a different compute size, pass function_size. See Choose a size for each run.

Follow a job

Check the status

Immediately after you call run(), the status is usually QUEUED:

job.status()

Output:

'QUEUED'

DONE, ERROR, and CANCELED are final, and job.in_terminal_state() returns True for them. A status loop that waits for another value, such as SUCCEEDED, never ends.

Wait for the job to end

To wait until the job ends, call wait_for_final_state(). It returns the final status, and with verbose=True it prints each status change. If you set timeout in seconds, wait_for_final_state() raises a TimeoutError when the time runs out before the job ends. The job keeps running:

job.wait_for_final_state(timeout=600, verbose=True)

Waiting for job 3f2a1c0e-6b1d-4c0e-9a51-2f0c4d1e7a90
0:00:00  QUEUED
0:02:10  RUNNING
0:03:05  RUNNING: WAITING_FOR_QPU
0:41:44  DONE
'DONE'

wait_for_final_state() checks the status after 1, 2, 4, 8, 16, and 32 seconds, and then every 60 seconds, so it sees the end of a short job within seconds and makes few requests for a long job. To check at a fixed interval instead, pass poll_interval in seconds.

Read the logs

Anything that the workflow writes to standard output or standard error goes to the job's logs. You can read them with job.logs() while the job runs, before it has a result:

print(job.logs())

Output:

Computing the expectation value for theta=0.5

• The workflow uploads new log lines about every 15 seconds, and one last time when the job ends or is canceled. Immediately after a job starts, job.logs() returns No logs yet.
• The workflow keeps the newest 50 MB of logs for each job, and adds a line at the top that says the logs were truncated.

To decide what your workflow logs, see Write logs. To show only the lines that match a regular expression, pass include to job.filtered_logs(). To hide lines, pass exclude.

Get the result

job.result() waits until the job ends, then returns the value that the workflow passed to save_result():

job.result()

Output:

{'theta': 0.5, 'expectation_value': 0.8775825618903728}

By default, job.result() checks the status every 30 seconds and waits with no time limit. Pass maxwait to limit the number of checks. For example, job.result(maxwait=40) waits about 20 minutes. To check one time without waiting, pass wait=False. If job.result() stops waiting before the job ends, it returns {}.

Check how long each stage took

job.events() lists every status and stage change of a job with its time, so you can see how long the job waited for compute and how long each stage took:

for event in reversed(job.events()):
    print(
        event.created,
        event.data.get("status") or event.data.get("sub_status"),
    )

Output:

2026-11-02T14:43:07.134377Z QUEUED
2026-11-02T14:43:24.474199Z PENDING
2026-11-02T14:46:07.309519Z RUNNING
2026-11-02T14:46:21.115038Z SUCCEEDED

The events use the stored status names. PENDING is INITIALIZING, and SUCCEEDED is DONE. For the QPU usage of a run, see Find the QPU jobs of a run.

Find the cause of an error

When the workflow raises an exception, the job ends with the ERROR status. For example, run the workflow with an angle that it rejects:

failed_job = my_workflow.run(theta=-1.0)

failed_job.status()

Output:

'ERROR'

The logs show the traceback:

print(failed_job.logs())

Output:

Traceback (most recent call last):
  File "/function_user_data/my_workflow.py", line 8, in <module>
    raise ValueError(f"theta must be between 0 and pi, got {theta}")
ValueError: theta must be between 0 and pi, got -1.0

job.result() raises a QiskitServerlessException:

from qiskit_serverless.exception import QiskitServerlessException

try:
    failed_job.result()
except QiskitServerlessException as error:
    print(error)

Output:

    raise ValueError(f"theta must be between 0 and pi, got {theta}")
ValueError: theta must be between 0 and pi, got -1.0

If the workflow reports errors with send_error(), read them with job.events(type="ERROR"), and read their messages with job.error_message(). For a job without reported errors, job.events(type="ERROR") returns an empty list, and job.error_message() returns {}.

for event in job.events(type="ERROR"):
    print(event.created, event.data.get("code"), event.data.get("message"))

Manage jobs

Cancel a job

Cancel a job with job.cancel(). The status changes to CANCELING while the workflow stops the job, and then to CANCELED:

job_to_cancel = my_workflow.run(theta=1.0)
job_to_cancel.cancel()

Output:

'Job has been stopped. No active runtime job ID associated with this serverless job ID.'

job.cancel() returns a message about the job and about its linked QPU jobs, which it also tries to cancel. For a running job, the workflow sends SIGTERM to your entrypoint. To save partial results, see Clean up when a job is canceled. If the job already ended, job.cancel() returns Job already in terminal state. and changes nothing.

Find a job from a new session

The Job object and the workflow exist only in your current Python session. After you restart a notebook kernel, pass the saved job ID to client.job(), or list your recent jobs with client.jobs():

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
job = client.job(job_id)
my_workflow = client.load("my-workflow")

client.jobs(function=my_workflow, status="ERROR", limit=5)

Output:

[<Job | e6b1ffa1-cd26-437d-bd53-b265dd4ba952>,
 <Job | 11308f2a-8d7a-40a0-a970-7fdeca2b171b>,
 <Job | af597785-3470-469a-9051-607e0185892e>,
 <Job | a9051931-3bc2-4b80-9ce6-c39d28ce260c>,
 <Job | e329e8e3-e955-4188-9bfd-8f3050ee2d39>]

client.jobs() returns your 10 most recent jobs, newest first. Filter it with function, status (from the status table), or created_after (an ISO 8601 timestamp), and page through it with limit and offset. To list the workflows that you uploaded, call client.list().

The Workloads table on IBM Quantum Platform also lists your jobs, with their status, compute profile, and run times.

Time limit

A managed workflow cancels a job that runs for more than 24 hours, and the job ends with the CANCELED status. During long runs, save checkpoints to file storage so that you do not lose work.

Troubleshoot jobs

Next steps