[Python] How to implement non-blocking async checks between turns in Pipecat Flows (0.0.18)?

I’m working with the latest version of Pipecat and Pipecat Flows (0.0.18), and I’m trying to implement a pattern where the bot can continue interacting with the user while an async background task runs, without blocking the conversation.

Use case (POC)

1. 
   When the bot starts, it should launch and forget an async background task (e.g., fetching a profile).
   
   
2. 
   At the same time, the bot greets the user.
   
   
3. 
   While the user and bot exchange turns, the background task may complete at any moment.
   
   
4. 
   If the task has completed by the time of the next bot turn, the bot should end the conversation (e.g., say goodbye).
   
   
5. 
   If the task is still running, the bot should continue the conversation and check again on the next turn.
   

My approach so far


I tried using pre_actions, but those only trigger on node entry (not per turn). To work around this, I created a simple “ping-pong” loop with two nodes:

• 
  check_profile node: launches the async fetch (if not already started), checks the state, and either routes to an end node or to the talk node.
  
  
• 
  talk node: handles ongoing conversation, then routes back to the check_profile node.
  

Here’s a simplified version of the relevant code:

TEST_DELAY = 1.0

def build_check(fm: FlowManager) -> NodeConfig:
async def _start_profile_fetch_action(*_args, **_kwargs):
logger = get_logger(fm)
logger.info("[Action:start_profile_fetch] Launching background fetch...")
prev: bool = fm.state.get(STATE_FETCH_STARTED) or False
profile = fm.state.get(STATE_PATIENT)

if not prev and not profile:
asyncio.create_task(
_fetchProfile(fm),
name="patient_profile_fetch"
)
fm.state[STATE_FETCH_STARTED] = True

async def _check_profile_handler_bound(_args: Dict[str, Any], *_unused):
logger = get_logger(fm)
profile = fm.state.get(STATE_PATIENT)

if profile:
logger.debug("[FOUND] %s", profile)
await fm.set_node_from_config(build_node_end(fm))
else:
logger.debug("[NOT FOUND PROFILE]")
await fm.set_node_from_config(build_talk(fm))

return {
"name": "check_profile",
"respond_immediately": False,
"task_messages": [],
"pre_actions": [
{"type": "function", "handler": _start_profile_fetch_action},
{"type": "function", "handler": _check_profile_handler_bound},
],
}

def build_talk(fm: FlowManager) -> NodeConfig:
async def _go_check(_args: Dict[str, Any], *_unused):
logger = get_logger(fm)
logger.debug("===============leaving talk node")
await fm.set_node_from_config(build_check(fm))

return {
"name": "talk",
"respond_immediately": True,
"task_messages": [
{"role": "system", "content": "Just say hello to the user and keep conversation"},
],
"post_actions": [
{"type": "function", "handler": _go_check},
],
}

async def _fetchProfile(fm: FlowManager):
logger = get_logger(fm)
try:
logger.info("[BG] Starting profile fetch (%.1fs)...", TEST_DELAY)
await asyncio.sleep(TEST_DELAY)
profile = {"name": "Alex Johnson", "age": 67}
fm.state[STATE_PATIENT] = profile
logger.info("[BG] Profile fetched -> %s", profile)
except Exception as e:
logger.exception("[BG] Fetch error: %s", e)

The problem

• 
  post_actions don’t seem to run after the bot finishes speaking, so my loop doesn’t always cycle back to the check_profile node as intended.
  
  
• 
  I want the system to always remain responsive (never waiting on the background task), and to make conversation decisions purely based on the current state.
  
  
• 
  Eventually, I’d like to support multiple async tasks updating a centralized state, where the conversation reacts accordingly.
  

Question


Is this the right approach to implement async background checks in Pipecat Flows? If not, what’s the recommended way to:

• 
  Launch background tasks without blocking
  
  
• 
  Repeatedly check state between turns
  
  
• 
  Route nodes dynamically based on whether results are ready
  

Continue reading...