asynccontextmanager
学习MCP开发，经常看到@asynccontextmanager。
@asynccontextmanager 是 Python 3.7+ 标准库 contextlib 提供的一个 装饰器，用来把 异步生成器函数 一键变成 异步上下文管理器（async context manager）。
它让你用 async with 管理异步资源的获取/释放时，不必写一整块样板类，只需几行 yield 代码即可。
原理
异步上下文管理器协议 =
__aenter__/__aexit__两个协程方法。@asynccontextmanager帮你自动生成这两个方法；
你的函数只需负责：
① 在yield前完成异步初始化（获取资源）；
② 在yield后完成异步清理（释放资源）。yield值会成为async with ... as xxx里的xxx；
如果yield后面还有代码，则无论with块是否抛异常，都会执行（等价于__aexit__）。
举例
import asyncio
from contextlib import asynccontextmanager@asynccontextmanager
async def open_connection(host, port):"""异步建立 TCP 连接，退出时自动关闭。"""reader, writer = await asyncio.open_connection(host, port)print("🔗 连接已建立")try:yield reader, writer # 把资源交给 with 块finally:writer.close()await writer.wait_closed()print("🔌 连接已关闭")async def main():async with open_connection('httpbin.org', 80) as (rd, wr):wr.write(b'GET /get HTTP/1.0\r\nHost: httpbin.org\r\n\r\n')await wr.drain()data = await rd.read(1024)print('收到:', data[:60])if __name__ == '__main__':asyncio.run(main())高级技巧
| 场景 | 正确姿势 |
| 异常处理 | yield 放在 try/finally 里，保证释放逻辑必跑；如果想区分异常类型，在 except 里重新 raise。 |
| 传参给清理逻辑 | 把需要释放的字段放在 yield 之前的作用域即可，如上面的 writer。 |
| 可重入？ | 每次 async with 都会重新执行一次生成器，天然线程/协程隔离。 |
与同步 @contextmanager 混用 | 不要混用！一个函数只能被一种装饰器装饰。 |
需要 asyncio 调度器？ | 不需要，只要运行环境是 asyncio.run 或 loop 即可。 |
手写版本
async with 会自动调用
__aenter__ __aexit__
class OpenConnection:def __init__(self, host, port):self.host, self.port = host, portasync def __aenter__(self):self.r, self.w = await asyncio.open_connection(self.host, self.port)return self.r, self.wasync def __aexit__(self, exc_t, exc_v, tb):self.w.close()await self.w.wait_closed()不要把同一个 asynccontextmanager 实例在不同协程里并发
async def maker(): # 被 @asynccontextmanager 装饰的「函数」 ... yield ...cm_func = maker # 这是「工厂函数」，每次调用返回新异步生成器
cm_obj = cm_func() # 这是「异步生成器对象」，内部保存 ag_running / ag_frame 状态
acm = _AsyncGeneratorContextManager(cm_obj) # 被装饰后得到的「上下文管理器实例」关键：
acm内部只持有同一个cm_obj如果两个协程 并发
await acm.__aenter__()，就会 同时驱动同一个异步生成器，
导致ag_running标志冲突、帧栈错位，最终抛RuntimeError: anext(): asynchronous generator is already running。
翻车示例
import asyncio
from contextlib import asynccontextmanager@asynccontextmanager
async def db():print("[db] connect")yield "connection"print("[db] close")acm = db() # ⚠️ 只创建了一次上下文管理器实例！async def task(n):async with acm as conn: # 两个协程并发进入同一个 acmprint(f"[task{n}] got {conn}")await asyncio.sleep(0.1)async def main():await asyncio.gather(task(1), task(2))if __name__ == '__main__':asyncio.run(main())
[db] connect
[db] close
[task1] got connection
RuntimeError: anext(): asynchronous generator is already running第二条协程试图再
anext时，生成器还在运行，直接炸。更严重：第一条协程的
__aexit__可能永远不会被调用，资源泄漏。