aljaz/mcp-python-sdk
1
0
Fork 0
mirror of https://github.com/aljazceru/mcp-python-sdk.git synced 2025-12-19 23:04:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

Drop AbstractAsyncContextManager for proper type hints (#257)

Browse Source
This commit is contained in:
Marcelo Trylesinski
2025-03-11 14:17:15 +01:00
committed by GitHub
parent 5cbea24ecb
commit 3e0ab1e7ee
1 changed files with 3 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/mcp/shared/session.py
View File

@@ -8,6 +8,7 @@ import anyio.lowlevel
import httpx import httpx
from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
from pydantic import BaseModel from pydantic import BaseModel
from typing_extensions import Self
from mcp.shared.exceptions import McpError from mcp.shared.exceptions import McpError
from mcp.types import ( from mcp.types import (
@@ -60,7 +61,7 @@ class RequestResponder(Generic[ReceiveRequestT, SendResultT]):
request_id: RequestId, request_id: RequestId,
request_meta: RequestParams.Meta | None, request_meta: RequestParams.Meta | None,
request: ReceiveRequestT, request: ReceiveRequestT,
session: "BaseSession", session: "BaseSession[SendRequestT, SendNotificationT, SendResultT, ReceiveRequestT, ReceiveNotificationT]",
on_complete: Callable[["RequestResponder[ReceiveRequestT, SendResultT]"], Any], on_complete: Callable[["RequestResponder[ReceiveRequestT, SendResultT]"], Any],
) -> None: ) -> None:
self.request_id = request_id self.request_id = request_id
@@ -134,7 +135,6 @@ class RequestResponder(Generic[ReceiveRequestT, SendResultT]):
class BaseSession( class BaseSession(
AbstractAsyncContextManager,
Generic[ Generic[
SendRequestT, SendRequestT,
SendNotificationT, SendNotificationT,
@@ -183,7 +183,7 @@ class BaseSession(
]() ]()
) )
async def __aenter__(self): async def __aenter__(self) -> Self:
self._task_group = anyio.create_task_group() self._task_group = anyio.create_task_group()
await self._task_group.__aenter__() await self._task_group.__aenter__()
self._task_group.start_soon(self._receive_loop) self._task_group.start_soon(self._receive_loop)