6.4 异步编程介绍#

在第 6.3 节中，我们围绕 MCP 的基本概念与通信模型进行了介绍，可以看，MCP Client 与 MCP Server 之间都需要进行大量的 I/O 交互。这种场景下，如果继续沿用前面章节中惯用的同步编程方式，就会出现“等待一次 I/O 完成才能继续推进下一步”的问题，进而导致整体吞吐被拉低。

I/O 交互：指程序的主要耗时来自等待外部数据或外部设备完成操作，而不是来自 CPU 本身的计算耗时。例如请求第三方服务的等待时间。

更进一步，在第 6.5 节中我们将基于 langchain-mcp-adapters 与 mcp.server.fastmcp 实现一个完整的 MCP 客户端与自定义 MCP Server。其中，初始化 MCP 客户端、拉取工具列表、调用远端工具等操作在 SDK 层都被设计为异步接口；自定义 MCP Server 在处理并发请求时也需要借助异步任务来调度多个工具调用。因此，理解异步编程就成了理解与实现 MCP 的前置条件。当然，如果你翻阅大模型 SDK 的相关实现，里面也还有大量的异步操作。

在本节内容中，我们将从同步编程在 I/O 场景下的局限出发，依次介绍协程与 async/await 的基本概念、asyncio 的事件循环运行机制、异步任务的并发执行方式等的常见用法。所有内容都围绕一个目标展开，为下一节实现 MCP 服务提供基础。

6.4.1 同步编程的局限性#

Python 中最常见的编程模式就是同步编程，这也是我们日常90%情况下使用的模式。一个同步函数在被调用时会按照代码顺序依次执行每一行语句，直到函数返回或抛出异常为止。下面，我们通过一个简单的示例来形象地看一下同步编程在 I/O 密集场景下的局限性。

某一天下午你很开心，觉得去逛街吃点好吃的。于是你来到了美食广场，打算买奶茶、买手抓饼、买烤面筋。于是，你接下来先排队买奶茶花了8分钟，然后买手抓饼花了7分钟，最后买面筋又花了5分钟，总共耗时20分钟。

接着，我们通过程序来模拟上述整个过程。首先，定义3个 http 服务naicha()、shouzhuabing()和 kaomianjing() 来分别模拟3家店铺，且为了验证我们将分钟改为秒。例如，对于naicha()这个服务来说，其定义如下所示

1 @store.get("/naicha")
2 def naicha():
3     start = datetime.now()
4     print(f"开始做奶茶，预计等待8秒钟：{start}")
5     sleep(8) # 模拟等待过程
6     print(f"奶茶制作结束：{datetime.now()}")
7     return {"message": "naicha finished",
8             "start_time":start,
9             "end_time": datetime.now()}

完整代码见 C02_http_service.py 文件。

本节所有示例代码可点击此处 async_asyncio.zip 下载。

这样，我们在启动改服务以后就可以模拟整个制作耗时过程，示例代码如下，完整代码见 C03_sync_async.py：

 1 def customer_sync(store):
 2     print(f"去 {store} 店点餐 | {datetime.now()}")
 3     response = requests.get(f"http://127.0.0.1:5000/{store}")
 4     print(response.json())
 5 
 6 if __name__ == '__main__':
 7     now = datetime.now()
 8     stores = ['naicha', 'shouzhuabing', 'kaomianjing']
 9     for store in stores:
10         customer_sync(store)
11     print(f"一共耗时：{datetime.now() - now}s")

注意，这里需要先启动 C02_http_service.py 对应的服务，然后再运行 C03_sync_async.py 部分。

在上述代码中，第3行便是你去对应店铺中下单的模拟过程，对应理解就是 Agent 通过 MCP 去请求外部服务的过程。第8~10行则是依次访问3个服务（去每家店点餐）的过程。

上述代码运行结束以后便会看到类似如下结果：

去 naicha 店点餐 | 2026-09-16 21:01:41.874305
{'end_time': 'Wed, 16 Sep 2026 21:01:49 GMT', 'message': 'naicha finished', 'start_time': 'Wed, 16 Sep 2026 21:01:41 GMT'}
去 shouzhuabing 店点餐 | 2026-09-16 21:01:49.887027
{'end_time': 'Wed, 16 Sep 2026 21:01:54 GMT', 'message': 'shouzhuabing finished', 'start_time': 'Wed, 16 Sep 2026 21:01:49 GMT'}
去 kaomianjing 店点餐 | 2026-09-16 21:01:54.897890
{'end_time': 'Wed, 16 Sep 2026 21:02:01 GMT', 'message': 'kaomianjing finished', 'start_time': 'Wed, 16 Sep 2026 21:01:54 GMT'}
一共耗时：0:00:20.031385s

从上述结果可以看出，main 函数连续发起 3 次这样的调用，按照顺序执行所有请求，最终耗时约为 20 秒。这个结果在同步编程下完全符合直觉，因为每一次调用都要等待上一次完成才能开始下一次。

可以看到，这种执行模式在 CPU 密集型任务中是合理的，因为计算资源本身需要被顺序占用。但在 I/O 密集型任务中，同步编程会暴露出两个明显的问题：

首先，整体时延被 I/O 等待时间主导。在 customer_sync 这类模拟代码中，函数真正消耗 CPU 的时间几乎为零，绝大部分时间都花在等待外部系统返回结果上。如果一次任务包含 3 次、5 次甚至更多次 I/O 调用，那么总时延会随着调用次数线性增长。

其次，并发能力无法被自然表达。在同步编程中，开发者只能借助多线程、多进程等机制手动管理并发，但这又会引入锁、竞态条件、状态共享等一系列额外复杂度。对于“等待一次 I/O 之后再继续”这种简单场景，同步编程提供不了足够的表达力。

因此，对于 MCP 这类以 I/O 交互为核心的协议，最自然的选择就是切换到异步编程范式上。异步编程的核心思想可以概括为：当一个任务在等待 I/O 完成时，让程序有机会切换到其它任务，从而避免把时间浪费在“空等”上。

例如在很多网页上， 一些功能按钮你点击以后马上开始转圈，然后大约等30秒才出结果，且在这30秒内你就只能傻等，这就是同步的体验过程；而有些按钮你点击后立马给你反馈诸如“执行成功”之类的提示，但是你点击查看结果却提示“正在计算中”，但在结果计算出来以前你还能够去干其它事情，这就是异步的体验过程。

6.4.2 异步版本的点餐过程#

在上面我们使用了同步方式依次买奶茶、买手抓饼与买烤面筋，整个过程耗时约 20 秒。这一结果在同步编程下是不可避免的，因为每一次 HTTP 请求都要等待上一次返回之后才能开始。下面，我们把同一段逻辑改写为异步版本，观察整体耗时是否能够明显缩短，完整代码见 C04_sync_async.py。

 1 async def customer_async(store):
 2     client = httpx.AsyncClient(timeout=20.0)
 3     print(f"去 {store} 店点餐 | {datetime.now()}")
 4     response = await client.get(f"http://localhost:5000/{store}")
 5     await client.aclose()
 6 
 7 async def main(stores):
 8     await asyncio.gather(
 9         customer_async(stores[0]),
10         customer_async(stores[1]),
11         customer_async(stores[2]))
12 
13 if __name__ == '__main__':
14     now = datetime.now()
15     stores = ['naicha', 'shouzhuabing', 'kaomianjing']
16     now = datetime.now()
17     asyncio.run(main(stores))
18     print(f"一共耗时：{datetime.now() - now}s")

注意，这里仍旧需要先启动 C02_http_service.py 对应的服务，然后再运行 C04_sync_async.py 部分。

对比第6.4.1节中的同步版本，这段代码主要引入了 3 个新元素。第一，HTTP 客户端从 requests 切换为支持异步语义的 httpx.AsyncClient。httpx 提供了与 requests 相近的 API，但 AsyncClient 不会在每次请求时阻塞整个线程，从而为后续并发执行提供了基础。第二，新增加的 customer_async 是一个协程函数，它使用 async def 定义，并在 client.get(...) 之前使用 await，表示这里会等待 I/O 完成但不会阻塞其它协程。第三，主流程通过 asyncio.gather 把 3 次 customer_async 调用同时交给事件循环，并使用 asyncio.run 启动整个异步流程。asyncio.gather 在语义上相当于“同时去 3 家店下单，谁先做完就先返回”，3 个调用在等待 I/O 期间是相互重叠的。

上述代码运行结束以后会输出类似如下结果

去 naicha 店点餐 | 2026-09-16 21:02:50.040000
去 shouzhuabing 店点餐 | 2026-09-16 21:02:50.041000
去 kaomianjing 店点餐 | 2026-09-16 21:02:50.042000
{'end_time': 'Wed, 16 Sep 2026 21:02:55 GMT', 'message': 'kaomianjing finished', 'start_time': 'Wed, 16 Sep 2026 21:02:50 GMT'}
{'end_time': 'Wed, 16 Sep 2026 21:02:57 GMT', 'message': 'shouzhuabing finished', 'start_time': 'Wed, 16 Sep 2026 21:02:50 GMT'}
{'end_time': 'Wed, 16 Sep 2026 21:02:58 GMT', 'message': 'naicha finished', 'start_time': 'Wed, 16 Sep 2026 21:02:50 GMT'}
一共耗时：0:00:08.040000s

从结果中可以清楚地看到异步部分的总耗时缩短为 8 秒——这恰好是 3 次请求中耗时最长的一次（naicha 等待 8 秒），且 3 次请求几乎在同一时刻开始。换言之，异步版本并没有减少每家店的实际制作时间，而是让原本顺序的等待变成了并发的等待。这正是第 6.4.1 节末尾所描述的“当一个任务在等待 I/O 完成时，让程序有机会切换到其它任务”的具体体现。

对比同步与异步之间的差异，我们可以通过如图6-5所示的过程进行表示。

根据图6-5可知，同步方式是一个串行执行的过程，而异步方式则是一个并行执行（协程并行）的过程。

6.4.3 协程与 async/await 基本概念#

异步版本之所以能够并发地发起多次 HTTP 请求，最直接的语法基础就是 async def 与 await。理解这两个关键字，是理解后续事件循环与并发接口的前提。

在 Python 中，使用 async def 定义的函数被称为协程函数。协程函数与普通函数的一个核心区别是：调用协程函数并不会立即执行其内部代码，而是返回一个协程对象（Coroutine Object），最终协程对象需要被显式驱动才能真正运行。

以下是一个最小示例如下（完整代码见 C05_async.py）：

1 async def greet(name: str) -> str:
2     return f"hello, {name}"
3 
4 if __name__ == '__main__':
5     coro = greet("alice")
6     print(coro)  
7     print(type(coro))
8     print(asyncio.run(coro))
9     coro.close() 

运行上述代码会得到：

<coroutine object greet at 0x105d0d840>
<class 'coroutine'>
hello, alice

在上述代码中，第 6 行打印的是协程对象的字符串表示。第8行是通过 asyncio.run() 来真正执行greet 函数体，即对应的协程对象，也就是说只有当 asyncio.run(greet("alice")) 被调用时，greet 函数体才真正被执行，并最终返回字符串 "hello, alice"。

同样的关系也体现在第 6.4.2 节的 customer_async 上，单独调用 customer_async('naicha') 后不会立刻发起 HTTP 请求，它只是构造了一个尚未运行的协程对象；只有当它被 asyncio.gather 收集、并由 asyncio.run 执行后，函数体才开始真正运行。这一“调用 ≠ 执行”的特性，就是异步编程能够实现并发等待的基础。

进一步，async def 修饰的函数内部可以包含 await 表达式。await 的语义是把当前协程挂起，让事件循环有机会调度其它协程，直到被 await 的异步操作完成后再恢复执行。

例如在 customer_async 中：

response = await client.get(f"http://localhost:5000/{store}")

这行代码表示主程序发起一个 HTTP 请求，并在等待请求响应的过程中把当前协程挂起，以使得主程序可以推进其它正在等待 I/O 的协程，待 HTTP 请求响应到达后再恢复执行并把结果绑定到 response。理解这一行为的关键是把 await 看作“让出执行权”，而不是“阻塞等待”。需要补充的是，await 表达式只能出现在使用 async def 修饰的函数内部。如果在普通函数中写 await，Python 解释器会直接报 SyntaxError。换言之，async 是 await 的合法前提。

6.4.4 事件循环与 asyncio 运行机制#

协程函数与 await 表达式为异步编程提供了语法层面的入口，但真正把它们调度起来的是事件循环（Event Loop）。事件循环可以理解为异步程序运行时的调度中心：它会持续从任务队列中取出协程，并把协程推进到下一个 await、处理 await 完成后的回调，并把恢复的协程重新放回队列。

asyncio 是 Python 标准库提供的异步编程框架，其中最常用的入口函数是 asyncio.run。asyncio.run(coro) 会在内部创建一个新的事件循环，运行传入的协程，并在协程结束后自动关闭事件循环。

下面给出一个最简化、便于观察事件循环调度行为的示例（完整代码见 C06_async.py）：

 1 async def task(name: str, delay: float) -> None:
 2     print(f"{name} start")
 3     await asyncio.sleep(delay)
 4     print(f"{name} end")
 5 
 6 async def main():
 7     await asyncio.gather(
 8         task("A", 1),
 9         task("B", 2),
10         task("C", 3))
11 
12 if __name__ == '__main__':
13     asyncio.run(main())

在上述代码中，第1~4行里 task 是一个典型的协程函数，即先打印起始日志，再 await asyncio.sleep 一段时间，最后打印结束日志。第6~10行里main 通过 asyncio.gather 把 3 个协程并发地交给同一个事件循环。

上述代码运行结束后会观察到类似如下的执行结果：

A start 2026-09-16 21:33:04.305256
B start 2026-09-16 21:33:04.305282
C start 2026-09-16 21:33:04.305289
A end 2026-09-16 21:33:05.306658
B end 2026-09-16 21:33:06.306582
C end 2026-09-16 21:33:07.306432

从输出结果可以看到，3 个任务几乎同时启动，而结束顺序则由各自的等待时长决定，且最终时常由3个任务中用时最长的任务决定。这意味着当 A 在 await asyncio.sleep(1) 中挂起时事件循环并没有停滞，而是继续把 B 和 C 也推进到了自己的 await 阶段。等到 A 的等待时间结束，它才被恢复并打印结束日志。如果把 asyncio.sleep 替换成 time.sleep，则只会看到 A start、A end、B start、B end、C start、C end 这种严格串行的结果，整个程序的总耗时也会变成 6 秒。

time.sleep 会阻塞整个线程，而 asyncio.sleep 只会挂起当前协程，线程仍然可以执行其它协程

回到第 6.4.2 节中的 customer_async，当 asyncio.gather(customer_async('naicha'), customer_async('shouzhuabing'), customer_async('kaomianjing')) 被执行时，事件循环并不是依次等待 3 个 HTTP 请求完成，而是把 3 个协程几乎同时推进到 await client.get(...) 阶段。在等待的 8 秒、7 秒、5 秒中，3 个协程都处于挂起状态；当 kaomianjing 的 5 秒等待结束时，对应的协程最先被恢复并打印响应。事件循环就是通过这种“挂起-恢复”的循环，把原本串行的 20 秒总等待压缩为 8 秒。

需要说明的是，事件循环虽然提供了并发能力，但它本身仍然是单线程的。这意味着异步编程并不能直接加速 CPU 密集型任务，它真正的用武之地是 I/O 密集型任务——也就是 MCP 客户端调用、HTTP 请求、数据库查询等这类等待外部系统的场景。

6.4.5 异步任务的并发执行方式#

asyncio 中用于并发执行异步任务的最常用接口是 asyncio.gather 与 asyncio.create_task。它们都能把多个协程并行地交给事件循环，但适用场景略有不同。在第6.4.2 节中的 asyncio.gather 是「声明一组任务等它们一起完成」的典型写法，本节进一步展开这两类接口的使用方式。

1. asyncio.gather() 执行方式

asyncio.gather() 会把多个协程包装为一组并发任务，并返回一个可迭代的结果对象。当所有任务都完成后，gather 会把每个任务的返回值按顺序收集为一个列表。如果其中任何一个任务抛出了异常，gather 默认会把异常立刻抛出。典型用法如下（完整代码见C07_async_asyncio.py）：

 1 async def fetch_user(user_id: int) -> str:
 2     print(f"{user_id} start {datetime.now()}")
 3     await asyncio.sleep(2) # 模拟等待时间
 4     print(f"{user_id} end {datetime.now()}")
 5     return f"user-{user_id}"
 6 
 7 async def main():
 8     results = await asyncio.gather(
 9         fetch_user(1),
10         fetch_user(2),
11         fetch_user(3))
12     print(results)
13 
14 if __name__ == '__main__':
15     asyncio.run(main())

在上述代码中，3 次 fetch_user 并发调用被交给同一个事件循环，总耗时从同步版本下的约 6 秒缩短为约 2 秒。这种写法的优点是简洁，缺点就是在具体对某一个协程进行精细化管理时不那么方便。

上述代码输出结果为：

1 start 2026-09-16 21:57:48.517651
2 start 2026-09-16 21:57:48.517699
3 start 2026-09-16 21:57:48.517708
1 end 2026-09-16 21:57:50.519432
2 end 2026-09-16 21:57:50.519571
3 end 2026-09-16 21:57:50.519597
['user-1', 'user-2', 'user-3']

2. asyncio.create_task() 执行方式

进一步，asyncio.create_task() 提供了一种更细粒度的并发控制方式，它会把协程包装为一个 Task 对象并立即调度执行，返回的 Task 对象允许开发者在协程运行期间查询状态、取消执行或在稍后等待其结果。

下面，我们通过一个示例来进行介绍，示例代码如下（C08_async_asyncio.py）：

 1 async def slow_task(name: str, sleep: int = 0) -> str:
 2     print(f"task: {name} （协程）获得执行权限， start: {datetime.now()}")
 3     print(f"task: {name} （协程）即将失去执行权限，进入I/O 阻塞: {datetime.now()}")
 4     await asyncio.sleep(sleep) # 模拟代替之前的http异步请求
 5     print(f"task: {name} （协程）I/O阻塞结束后再次获得执行权限， start: {datetime.now()}")
 6     print(f"task: {name} （协程）结束 : {datetime.now()}")
 7     return f"{name} done"
 8 
 9 async def main():
10     task_a = asyncio.create_task(slow_task("A", 2))
11     task_b = asyncio.create_task(slow_task("B", 5))
12     print("主协程：所有任务调度完成！")
13     wait(3,task_a,task_b)
14     result_b = await task_b
15     print(task_a.result())
16     print(result_b)
17 
18 if __name__ == '__main__':
19     asyncio.run(main())

在上述代码中，第1~7行定义了一个协程函数，并打印出了关键信息。第9~16行定义了一个主协程函数，用于创建任务和模拟整个调度过程。其中，第13行代码会强制等待3秒并打印出两个任务的状态，预期情况是两个任务均未完成，因为此时执行权限还在主协程中。接着，第14行代码await task_b 则会阻塞当前主协程，直到 task_b 完成。需要注意的是，create_task 一旦被调用就会立即把任务提交到事件循环，当出现 await 时所有处于事件循环中的协程几乎同时进入运行状态。 进一步，由于在第10~11行代码中，我们设定了任务 A 的阻塞时间为2秒、B 的阻塞时间为 5 秒，因此当第14行代码执行结束以后，任务 A 也同时执行完成了。

对于被调度的协程来说，当任务完成时 asyncio 会更新它的状态并唤醒正在等待它的主程序（谁 await 了它就唤醒谁）。

对于上述代码的运行过程来说，我们还可以通过如图6-6所示的过程来进行表示。

上述代码运行结束后的结果为：

主协程：所有任务调度完成！
主协程：此阶段主协程拥有执行权限，等待3秒
task a 状态: done = False, 2026-09-16 22:11:44.128329
task b 状态: done = False, 2026-09-16 22:11:44.128440
------------------------------
task a 状态: done = False, 2026-09-16 22:11:45.130611
task b 状态: done = False, 2026-09-16 22:11:45.130719
------------------------------
task a 状态: done = False, 2026-09-16 22:11:46.133827
task b 状态: done = False, 2026-09-16 22:11:46.133896
------------------------------
task: A （协程）获得执行权限， start: 2026-09-16 22:11:46.134049
task: A （协程）即将失去执行权限，进入I/O 阻塞: 2026-09-16 22:11:46.134066
task: B （协程）获得执行权限， start: 2026-09-16 22:11:46.134169
task: B （协程）即将失去执行权限，进入I/O 阻塞: 2026-09-16 22:11:46.134176
task: A （协程）I/O阻塞结束后再次获得执行权限， start: 2026-09-16 22:11:48.135457
task: A （协程）结束 : 2026-09-16 22:11:48.135520
task: B （协程）I/O阻塞结束后再次获得执行权限， start: 2026-09-16 22:11:51.135442
task: B （协程）结束 : 2026-09-16 22:11:51.135492
A done
B done

进一步，大家可以对照代码及其输出结果，并结合图6-6所示的状态分析整个执行过程。

除了上面用到的 task_b.done() 方法以外，asyncio.Task 还提供了几个常用方法，如下所示：

至此我们就介绍完了两种异步任务的并发执行方式。从 MCP 服务的实现角度看，asyncio.gather 与 asyncio.create_task 是两个最常用的并发工具，当多个 MCP 工具调用彼此独立、希望并发执行时，可以使用 asyncio.gather；当需要在任务运行过程中保留取消、状态查询等更细的控制时，可以使用 asyncio.create_task。

等待方不一定是主协程，也可以是其他协程。等待方可以通过 await task 获取单个任务的结果，也可以通过 await asyncio.gather() 同时等待多个任务全部完成并获取结果，也可以通过 asyncio.wait() 等方式等待指定的完成条件。整个过程由事件循环负责调度，不需要主协程主动轮询任务状态。

6.4.6 异步上下文管理器与迭代器#

除了异步函数和并发任务之外，Python 还提供了两类与异步语义配套的对象：异步上下文管理器与异步迭代器。它们分别对应着同步编程中 with 语句与 for 循环的异步版本。

异步上下文管理器通过实现 __aenter__ 与 __aexit__ 两个魔术方法对外提供 async with 支持，如果熟悉同步编程中 with 的实现方法，那么对于 async with 的封装过程就会一目了然了。

对于 async with 来说，典型用法是管理那些建立与销毁都需要异步等待的资源，例如 HTTP 连接、数据库连接池等。在 MCP 服务实现中，自定义 Server 经常会借助异步上下文管理器来管理与 Client 之间的连接或会话。示例代码如下：

 1 import asyncio
 2 
 3 class AsyncConnection:
 4     async def __aenter__(self):
 5         await asyncio.sleep(0.1)
 6         print("connection opened")
 7         return self
 8 
 9     async def __aexit__(self, exc_type, exc, tb):
10         await asyncio.sleep(0.1)
11         print("connection closed")
12 
13 async def main():
14     async with AsyncConnection() as conn:
15         print("using", conn)
16 
17 if __name__ == '__main__':
18     asyncio.run(main())

在上述代码中，__aenter__ 与 __aexit__ 都是协程函数，因此可以在其中使用 await。第14行 async with 语句会在进入时执行 await __aenter__，在退出时执行 await __aexit__，从而保证资源建立与销毁的全过程都符合异步语义。

进一步，异步迭代器则对应 async for 循环。它需要实现 __aiter__ 与 __anext__ 两个魔术方法，并且 __anext__ 必须是一个协程函数。下面给出一个简单示例：

 1 class AsyncCounter:
 2     def __init__(self, limit: int):
 3         self.limit = limit
 4         self.current = 0
 5 
 6     def __aiter__(self):
 7         return self
 8 
 9     async def __anext__(self):
10         if self.current >= self.limit:
11             raise StopAsyncIteration
12         await asyncio.sleep(0.5)
13         self.current += 1
14         return self.current
15 
16 async def main():
17     async for value in AsyncCounter(3):
18         print(value)
19 
20 if __name__ == '__main__':
21     asyncio.run(main())

在上述代码中，AsyncCounter 模拟一个异步计数器，每次 __anext__ 都会先 await asyncio.sleep(0.5) 再返回下一个整数。async for 循环的写法与普通 for 循环几乎一致，但每一次迭代都隐含着异步等待。

可以看到，异步上下文管理器与异步迭代器本质上都是把异步等待封装进了一个对外使用方式与同步版本相似的接口。这种设计让开发者既能保持与同步 API 相近的写法，又能享受到异步执行带来的并发能力。

在 MCP 协议中，远端 Server 返回的流式数据通常就以异步迭代器的形式被消费，因此理解 async for 的工作机制将有助于理解 MCP 流式工具的实现。

对于这部分内容，各位读者在此作为了解即可，等后续实际运用到了这部分知识再回过头来阅读会有更清晰地认识。

6.4.7 小结#

本节围绕 Python 异步编程进行了系统性介绍。我们从同步编程在 I/O 场景下的局限出发，给出了同一买奶茶场景在异步版本下的实现，把总耗时从 20 秒压缩为 8 秒；随后说明了 async def 与 await 的基本语义，并介绍了事件循环在 asyncio 中的运行机制。在此基础上，演示了 asyncio.gather 与 asyncio.create_task 这两类常见的并发执行模式，最后补充了异步上下文管理器与异步迭代器的使用场景。

可以看到，异步编程并不是用来加速 CPU 计算的工具，而是用来高效等待 I/O 的语言机制。它的核心思想是让出当前协程的执行权，从而在单线程中实现对多个 I/O 任务的并发处理。这一思想与 MCP 协议所强调的“跨进程、跨服务工具交互”天然契合，因此它构成了下一节实现 MCP 服务的重要前置知识。

引用#

[1] Python Software Foundation. asyncio — Asynchronous I/O. https://docs.python.org/3/library/asyncio.html

[2] PEP 492. Coroutines with async and await syntax. https://peps.python.org/pep-0492/

[3] PEP 3156. Asynchronous I/O Support Rebooted: the “asyncio” Module. https://peps.python.org/pep-3156/