Skip to main content
What this builds. The async twin of background-run: build and deploy a pipeline with await-able methods, then start it in the background and await the final result. You’ll end up with. A task_id you can log immediately, then the awaited result dict once the background run finishes. 
import asyncio

from vectorshift.pipeline import Pipeline


PIPELINE_NAME = "bg_run_async_example"


async def main():
    try:
        pipeline = await Pipeline.afetch(name=PIPELINE_NAME)
        print(f"Pipeline fetched: id={pipeline.id}, branch_id={pipeline.branch_id}")
    except Exception as e:
        print(f"Error fetching pipeline: {e}")
        pipeline = await Pipeline.anew(name=PIPELINE_NAME)
        print(f"Pipeline created: id={pipeline.id}, branch_id={pipeline.branch_id}")
    inp = pipeline.add(name="input_0", id="input_0").input(input_type="string")
    llm = pipeline.add(name="llm", id="llm").llm(
        provider="openai", model="gpt-4o", prompt=inp.text
    )
    pipeline.add(name="output_0", id="output_0").output(
        output_type="string", value=llm.response
    )
    await pipeline.asave(deploy=True)

    # Start in background
    handler = await pipeline.astart(
        inputs={"input_0": "Tell me a fun fact about oceans."}
    )
    print(f"Task ID: {handler.task_id}")

    # Check status
    status = await handler.arun_status()
    print(f"Status: {status['status']}")

    # Wait for completion
    result = await handler.aresult(poll_interval=2.0, timeout=60.0)
    print(f"Result: {result}")


asyncio.run(main())

​
Expected output

Pipeline created: id=..., branch_id=...
Task ID: ...
Status: running
Result: {'status': 'success', 'result': {'output_0': '...'}, ...}
Every blocking method on Pipeline and RunHandler has an a-prefixed async counterpart — afetch, anew, asave, astart, arun_status, aresult.

​
See also

Background runs

Same pattern in synchronous code.

Run pipeline (async)

await pipeline.arun(...) for fully-awaited execution.

Pipeline reference

Full method surface.