Cooking with Coroutines: A Hands-On Guide to Asynchronous Programming in Python

A hands-on guide to Python’s asynchronous framework that uses a cooking analogy to explain coroutines, the event loop, and concurrent I/O. Includes runnable examples for Jupyter and scripts, plus notes on using async patterns in AI workflows and reproducible multi-model pipelines.

LLM
asyncio
coroutines
AI workflows
Author

Mei-Chin Pang

Published

February 1, 2026

1 What is Asynchronous Programming?

If you have experience coding in Python, you are probably used to writing synchronous, blocking code like performing simple arithmetic calculations as follows:

1def add(
   x_param: float,
   y_param: float):
   print("Calculating summation now:")
   result = x_param + y_param
   print(f"Result inside function: {result}")
   print("Finish calculating summation.")
   return result

2def subtract(
   x_param: float,
   y_param: float):
   print("Calculating substraction now:")
   result = x_param - y_param
   print(f"Result inside function: {result}")
   print("Finish calculating substraction.")

   return result

z_total = add(12.1, 5.0)
print("-"*70)
z_total = subtract(20.8, 5.5)
1
Define the first function add() to perform the addition of two floating-point numbers.
2
Define the second function subtract() to perform the subtraction of two
floating-point numbers. 
Calculating summation now:
Result inside function: 17.1
Finish calculating summation.
----------------------------------------------------------------------
Calculating substraction now:
Result inside function: 15.3
Finish calculating substraction.

Here, you define a function starting with the def keyword, input parameters xparam and yparam, perform the calculations, and return the results. When you call the first function add(12.1, 5.0), the program executes the function and waits until the results are computed and returned before moving on to the next function. After that, the program calls the second function subtract(20.8, 5.5) and again waits until the results are computed and returned before moving on. This approach is known as synchronous programming, where tasks are performed one after another in a sequential manner.

While this type of computation is simple and intuitive, it becomes inefficient for I/O-bound workloads that spend most of their time waiting, for example, fetching data from remote servers, reading and writing files to disk, or waiting for responses from user inputs. If one task in the pipeline needs five minutes to wait for a response from a server, the next task will be suspended for five minutes until the previous task completes.

Unlike synchronous programming, asynchronous programming lets you write concurrent non-blocking code by allowing programs to continue executing other tasks while waiting for I/O-bound operations to complete. We can implement asynchronous frameworks in Python by using the asyncio library to create and manage asynchronous functions (also known as coroutines). This library was added in Python 3.4 and designed for situations where tasks spend much of their time waiting.

Figure 1: Comparison of a synchronous workflow and an asynchronous workflow (Image by Author)

1.1 Why is Asynchronous Programming Important?

Before we dive into the details of how to use asyncio in Python, let’s first understand why understanding asynchronous programming is important, especially to establish agentic workflows with multiple AI models using the Model Context Protocol (MCP). MCP is an open-source standard protocol that connects different AI models, such as ChatGPT and Claude, to external systems, including databases, APIs, and other services. This type of workflow often involves multiple I/O-bound operations, such as sending requests to different AI models, waiting for their responses, and processing the results.

Figure 1 illustrates the difference between a synchronous workflow and an asynchronous workflow in the context of a server-client communication framework. By relying solely on synchronous programming, the server would handle one request at a time, blocking other requests in the queue. This approach can lead to delays and inefficiencies, especially when multiple clients are trying to connect to the same server.

In contrast, asynchronous frameworks allow the server to handle multiple requests at the same time without blocking the entire application. If the network communication is slow due to latency issues, the relevant MCP servers or clients can perform other tasks instead of waiting in idle mode during this time. This implementation leads to improved responsiveness and better resource utilization, which allows MCP servers and clients to handle more requests efficiently.

2 Asyncio Explained with a Cooking Example

Asynchronous programming can be confusing and hard to grasp, so I will illustrate the concept with a simple cooking example from the kitchen.

Preparation of an Italian dinner with spaghetti carbonara, salad and mushroom soup (AI-generated Image)

Imagine you are preparing an Italian dinner for your family this weekend, and you would like to prepare three different dishes for them, namely spaghetti carbonara, salad, and mushroom soup. Each dish takes a different amount of time to prepare:

  • Spaghetti carbonara: 30 minutes
  • Salad: 10 minutes
  • Mushroom Soup: 20 minutes

2.1 Synchronous (Blocking) Approach

If you prepare these dishes synchronously (like traditional blocking I/O), you would:

  1. Start preparing spaghetti carbonara, which takes 30 minutes. During this time, you do nothing else but focus on the pasta.
  2. Once the pasta is done, you start preparing the salad, which takes another 10 minutes.
  3. Finally, when the pasta and salad are prepared, you start cooking the mushroom soup, which takes another 20 minutes.

We can simulate this synchronous cooking process with the following code:

import time

def synchronous_cooking():
   
   # Prepare each dish one by one
   print("Synchronous cooking:")
1   start_time = time.time()

   # -------------------------------------------------------------------------
   # (1) Cooking spaghetti carbonara
   print("Start preparing dish (1): " 
         +"spaghetti carbonara (estimated 30 minutes)")
2   time.sleep(30)
   print("spaghetti carbonara is ready!")
   print("-"*70)

   # -------------------------------------------------------------------------
   # (2) Preparing salad
   print("Start preparing dish (2): salad (estimated 10 minutes)")
   time.sleep(10)
   print("Salad is ready!")
   print("-"*70)

   # -------------------------------------------------------------------------
   # (3) Cooking mushroom soup
   print("Start preparing dish (3): mushroom soup (estimated 20 minutes)")
   time.sleep(20)
   print("Soup is ready!")
   print("-"*70)
    
3   end_time = time.time()
   print(
      "Total time taken (synchronous): " 
      + f"{end_time - start_time:.2f} minutes\n")

prep_italian_dinner = synchronous_cooking()
1
We start the timer before preparing the first dish.
2
We simulate the time taken to prepare each dish using time.sleep() to mimic the cooking time. Here, we scale the simulation time to represent real cooking time, for example, 30 seconds in our simulation correspond to 30 minutes of the real cooking time.
3
We stop the timer after all dishes are prepared and calculate the total time taken. 
Synchronous cooking:
Start preparing dish (1): spaghetti carbonara (estimated 30 minutes)
spaghetti carbonara is ready!
----------------------------------------------------------------------
Start preparing dish (2): salad (estimated 10 minutes)
Salad is ready!
----------------------------------------------------------------------
Start preparing dish (3): mushroom soup (estimated 20 minutes)
Soup is ready!
----------------------------------------------------------------------
Total time taken (synchronous): 60.00 minutes
Figure 2: Comparison of a synchronous and an asynchronous dinner preparation period (Image by Author)

As shown by Figure 2, because the dishes are prepared sequentially, the total time spent preparing all three dishes is 30 + 10 + 20 = 60 minutes. While you are boiling the water for the pasta, the other tasks are waiting. You could have been chopping vegetables for the salad while the pasta water was boiling. Instead, you are blocked from doing anything else until the pasta is ready.

2.2 Asynchronous (Non-Blocking) Approach

Now suppose that your family is hungry and they want their Italian dinner faster. You can speed up the cooking process by using an asynchronous approach. In this case, instead of preparing one dish at a time, you can multitask by starting one dish, letting it cook, and then moving on to the next task while waiting for the first dish to finish.

Here’s how it would work:

  1. Start cooking the pasta and let it boil for 8 minutes.
  2. While the pasta is boiling, you start chopping the vegetables and preparing the salad.
  3. After finishing the salad, you start preparing the soup. While the soup is cooking, you check on the pasta and finish it up with the carbonara sauce and ingredients.

Let’s simulate this asynchronous cooking process with the following code:

import asyncio
import time

# Create a coroutine for each dish using the async def and await keywords
1async def prepare_pasta():
   print("Start preparing dish (1): " 
         +"spaghetti carbonara (estimated 30 minutes)")
2   await asyncio.sleep(30)
   print("Pasta is ready!")

async def prepare_salad():
   print("Start preparing dish (2): salad (estimated 10 minutes)")
   await asyncio.sleep(10)
   print("Salad is ready!")

async def prepare_soup():
   print("Start preparing dish (3): mushroom soup (estimated 20 minutes)")
   await asyncio.sleep(20)
   print("Soup is ready!")


async def asynchronous_cooking():
   
   print("Asynchronous Cooking:")
   start_time = time.time()

   # Gather all coroutines and run them concurrently
3   results = await asyncio.gather(
      prepare_pasta(),
      prepare_salad(),
      prepare_soup())

   end_time = time.time()
   print(
      "Total time taken for asynchronous cooking: " 
      + f"{end_time - start_time:.2f} minutes\n")

# Run the asynchronous cooking simulation
# When executing in a Jupyter notebook, use: await asynchronous_cooking()
4await asynchronous_cooking()
1
We define three asynchronous functions (coroutines) using the async def syntax for each dish: prepare_pasta(), prepare_salad() and prepare_soup().
2
We simulate the time taken to prepare each dish using await asyncio.sleep() to mimic the cooking time without blocking the event loop.
3
We use await asyncio.gather() to gather and run all three coroutines concurrently.
4
We execute the asynchronous cooking simulation using await in a Jupyter notebook or asyncio.run() in a Python script. 
Asynchronous Cooking:
Start preparing dish (1): spaghetti carbonara (estimated 30 minutes)
Start preparing dish (2): salad (estimated 10 minutes)
Start preparing dish (3): mushroom soup (estimated 20 minutes)
Salad is ready!
Soup is ready!
Pasta is ready!
Total time taken for asynchronous cooking: 30.01 minutes

In this scenario, the total time to prepare all three dishes is reduced to approximately 30 minutes, as you are able to prepare multiple dishes concurrently (see Figure 2). Your family will get their dinner in 30 minutes instead of 60 minutes, and you use your time more efficiently.

2.3 How does Asyncio Work in Python?

The fundamentals of asynchronous programming in Python revolve around the event loop, which is responsible for managing and scheduling the execution of asynchronous tasks (coroutines). The event loop continuously checks for tasks that are ready to run and executes them. When a task encounters an await statement, the control is given back to the event loop, allowing other tasks to be executed while waiting for the awaited operation to complete.

When we use the syntax async def, we are creating a coroutine, which is a special type of function that can be suspended and resumed. The await keyword is used to pause the execution of the coroutine until the awaited task is complete, allowing other tasks to run in the meantime. In our example, when we define await asyncio.sleep(30), the coroutine would then pause for 30 seconds, allowing other coroutines, such as async def prepare_salad() and async def prepare_soup() to run during that time. Therefore, when you look at the output of the asynchronous cooking simulation, you will see that the tasks are not executed sequentially, but rather concurrently. The salad and soup will be prepared while the pasta is being cooked.

2.4 How to Execute a Coroutine?

A coroutine cannot be called directly like a regular function. Let’s compare the difference between a synchronous function (also known as a subroutine) and an asynchronous function (coroutine) by simulating the preparation of a salad. To create a standard subroutine, we would use the def keyword, while for the coroutine, we use the async def keywords.

import time

1def prepare_salad():
   print("Start preparing salad with a subroutine:")
   time.sleep(10)  # Simulate salad preparation time
   print("Salad is ready!")

2prepare_salad()
1
Define a standard subroutine prepare_salad using the def keyword.
2
Call the subroutine directly without any issues. 
Start preparing salad with a subroutine:
Salad is ready!

Let’s now create the coroutine for preparing our salad.

import asyncio

1async def prepare_salad():
   print("Start preparing salad with a coroutine:")
2   await asyncio.sleep(10)
   print("Salad is ready!")

3await prepare_salad()
1
Define the coroutine prepare_salad using the async def keyword.
2
Use await asyncio.sleep(10) to simulate the salad preparation time without blocking the event loop.
3
Execute the coroutine using the await keyword in a Jupyter notebook. (In a Python script, you would use asyncio.run(prepare_salad()) instead.) 
Start preparing salad with a coroutine:
Salad is ready!
ImportantExecute a Coroutine in a Notebook vs in a Python Script
  • While the standard subroutine prepare_salad can be called directly, the coroutine cannot be called directly and will raise a runtime error (coroutine 'prepare_salad' was never awaited) if you try to do so.
  • To run the coroutine in an interactive environment (i.e., Jupyter notebook), you can use the await keyword before calling the coroutine.
  • In a Python script, you would typically use asyncio.run() to initiate the event loop and run the coroutine. It should only be called once in a program, typically in the if __name__ == "__main__": block, as it creates and manages the entire event loop.

2.5 How to Define Coroutines with Input Arguments and Return Values?

Now, let’s modify the coroutine to accept an argument for the salad ingredients and return a message indicating that the salad is ready with the specified ingredients. A coroutine can return a value just like a regular function, and you can capture the return value by assigning to a variable when you await the coroutine.

import asyncio

1async def prepare_salad(ingredients=None):
   print("Start preparing salad with:", ingredients)
   await asyncio.sleep(10)  
   salad_ingredients = f"Salad is ready with {ingredients}"
2   return salad_ingredients

3salad_prep_status = await prepare_salad("lettuce, tomatoes, cucumbers")
print(f"Salad preparation status: {salad_prep_status}")
1
Define the coroutine prepare_salad to accept an optional argument ingredients.
2
Use the return statement to return a message indicating that the salad is ready with the specified ingredients.
3
Execute the coroutine with an input argument and capture the return value by assigning it to the variable salad_prep_status. 
Start preparing salad with: lettuce, tomatoes, cucumbers
Salad preparation status: Salad is ready with lettuce, tomatoes, cucumbers

2.6 How to Define and Capture Multiple Coroutines Concurrently?

Now, let’s extend the previous example by defining another coroutine within the prepare_pasta_dinner coroutine. This coroutine will call the prepare_salad() coroutine and capture its return value. Finally, we will execute the prepare_pasta_dinner() coroutine and capture its return value as well.

In the earlier examples, we have only used await to pause the execution of a coroutine for a specified duration without blocking the event loop. However, as shown in the following example, you can also use await within a coroutine to call another coroutine and capture its return value. This implementation allows you to build a complex asynchronous workflow by composing multiple coroutines together.

import asyncio

1async def prepare_salad(salad_ingredients=None):
   print("Start preparing salad with:", salad_ingredients)
   await asyncio.sleep(10)  
   salad_ingredients = f"salad is ready with {salad_ingredients}"
2   return salad_ingredients

3async def prepare_pasta_dinner(pasta_ingredients=None):
   print("Start preparing salad coroutine within a pasta dinner coroutine.")
   await asyncio.sleep(30)

   salad_prep_status = await prepare_salad(
4      "lettuce, tomatoes, cucumbers")
   print(f"Salad preparation status: {salad_prep_status}")

   pasta_salad_ingredients = (
      f"Pasta is ready with {pasta_ingredients} \n"
      + f"and {salad_prep_status}")
5   return pasta_salad_ingredients

pasta_salad_prep_status = await prepare_pasta_dinner(
6   "spaghetti, eggs, pancetta")
print("Pasta and salad preparation status:")
print(pasta_salad_prep_status)
1
Define the first coroutine prepare_salad to accept an optional argument salad_ingredients.
2
Use the return statement to return results from the first coroutine.
3
Define the second coroutine prepare_pasta_dinner to accept an optional argument pasta_ingredients.
4
Within the second coroutine, use await to call the first coroutine prepare_salad() and capture its return value in the variable salad_prep_status.
5
Use the return statement to return results from the second coroutine.
6
Execute the second coroutine with an input argument and capture the return value by assigning it to the variable pasta_salad_prep_status. 
Start preparing salad coroutine within a pasta dinner coroutine.
Start preparing salad with: lettuce, tomatoes, cucumbers
Salad preparation status: salad is ready with lettuce, tomatoes, cucumbers
Pasta and salad preparation status:
Pasta is ready with spaghetti, eggs, pancetta 
and salad is ready with lettuce, tomatoes, cucumbers

2.7 How to Create and Run Tasks?

We have used await to call coroutines sequentially within another coroutine previously. However, sometimes you may want to run multiple coroutines concurrently without waiting for each one to finish before starting the next. In such cases, you can create tasks using asyncio.create_task() and then use await asyncio.gather() to wait for all tasks to complete. Let’s modify the asynchronous cooking example to demonstrate this approach.

import asyncio
import time

async def prepare_pasta(pasta_ingredients=None):
   print("Start preparing dish (1): " 
         +"spaghetti carbonara (estimated 30 minutes)")
   await asyncio.sleep(30)
   pasta_ingredients = f"Pasta is ready with {pasta_ingredients}"
   print("Pasta is ready!")
   return pasta_ingredients


async def prepare_salad(salad_ingredients=None):
   print("Start preparing dish (2): salad (estimated 10 minutes)")
   await asyncio.sleep(10)
   salad_ingredients = f"Salad is ready with {salad_ingredients}"
   print("Salad is ready!")
   return salad_ingredients

async def prepare_soup():
   print("Start preparing dish (3): mushroom soup (estimated 20 minutes)")
   await asyncio.sleep(20)
   print("Soup is ready!")

async def asynchronous_cooking_with_tasks():
   
   print("Asynchronous Cooking:")
   start_time = time.time()
   
   prep_spaghetti_task = asyncio.create_task(
1      prepare_pasta(pasta_ingredients="spaghetti, eggs, pancetta"))

   prep_salad_task = asyncio.create_task(
      prepare_salad(salad_ingredients="lettuce, tomatoes, cucumbers"))
   
2   prep_soup_task = asyncio.create_task(prepare_soup())

   all_tasks_results = await asyncio.gather(
      prep_spaghetti_task,
      prep_salad_task,
3      prep_soup_task)

   print("Tasks execution results:")
   for index, result in enumerate(all_tasks_results):
      print(index, result)
   print("-"*70)

   end_time = time.time()
   print(
      "Total time taken for asynchronous cooking: " 
      + f"{end_time - start_time:.2f} minutes\n")

await asynchronous_cooking_with_tasks()
1
Create the first and second tasks for the prepare_pasta coroutine and
prepare_salad coroutine using asyncio.create_task() and pass the required input argument.
2
Create the third task for the prepare_soup coroutine without any input argument.
3
Use await asyncio.gather() to wait for all tasks to complete and capture their return values in the all_tasks_results list. 
Asynchronous Cooking:
Start preparing dish (1): spaghetti carbonara (estimated 30 minutes)
Start preparing dish (2): salad (estimated 10 minutes)
Start preparing dish (3): mushroom soup (estimated 20 minutes)
Salad is ready!
Soup is ready!
Pasta is ready!
Tasks execution results:
0 Pasta is ready with spaghetti, eggs, pancetta
1 Salad is ready with lettuce, tomatoes, cucumbers
2 None
----------------------------------------------------------------------
Total time taken for asynchronous cooking: 30.01 minutes

When we print the results of the tasks’ execution, we can see that the return values from the prepare_pasta and prepare_salad coroutines are captured in the results list, while the prepare_soup coroutine does not return any value, hence it is represented as None in the results list. Although the dish salad is completed first, followed by the soup and pasta, the task execution results are sorted and returned in the order the tasks were created.

2.8 How to Check Tasks Completion Status?

If you create tasks for your coroutines, you can check the completion status of each task using the done() method of the task object. This method returns True if the task has completed, and False otherwise. Let’s modify the previous example to include a check for the completion status of each task.

We can also access the results of each task individually using the result() method of the task object after completing all tasks.

import asyncio
import time

async def prepare_pasta(pasta_ingredients=None):
   print("Start preparing dish (1): " 
         +"spaghetti carbonara (estimated 30 minutes)")
   await asyncio.sleep(30)
   pasta_ingredients = f"Pasta is ready with {pasta_ingredients}"
   print("Pasta is ready!")
   return pasta_ingredients

async def prepare_salad(salad_ingredients=None):
   print("Start preparing dish (2): salad (estimated 10 minutes)")
   await asyncio.sleep(10)
   salad_ingredients = f"Salad is ready with {salad_ingredients}"
   print("Salad is ready!")
   return salad_ingredients

async def prepare_soup():
   print("Start preparing dish (3): mushroom soup (estimated 20 minutes)")
   await asyncio.sleep(20)
   print("Soup is ready!")


async def asynchronous_cooking_with_tasks():
   
   print("Asynchronous Cooking:")
   start_time = time.time()
   
   # Task (1): Spaghetti -----------------------------------------------------
   # create task for prepare_pasta coroutine including the input argument
   prep_spaghetti_task = asyncio.create_task(
      prepare_pasta(pasta_ingredients="spaghetti, eggs, pancetta"))


1   if prep_spaghetti_task.done():
      print("Pasta task is completed.")
   else:
      print("Pasta task is still being prepared.")

   # Task (2): Salad ---------------------------------------------------------
   # create task for prepare_salad coroutine including the input argument
   prep_salad_task = asyncio.create_task(
      prepare_salad(salad_ingredients="lettuce, tomatoes, cucumbers"))

   # Check if the salad task is done
   if prep_salad_task.done():
      print("Salad task is completed.")
   else:
      print("Salad task is still being prepared.")
   
   # Task (3): Soup ----------------------------------------------------------
   # create task for prepare_soup coroutine without input argument
   prep_soup_task = asyncio.create_task(prepare_soup())

   # Check if the soup task is done
   if prep_soup_task.done():
      print("Soup task is completed.")
   else:
      print("Soup task is still being prepared.")


   # Gather all tasks and wait for their completion --------------------------
   results = await asyncio.gather(
      prep_spaghetti_task,
      prep_salad_task,
2      prep_soup_task)
   
   # Access the results of each task individually
3   print(f"Pasta result: {prep_spaghetti_task.result()}")
   print(f"Salad result: {prep_salad_task.result()}")
   print(f"Soup result: {prep_soup_task.result()}")

   end_time = time.time()
   print(
      "Total time taken for asynchronous cooking: " 
      + f"{end_time - start_time:.2f} minutes\n")

await asynchronous_cooking_with_tasks()
1
Use the done() method to check if each task has completed before gathering the results.
2
Use await asyncio.gather() to wait for all tasks to complete.
3
Use the result() method to access the results of each task individually. 
Asynchronous Cooking:
Pasta task is still being prepared.
Salad task is still being prepared.
Soup task is still being prepared.
Start preparing dish (1): spaghetti carbonara (estimated 30 minutes)
Start preparing dish (2): salad (estimated 10 minutes)
Start preparing dish (3): mushroom soup (estimated 20 minutes)
Salad is ready!
Soup is ready!
Pasta is ready!
Pasta result: Pasta is ready with spaghetti, eggs, pancetta
Salad result: Salad is ready with lettuce, tomatoes, cucumbers
Soup result: None
Total time taken for asynchronous cooking: 30.01 minutes

2.9 How to Handle Exceptions in Tasks?

When working with asynchronous tasks, it is essential to handle exceptions that may occur during their execution. If a task raises an exception, it will propagate to the point where you await the task, and you can catch it using a try-except block. Let’s extend the prior examples to catch and handle exceptions that may arise during the execution of each task.

import asyncio
import time

async def prepare_pasta(pasta_ingredients=None):
   try:
      print("Start preparing dish (1): spaghetti carbonara")
      await asyncio.sleep(30)

      if len(pasta_ingredients) <= 4:
1         raise ValueError("Ran out of pasta!")
      return f"Pasta is ready with {pasta_ingredients}\n"

   except ValueError as e:
      print(f"Error while preparing pasta: {e}")

async def prepare_salad(salad_ingredients):
   try:
      print("Start preparing dish (2): Salad")

2      if not salad_ingredients:
         raise ValueError("No ingredients provided for the salad!")
      
      await asyncio.sleep(10)
      salad_ingredients = f"Salad is ready with {salad_ingredients}\n"
      print("Salad is ready!")
      return salad_ingredients

   except Exception as e:
      print(f"Error while preparing salad: {e}")

async def prepare_soup():
3   try:
      print("Start preparing dish (3): Mushroom Soup")
      await asyncio.sleep(20)
      return "Soup is ready!"
   except Exception as e:
      print(f"Error while preparing soup: {e}")


async def asynchronous_cooking_with_tasks():
   
   print("Asynchronous Cooking:")
   start_time = time.time()
   
   # Task (1): Spaghetti -----------------------------------------------------
   # create task for prepare_pasta coroutine including the input argument
   prep_spaghetti_task = asyncio.create_task(
4      prepare_pasta(pasta_ingredients=["spaghetti, eggs, pancetta"]))


   # Task (2): Salad ---------------------------------------------------------
   # create task for prepare_salad coroutine including the input argument
   prep_salad_task = asyncio.create_task(
5      prepare_salad(salad_ingredients=""))

   # Task (3): Soup ----------------------------------------------------------
   # create task for prepare_soup coroutine without input argument
   prep_soup_task = asyncio.create_task(prepare_soup())


   # Gather all tasks and wait for their completion --------------------------
   results = await asyncio.gather(
      prep_spaghetti_task,
      prep_salad_task,
      prep_soup_task)

   # Access the results of each task individually
6   print(f"Pasta result: {prep_spaghetti_task.result()}")
   print(f"Salad result: {prep_salad_task.result()}")
7   print(f"Soup result: {prep_soup_task.result()}")

   end_time = time.time()
   print(
      "Total time taken for asynchronous cooking: " 
      + f"{end_time - start_time:.2f} minutes\n")

await asynchronous_cooking_with_tasks()
1
In the prepare_pasta coroutine, we simulate an exception by raising a ValueError if the length of the pasta_ingredients is less than or equal to 4. We catch this exception and print an error message.
2
In the prepare_salad coroutine, we raise a ValueError if no ingredients are provided for the salad. We catch any exception and print an error message.
3
In the prepare_soup coroutine, we wrap the entire function in a try-except block to catch any exceptions that may occur during its execution.
4
When creating the pasta task, we pass a list with a length less than or equal to 4 to trigger the exception.
5
When creating the salad task, we pass an empty string to simulate the exception for missing ingredients.
6
When accessing the results of the pasta and salad tasks, we handle the case where the tasks may have raised exceptions and returned None.
7
When accessing the result of the soup task, we expect it to complete successfully and return the string “Soup is ready!”. 
Asynchronous Cooking:
Start preparing dish (1): spaghetti carbonara
Start preparing dish (2): Salad
Error while preparing salad: No ingredients provided for the salad!
Start preparing dish (3): Mushroom Soup
Error while preparing pasta: Ran out of pasta!
Pasta result: None
Salad result: None
Soup result: Soup is ready!
Total time taken for asynchronous cooking: 30.01 minutes

If you run the above code, you will see that the exceptions raised during the execution of the tasks are caught and handled gracefully, allowing the program to continue running without crashing. The error messages are printed to the console, and the results of the tasks that completed successfully are also displayed.

3 Conclusion

Asynchronous programming provides a framework for writing concurrent non-blocking code without blocking the execution of other tasks in the queue. By using the async and await keywords from the Python asyncio library, you can now define and manage asynchronous functions (coroutines) that can be paused and resumed while allowing other tasks to run during the waiting time. As a result, we can handle multiple tasks and connections concurrently without blocking the entire application to improve the responsiveness with better resource utilization.

In this article, we explored the concept of asynchronous programming using a cooking example, demonstrating how to define coroutines, execute them concurrently, create tasks to wrap and gather different coroutines, check their completion status, and handle exceptions. With this understanding, I hope you will find it easier to implement asynchronous programming in your own Python applications, especially when building efficient and responsive agentic workflows with MCP servers and clients. Happy cooking, and happy coding!

4 References

  1. Brownlee, J., 2022. Python Asyncio Jump-Start: Asynchronous Programming And Non-Blocking I/O With Coroutines, Python Concurrency Jump-Start Series. SuperFastPython.com.