Event Loop Handling for Pipeline Tools

[Screenhandsaw]

Hi Team,

I wanted to open up this as a discussion before submitting an issue since I'm not sure this is intentional. I've been looking into how the async and event loop handling is done and came across this issue, which seems a bit unintuitive to me, so I just wanted to clarify things.

My setup is a simple Agent with a single tool, which I pass in as a PipelineTool. Now, Everything I'm doing is using the async versions, so I'm using AsyncPipeline and have defined run_async on all my components. I've also verified that all components report: __haystack_supports_async__ as True.

However, I wanted to write to an external event queue from inside of my tool and I came across issues with async/threading. I hunted this down to the following.

1. PipelineTool sets up a SuperComponent internally which has a run and run_async method defined:
   

   haystack/haystack/tools/pipeline_tool.py

   Line 197 in a8a6495

   component=SuperComponent(pipeline=pipeline, input_mapping=input_mapping, output_mapping=output_mapping),

2. However, inside of the ComponentTool class a component_invoker function is defined
   

   haystack/haystack/tools/component_tool.py

   Line 190 in a8a6495

   def component_invoker(**kwargs: Any) -> dict[str, Any]:

   
   This is then passed as the function call to the Tool class which is the base class of the ComponentTool.
   

   haystack/haystack/tools/component_tool.py

   Line 231 in a8a6495

   function=component_invoker,

3. I'm pretty sure this breaks the async-await and causes the tool to always execute off the main event loop - which may be intentional. However, this means that when I wrap an AsyncPipeline in a PipelineTool it will always execute from the synchronous SuperComponent's run method. This is easy to verify by adding a logger line in the _SuperComponent.run method: logger.info(f"Super Component Running sync pipeline {self.pipeline}"). Which when wrapping an AsyncPipeline returns:

haystack.core.super_component.super_component | INFO | Super Component Running sync pipeline <haystack.core.pipeline.async_pipeline.AsyncPipeline>

So my question is. Is this an intentional design choice or is this an oversight? Because to me it doesn't seem entirely clear that wrapping an AsyncPipeline with a PipelineTool would not be asynchronous.

Thanks

[apoorva-01]

Your read is right, and it's not specific to your setup. Wrapping an AsyncPipeline in a PipelineTool always runs it synchronously. Here's the chain, from the current main:

Tool is sync by contract. Tool.__post_init__ actually raises if you hand it a coroutine function, so there's no async tool body anywhere in the tool layer. ComponentTool builds that sync component_invoker, and it calls component.run(...) (component_tool.py) — never run_async. Then SuperComponent.run calls self.pipeline.run(...), the sync driver, even when the pipeline is an AsyncPipeline. SuperComponent.run_async exists but nothing in the tool path ever reaches it. That's the exact log line you saw.

And the async agent doesn't save you either: ToolInvoker.run_async invokes the sync tool via loop.run_in_executor(...). So your tool body runs on a threadpool worker thread, off the main event loop. That's why the queue write is fighting you — you're in a different thread with no handle on the app's running loop, so awaiting anything or grabbing get_running_loop() won't do what you want.

Two ways to push to your async queue from inside the tool:

• Capture the app's loop before you run the agent and hand it in, then from the tool do asyncio.run_coroutine_threadsafe(queue.put(item), main_loop). That's the thread-safe cross-thread handoff.
• Or use a plain queue.Queue (or janus if you want an async side) and drain it from a task on the main loop.

On "intentional or oversight" — it's a current limitation of the tool layer, not a bug in your async pipeline. Tools are sync by design right now, so an async component always gets driven through its sync .run. If you want first-class async tools it's worth opening an issue; the run_in_executor hop in ToolInvoker is the natural place it'd have to change.

[Screenhandsaw]

Great, thank you for your answer. Yes, in the end I did what you're suggesting - and which is also done in Hayhooks.

This makes e.g the chat completion functions a bit clunky, but here's the workaround:

async def run_chat_completion_async(self, model: str, messages: list[dict], body: dict) -> AsyncGenerator:
    async def stream() -> AsyncGenerator:
        queue = asyncio.Queue()
        loop = asyncio.get_running_loop()
        
        try:
            # set event loops + pipeline logic
        finally:
            # reset loop and queue
     
    return stream()

Then create helper functions used in the tools which can write to that current context stream and loop.