Function
lifespan_wrapper
(
app: MCPServer[LifespanResultT],
lifespan: Callable[[MCPServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]],
)
Source from the content-addressed store, hash-verified
| 116 | |
| 117 | |
| 118 | def lifespan_wrapper( |
| 119 | app: MCPServer[LifespanResultT], |
| 120 | lifespan: Callable[[MCPServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]], |
| 121 | ) -> Callable[[Server[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]]: |
| 122 | @asynccontextmanager |
| 123 | async def wrap(_: Server[LifespanResultT]) -> AsyncIterator[LifespanResultT]: |
| 124 | async with lifespan(app) as context: |
| 125 | yield context |
| 126 | |
| 127 | return wrap |
| 128 | |
| 129 | |
| 130 | class MCPServer(Generic[LifespanResultT]): |
Tested by
no test coverage detected